Repository navigation
Limit the topic browse query on the JDBC store (#2630) - #2634
papinifrancesco wants to merge 1 commit into
Conversation
| if (topicStore != null) { | ||
| // a JDBC ack moves the last acked id of the subscription, so expiring a browsed | ||
| // message would also ack the non-expired messages before it | ||
| final boolean expireFromStore = topicStore.getType() == StoreType.JDBC; |
There was a problem hiding this comment.
This breaks modularity. There shouldn't be a check for an underlying store type and a subsequent change in logic. If there is a problem with the JDBC ack'n with browsed messages, we should fix that in the JDBC layer.
|
This is a significant change, and I'm wondering if there is a better way to update this (perhaps add new first-class methods to the store) vs branching logic based on the back-end store type. Is there a way to address the first part-- add a limit query based on maxBrowsePageSize without dissecting the entire Topic expiry code flow? That would allow us to address the core issue quickly, and then spend some time on the design to optimize expiry processing as a follow-on. |
Topic.doBrowse(), also used by the topic expiry task for stores without recoverExpired(), recovered the topic through MessageStore.recover(). The JDBC store ran the query with no limit and relied on the listener to stop reading, but pgjdbc reads the whole result set in executeQuery(), so a large durable subscription backlog was loaded into the heap on every expiry run. Add MessageStore.recover(listener, maxReturned), whose default calls recover(listener), so stores that do not override it are unchanged. The JDBC store puts the limit on the query (setMaxRows). Topic.doBrowse() passes its page size. What a browse returns or expires does not change. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…rowse-limit.patch) Upstream asked for a smaller first fix (apache/activemq#2634): only bound the browse query by the page size. build-patched.sh takes FIX to pick the patch. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
07cc7f7 to
d579453
Compare
papinifrancesco
changed the title
|
Thanks @mattrpav, agreed. This PR is now only that fix, one commit on current
On PostgreSQL, with this patch 6.3.2 and 6.2.10 survive the 200,000-message backlog from #2630 with a 1 GB heap, where unpatched they run out of memory (CI run). New H2 test: One question: you added The rest is now #2640: on the JDBC store, expiry acks can lose pending messages or redeliver consumed ones, with tests for both. The previous version of this PR is linked there as one possible fix. @cshannon: with this version, KahaDB's code path is unchanged. Disclosure: as before, the change, the tests and the reproduction were written with an AI assistant (Claude Code, Anthropic). The test results come from local runs and the linked CI run. |
2 participants
Link to #2630
Problem. For stores without
recoverExpired()(JDBC), the topic expiry task falls back toTopic.doBrowse(), which callsMessageStore.recover(). The JDBC store runsSELECT ID, MSG FROM ACTIVEMQ_MSGS WHERE CONTAINER=? ORDER BY IDwith no limit. Since AMQ-6067 the listener stops reading at the page size, but pgjdbc reads the whole result set insideexecuteQuery(), so with a large durable subscription backlog every expiry run loads the whole topic into the heap and the broker runs out of memory.Change.
MessageStore.recover(MessageRecoveryListener, int maxReturned): new default method that callsrecover(listener), so stores that don't override it are unchanged.ProxyMessageStoreandProxyTopicMessageStoredelegate it.JDBCMessageStoreoverrides it;DefaultJDBCAdapter.doRecover()gets an overload that setssetMaxRows(maxReturned)on the query (0 = no limit, as before).Topic.doBrowse()passes itsmax.Browse and expiry results are unchanged; only the number of rows the driver reads is bounded.
Testing.
JDBCTopicBrowseLimitTest(H2).JDBCPersistenceAdapterExpiredMessageTest: its proxy now overrides the newrecover(listener, maxReturned), which is whatTopiccalls; it still checks thathasSpace()is called during expiry.JDBCStoreBrokerTest,H2JDBCStoreBrokerTest,JDBCDurableSubscriptionTest,JDBCMessagePriorityTestandJDBCPersistenceAdapterExpiredMessageTestpass.Disclosure: written with an AI assistant (Claude Code, Anthropic); the commit carries a
Co-authored-bytrailer naming the tool.🤖 Generated with Claude Code