allthingssecurity opened a new pull request, #27389: URL: https://github.com/apache/camel/pull/27389
# Description [CAMEL-25313](https://issues.apache.org/jira/browse/CAMEL-25313) Suspending the route of a Kafka consumer only sets the requested state of each `KafkaFetchRecords` task (`pause()`/`resume()` run on the caller thread: route controller, JMX, the timer of `ThrottlingExceptionRoutePolicy`); the fetcher thread applies it in `updateTaskState()` after the next poll. Two defects make a resumed consumer stop consuming for good: - `run()` returns at once, and its do-while loop ends, when the consumer is *suspended* (`isKafkaConsumerRunnable()`), while the polling loop inside `startPolling` deliberately keeps running when suspended (CAMEL-18327). A thread that starts while the consumer is suspended (`ThrottlingExceptionRoutePolicy` with `keepOpen=true` suspends the consumer in `onStart`, before the pending consumer tasks are submitted; this is the "couldn't be resumed after we changed the keepOpen to false" of CAMEL-19358), or that leaves the polling loop while suspended (a failed poll with the default `pollOnError=ERROR_HANDLER`, a reconnect), terminates. `doResume` then sets the state of a task that no thread runs. - `updateTaskState` sets `PAUSED`/`RUNNING` unconditionally after `consumer.pause()`/`resume()`. A resume requested in between is overwritten (the consumer stays paused while the route is started); a suspend requested during a resume is overwritten by `RUNNING` (the consumer keeps consuming while suspended). With `keepOpen=true` this is not a race: when the route starts with the CamelContext, the fetcher tasks are only submitted once the context has started (`pendingConsumer`), after the policy's `onStart` suspended the consumer. This change moves to `PAUSED`/`RUNNING` with `compareAndSet` (a newer request is handled on the next iteration), lets the thread end only when the consumer is stopping or stopped (`isKafkaConsumerRunnableAndNotStopped()`, as the polling loop), and turns `PAUSED` into `PAUSE_REQUESTED` after a (re)connect, as the new Kafka consumer is not paused. Since a thread now also starts or reconnects while the consumer is suspended, the rebalance listener pauses the assigned partitions when a pause is requested or applied (`onPartitionsAssigned` runs on the fetcher thread inside `poll`), so that the poll which assigns them does not return their records: otherwise `keepOpen=true` would consume the first batch with the circuit open. Records of a batch that was already polled when the suspend arrives are still processed first (unchanged: "after current processing"). The defect was found with a TLA+ model of the fetcher thread against suspend/resume (`state.get()`, the consumer call and `state.set()` as separate steps, the run-loop conditions, a poll that can fail or reconnect): "once the operations are done and no request is pending, a started route has a polling thread with the consumer not paused, and a suspended route a paused consumer" is violated in 4 steps (suspend, the thread starts and returns, resume), and in 7 steps for the overwritten resume. With the fix it holds, as does the liveness property, with two suspend/resume cycles, failing polls and reconnects. Tests: new `KafkaConsumerSuspendResumeTest` (no broker: a `KafkaClientFactory` returning a `MockConsumer`, whose `pause`/`resume` can be held on a latch to force the interleaving; latches, Awaitility, `MockEndpoint`). Without the main-code change: ``` KafkaConsumerSuspendResumeTest.testPollErrorWhileSuspended:148 mock://result Received message count. Expected: <1> but was: <0> KafkaConsumerSuspendResumeTest.testResumeWhileThePauseIsApplied:100 mock://result Received message count. Expected: <1> but was: <0> KafkaConsumerSuspendResumeTest.testStartedWithOpenCircuit:163 mock://result Received message count. Expected: <1> but was: <0> KafkaConsumerSuspendResumeTest.testStartedWithOpenCircuitConsumesNothing:173 » ConditionTimeout (the fetcher thread is gone, the consumer is never paused) KafkaConsumerSuspendResumeTest.testSuspendWhileTheResumeIsApplied:119 » ConditionTimeout ... The Kafka consumer must be paused while the route is suspended ==> expected: <true> but was: <false> within 20 seconds. ``` `testStartedWithOpenCircuitConsumesNothing` (a record on the topic when the route starts with `keepOpen=true`) also fails without the rebalance-listener part: `Expected: <0> but was: <1>`. The TLA+ model does not cover that part (no rebalance, records abstracted). With the change the camel-kafka unit tests pass (224, 1 skipped). The integration tests need Docker and were not run. # 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). # 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 and tested the affected modules, including the formatter and import-sort plugins. 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]
