papinifrancesco commented on PR #2634: URL: https://github.com/apache/activemq/pull/2634#issuecomment-6615109584
Thanks @mattrpav. I tried it: `Topic.doBrowse()` calls `topicStore.resetBatching()` and then `topicStore.recoverNextMessages(max, listener)`, under a lock for the `doBrowse()` callers. On JDBC it works. `doRecoverNextMessages()` already limits the query (`setMaxRows`, plus `LIMIT` on PostgreSQL), and nothing else uses the topic store's batch position. On KahaDB it changes the browse. `recoverNextMessages(int, listener)` iterates the destination's default order-index cursor, and durable subscription recovery moves that same cursor ([`KahaDBStore` L1260-L1314](https://github.com/apache/activemq/blob/730b34d3af5574406e3924e72f4bd57ceda6eac0/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java#L1260-L1314): `resetCursorPosition()`, `setBatch()`, `stoppedIterating()`). The two calls aren't atomic, so a dispatch to a durable subscriber between `resetBatching()` and `recoverNextMessages()` makes the browse start in the middle of the topic. Today's `recover(listener)` uses `iteratorIsolated()` and isn't affected. A test that recovers 20 messages for a durable subscription right before the browse reads (50 messages, page size 10): | | JDBC (H2) | KahaDB | |---|---|---| | `resetBatching()` + `recoverNextMessages()` | starts at message0 | starts at **message20** | | this PR, `recover(listener, maxReturned)` | starts at message0 | starts at message0 | It would also change an AMQ-6067 assertion: `JDBCPersistenceAdapterExpiredMessageTest` checks that `hasSpace()` is called during expiry, and JDBC's `recoverNextMessages()` doesn't call it (the row count is the limit instead). So I'd keep a stateless bounded read. Two ways: 1. This PR as it is: `recover(listener, maxReturned)`, a default method that calls `recover(listener)`, overridden by the JDBC store only. 2. Your `recoverMessages(MessageRecoveryContext)` with the isolated cursor, which KahaDB already implements. `Topic.doBrowse()` would use it, and I'd add a JDBC implementation (head of the store, at most N rows). The memory store would need one too, or a default that falls back to `recover(listener)`. Which do you prefer? Either way I can add the test above to the PR. Disclosure: as with the rest of this PR, the trial, the test and the runs above were done with Claude Code (Anthropic's AI assistant). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected] For further information, visit: https://activemq.apache.org/contact
