allthingssecurity opened a new pull request, #29408: URL: https://github.com/apache/camel/pull/29408
# Description [CAMEL-26709](https://issues.apache.org/jira/browse/CAMEL-26709) A route stop/start (JMX, the route controller, a route policy) stops the resume strategy of the route and starts it again (`ResumableProcessor`, and `Splitter` with a resume strategy). `SingleNodeKafkaResumeStrategy` did not support this: - `stop()` closed the producer but kept it, and `start()` only creates a producer when there is none, so every later offset update failed with "Cannot perform operation after producer has been closed". The resumable completion only logs that, so the route keeps running, the offsets are lost, and the records are processed again after the next application restart. In a split with a resume strategy the exchange hung (fixed separately in CAMEL-26710). - `stop()` shut down the executor but kept it, and `loadCache()` submitted the refresh task to it again, so the offsets were not loaded any more. - `subscribed` stayed true, so the consumer of the next start did not subscribe to the topic. Also, one record of the resume topic that cannot be deserialized, for example one without a key or value (a tombstone on a compacted topic), made `poll()` throw (`ByteBuffer.wrap(null)`); `refresh()` logged it once and the refresh thread ended, so no later offset was loaded. CAMEL-23996 (#24692) fixed the unlock in `stop()`, not the restart. Change (`SingleNodeKafkaResumeStrategy`): - `stop()` clears the producer after closing it, and shuts the executor down through the `ExecutorServiceManager` (same 2 second wait, and the pool is deregistered, also from JMX) and clears it. `start()`/`loadCache()` then create new ones. - `subscribed` is reset for each new consumer in `refresh()`. - `produce()` throws an `IllegalStateException` when the strategy is stopped, instead of a `NullPointerException` (on main it was the `IllegalStateException` of the closed producer). - `poll()` skips a record that cannot be deserialized, with a WARN that names the offset, topic and partition. Not changed: as before, `stop()` also shuts down an executor passed to the constructor; after a restart the strategy then uses a new executor created by Camel. Not shutting down an executor the strategy did not create could be a follow-up. Tests (`SingleNodeKafkaResumeStrategyTest`, no broker: `localhost:9`, where nothing listens, is enough to create, subscribe and close the clients): - `testStrategyCanBeStartedAgainAfterStop`: start, load, stop, start, load; the producer is a new one, the executor is not shut down, and the second consumer subscribed to the topic (recorded in `checkAndSubscribe`). - `testRecordThatCannotBeDeserializedIsSkipped`: `consume()` returns a record without a key and then a valid one; the valid one is still deserialized. On main both tests fail (the closed producer and the shut-down executor are used again; `NullPointerException` from `poll()`); for the negative control only a `getExecutorService()` accessor was added to the main class. Mutant without the `subscribed` reset: the restart test fails (the second consumer has no subscription). With the fix 5/5, 20 runs 100/100 (about 1.5 s per run). camel-kafka unit tests: 308 run, 0 failures, 1 skipped (offline, `-DskipITs`, with the formatter and import-sort plugins). # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [x] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it): [CAMEL-26709](https://issues.apache.org/jira/browse/CAMEL-26709). # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes. (I built the core modules and camel-kafka with the formatter and import-sort plugins and ran the camel-kafka unit tests; no generated file changes. I did not run the full root build.) # AI-assisted contributions - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used. This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a `Co-Authored-By` trailer. _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
