You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "allthingssecurity (via GitHub)" <gi...@apache.org> on 2026/10/06 00:03:47 UTC

[PR] CAMEL-25313: camel-kafka - a suspended consumer must keep its fetcher thread, and a pause or resume request must not be lost [camel]

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: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-25313: camel-kafka - a suspended consumer must keep its fetcher thread, and a pause or resume request must not be lost [camel]

Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27389:
URL: https://github.com/apache/camel/pull/27389#issuecomment-6010013744

   :star2: Thank you for your contribution to the Apache Camel project! :star2:
     :robot: CI automation will test this PR automatically.
   
     :camel: Apache Camel Committers, please review the following items:
   
     * First-time contributors **require MANUAL approval** for the GitHub Actions to run
     * You can use the command `/component-test (camel-)component-name1 (camel-)component-name2..` to request a test from the test bot although they are normally detected and executed by CI.
     * You can label PRs using `skip-tests` and `test-dependents` to fine-tune the checks executed by this PR.
     * Build and test logs are available in the summary page. **Only** [Apache Camel committers](https://camel.apache.org/community/team/#committers) have access to the summary.
     
     :warning: Be careful when sharing logs. Review their contents before sharing them publicly.


-- 
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: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-25313: camel-kafka - a suspended consumer must keep its fetcher thread, and a pause or resume request must not be lost [camel]

Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus merged PR #27389:
URL: https://github.com/apache/camel/pull/27389


-- 
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: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-25313: camel-kafka - a suspended consumer must keep its fetcher thread, and a pause or resume request must not be lost [camel]

Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #27389:
URL: https://github.com/apache/camel/pull/27389#discussion_r4192761823


##########
docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc:
##########
@@ -3934,6 +3934,15 @@ the SSL endpoint options were ignored when `sslContextParameters` was configured
 When `sslContextParameters` is configured and `securityProtocol` is the default (`PLAINTEXT`), the security protocol
 `SSL` is now used. Previously `PLAINTEXT` was used unless `securityProtocol` (or `saslAuthType`) was configured.
 
+==== A suspended consumer keeps its partitions

Review Comment:
   Nit: this sits under `=== camel-kafka - sslContextParameters and the SSL endpoint options`, a heading about SSL only. A separate `=== camel-kafka - a suspended consumer keeps its partitions` would read better.



-- 
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: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-25313: camel-kafka - a suspended consumer must keep its fetcher thread, and a pause or resume request must not be lost [camel]

Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on PR #27389:
URL: https://github.com/apache/camel/pull/27389#issuecomment-6010338991

   Thanks for the review. Addressed in 7ce41e783cd9 (the branch was rebased on main first, to put the note under the existing camel-kafka heading, so the first commit is now 98a7103f5e62):
   
   - Upgrade guide: `==== A suspended consumer keeps its partitions` under the camel-kafka section. A route started with its consumer suspended (`keepOpen=true`) now joins the group and holds its partitions paused, where it never connected before.
   - `isRecoverable()`: it was a one-line change consistent with this PR, so I made it. It now uses `isKafkaConsumerRunnableAndNotStopped()`, as the fetcher thread of a suspended consumer keeps polling and reconnecting, and the unused `isKafkaConsumerRunnable()` is removed. New `testSuspendedConsumerIsRecoverable` fails without it (`expected: <true> but was: <false>`) and passes with it. It only changes the health-check message of a not-ready suspended consumer ("recovery in progress" instead of nothing).
   
   camel-kafka: 225 unit tests pass (1 skipped); the integration tests need Docker and were not run.
   
   _Claude Code on behalf of allthingssecurity_
   


-- 
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: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-25313: camel-kafka - a suspended consumer must keep its fetcher thread, and a pause or resume request must not be lost [camel]

Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27389:
URL: https://github.com/apache/camel/pull/27389#issuecomment-6010309760

   <!-- ci-tested-modules -->
   :test_tube: **CI tested the following changed modules:**
   
   - `components/camel-kafka`
   
   ---
   
   <details><summary>:microscope: Scalpel shadow comparison โ€” Scalpel: 9 of 698 tested, 27 compile-only โ€” current: 9 all tested</summary>
   
   [Maveniverse Scalpel](https://github.com/maveniverse/scalpel) detected **9 affected modules** (current approach: 9).
   
   **Skip-tests mode would test 9 modules** (1 direct + 8 downstream), **skip tests for 27** (generated code, meta-modules)
   
   <details><summary>Modules Scalpel would test (9)</summary>
   
   - `camel-jbang-mcp` โ† downstream of org.apache.camel:camel-catalog
   - `camel-jbang-plugin-mcp` โ† downstream of org.apache.camel:camel-jbang-core
   - `camel-jbang-plugin-route-parser` โ† downstream of org.apache.camel:camel-route-parser
   - `camel-jbang-plugin-tui` โ† downstream of org.apache.camel:camel-catalog
   - `camel-jbang-plugin-validate` โ† downstream of org.apache.camel:camel-yaml-dsl-validator
   - `camel-kafka` โ† components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java, components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerSuspendResumeTest.java
   - `camel-launcher-container` โ† downstream of org.apache.camel:camel-launcher
   - `camel-yaml-dsl-validator` โ† downstream of org.apache.camel:camel-catalog
   - `camel-yaml-dsl-validator-maven-plugin` โ† downstream of org.apache.camel:camel-yaml-dsl-validator
   
   </details>
   
   <details><summary>Modules with tests skipped (27)</summary>
   
   - `apache-camel`
   - `camel-allcomponents`
   - `camel-catalog`
   - `camel-catalog-console`
   - `camel-catalog-maven`
   - `camel-catalog-suggest`
   - `camel-componentdsl`
   - `camel-endpointdsl`
   - `camel-endpointdsl-support`
   - `camel-itest`
   - `camel-jbang-core`
   - `camel-jbang-it`
   - `camel-jbang-main`
   - `camel-jbang-plugin-edit`
   - `camel-jbang-plugin-generate`
   - `camel-jbang-plugin-kubernetes`
   - `camel-jbang-plugin-test`
   - `camel-kamelet-main`
   - `camel-launcher`
   - `camel-report-maven-plugin`
   - `camel-route-parser`
   - `camel-yaml-dsl`
   - `camel-yaml-dsl-deserializers`
   - `camel-yaml-dsl-maven-plugin`
   - `coverage`
   - `docs`
   - `dummy-component`
   
   </details>
   
   > :information_source: Shadow mode โ€” Scalpel observes but does not affect test execution. [Learn more](https://github.com/maveniverse/scalpel)
   
   </details>
   
   :warning: **Some tests are disabled on GitHub Actions** (`@DisabledIfSystemProperty(named = "ci.env.name")`) and require manual verification:
   
   - `components/camel-kafka`: 2 test(s) disabled on GitHub Actions
   
   <details><summary>All tested modules (36 modules, 4m 3s total)</summary>
   
   **Total reactor time:** 4m 3s
   
   | Module | Duration | Status |
   | --- | --- | --- |
   | Camel :: Launcher | 34.8s | SUCCESS |
   | Camel :: JBang :: Plugin :: TUI | 33.9s | SUCCESS |
   | Camel :: JBang :: MCP | 27.8s | SUCCESS |
   | Camel :: YAML DSL | 15.9s | SUCCESS |
   | Camel :: Catalog :: Camel Catalog | 15.8s | SUCCESS |
   | Camel :: Component DSL | 15.5s | SUCCESS |
   | Camel :: YAML DSL :: Validator | 14.1s | SUCCESS |
   | Camel :: JBang :: Plugin :: Kubernetes | 13.4s | SUCCESS |
   | Camel :: Docs | 10.4s | SUCCESS |
   | Camel :: Kamelet Main | 8.0s | SUCCESS |
   | Camel :: YAML DSL :: Deserializers | 7.1s | SUCCESS |
   | Camel :: Catalog :: Camel Route Parser | 6.4s | SUCCESS |
   | Camel :: JBang :: Plugin :: Testing | 6.1s | SUCCESS |
   | Camel :: Catalog :: Camel Report Maven Plugin | 5.2s | SUCCESS |
   | Camel :: JBang :: Plugin :: Validate | 4.5s | SUCCESS |
   | Camel :: All Components Sync point | 3.9s | SUCCESS |
   | Camel :: YAML DSL :: Validator Maven Plugin | 3.9s | SUCCESS |
   | Camel :: YAML DSL :: Maven Plugins | 2.5s | SUCCESS |
   | Camel :: Catalog :: Suggest (deprecated) | 2.1s | SUCCESS |
   | Camel :: Catalog :: Maven | 2.0s | SUCCESS |
   | Camel :: Catalog :: Dummy Component | 1.5s | SUCCESS |
   | Camel :: JBang :: Plugin :: Edit | 1.0s | SUCCESS |
   | Camel :: Catalog :: Console | 0.9s | SUCCESS |
   | Camel :: Coverage | 0.8s | SUCCESS |
   | Camel :: Assembly | 0.8s | SUCCESS |
   | Camel :: JBang :: Plugin :: MCP | 0.8s | SUCCESS |
   | Camel :: JBang :: Plugin :: Generate | 0.8s | SUCCESS |
   | Camel :: Endpoint DSL :: Support | 0.8s | SUCCESS |
   | Camel :: JBang :: Integration tests | 0.7s | SUCCESS |
   | Camel :: Launcher :: Container | 0.6s | SUCCESS |
   | Camel :: JBang :: Main | 0.6s | SUCCESS |
   | Camel :: JBang :: Plugin :: Route Parser | 0.5s | SUCCESS |
   | Camel :: Endpoint DSL | n/a |  |
   | Camel :: Integration Tests | n/a |  |
   | Camel :: JBang :: Core | n/a |  |
   | Camel :: Kafka | n/a |  |
   
   **Top 20 slowest modules:**
   - `Camel :: Launcher` (34.8s)
   - `Camel :: JBang :: Plugin :: TUI` (33.9s)
   - `Camel :: JBang :: MCP` (27.8s)
   - `Camel :: YAML DSL` (15.9s)
   - `Camel :: Catalog :: Camel Catalog` (15.8s)
   - `Camel :: Component DSL` (15.5s)
   - `Camel :: YAML DSL :: Validator` (14.1s)
   - `Camel :: JBang :: Plugin :: Kubernetes` (13.4s)
   - `Camel :: Docs` (10.4s)
   - `Camel :: Kamelet Main` (8.0s)
   - `Camel :: YAML DSL :: Deserializers` (7.1s)
   - `Camel :: Catalog :: Camel Route Parser` (6.4s)
   - `Camel :: JBang :: Plugin :: Testing` (6.1s)
   - `Camel :: Catalog :: Camel Report Maven Plugin` (5.2s)
   - `Camel :: JBang :: Plugin :: Validate` (4.5s)
   - `Camel :: All Components Sync point` (3.9s)
   - `Camel :: YAML DSL :: Validator Maven Plugin` (3.9s)
   - `Camel :: YAML DSL :: Maven Plugins` (2.5s)
   - `Camel :: Catalog :: Suggest (deprecated)` (2.1s)
   - `Camel :: Catalog :: Maven` (2.0s)
   
   </details>
   
   ---
   :gear: [View full build and test results](https://github.com/apache/camel/actions/runs/37391961775)


-- 
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: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-25313: camel-kafka - a suspended consumer must keep its fetcher thread, and a pause or resume request must not be lost [camel]

Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on code in PR #27389:
URL: https://github.com/apache/camel/pull/27389#discussion_r4192810054


##########
docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc:
##########
@@ -3934,6 +3934,15 @@ the SSL endpoint options were ignored when `sslContextParameters` was configured
 When `sslContextParameters` is configured and `securityProtocol` is the default (`PLAINTEXT`), the security protocol
 `SSL` is now used. Previously `PLAINTEXT` was used unless `securityProtocol` (or `saslAuthType`) was configured.
 
+==== A suspended consumer keeps its partitions

Review Comment:
   Thanks, done in 34318102316a: `=== camel-kafka - a suspended consumer keeps its partitions`.
   
   _Claude Code on behalf of allthingssecurity_



-- 
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: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org