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/19 00:10:34 UTC
[PR] CAMEL-25756: camel-kafka - a failed poll must not drop the records of an incomplete batch [camel]
allthingssecurity opened a new pull request, #28071:
URL: https://github.com/apache/camel/pull/28071
# Description
[CAMEL-25756](https://issues.apache.org/jira/browse/CAMEL-25756)
With `batching=true` the Kafka consumer keeps the records of several polls in `KafkaRecordBatchingProcessor` until the batch is complete (`maxPollRecords` records), the poll timeout expires or `batchingIntervalMs` passes. `KafkaFetchRecords.startPolling()` created a new record processor facade, and so a new, empty batch, every time it was entered. It is entered again on the same connected Kafka consumer after every exception of the polling loop that the poll exception strategy continues on: a failed `poll` with the default `pollOnError=ERROR_HANDLER`, and with `DISCARD` or `RETRY`.
The records held for the incomplete batch were then dropped. The Kafka consumer does not return them again (its position is past them), and the next complete batch commits its highest offset per partition, past them. Once a later batch of the same partition was committed, they were never routed and never redelivered, even after a restart, and nothing was logged about them. (If no later batch of that partition was committed before a stop or a revocation, they were polled again from the committed offset.)
A realistic trigger is a record that a failing deserializer (Avro, JSON schema, a custom one) cannot read: `poll` throws `RecordDeserializationException`, the default error handler skips that record, and the good records polled before it (on every partition) were lost with it. Up to `maxPollRecords - 1` records per failed poll.
This change creates the facade when the Kafka consumer is connected or reconnected, and keeps it while the consumer is connected. A failed poll keeps the held records; they are routed and committed with the next batch. After a reconnect the new Kafka consumer starts from the committed offsets, so a new facade, with an empty batch, is still created then. The streaming facade holds no state between polls and behaves as before. `KafkaFetchRecords` is the only production change.
Not changed:
- `breakOnFirstError` in batching mode still commits the failed batch before the reconnect (as `KafkaBatchingProcessingBreakOnFirstErrorIT` expects);
- the records held for an incomplete batch are still dropped when the route stops; they were not committed, so they are polled again after the restart.
Tests:
- New `KafkaBatchingConsumerPollErrorTest` (2 tests, `MockConsumer`, Awaitility): a real route with `batching=true&maxPollRecords=10` and Kafka's `MockConsumer`. Three records are polled and held, then a poll fails with a `KafkaException` (first test) or with a `RecordDeserializationException` for offset 3 (second test), then ten more records arrive. Each test checks the records of the first batch and the committed offset.
- Without the change (main's `KafkaFetchRecords`, same tests) both fail: `expected: <[a0, a1, a2, b0, b1, b2, b3, b4, b5, b6]> but was: <[b0, b1, b2, b3, b4, b5, b6, b7, b8, b9]>`.
- With the change, the camel-kafka unit tests pass: 222 run, 0 failures, 1 skipped. The integration tests need a Kafka broker and were not run.
How it was found: deterministic simulation testing with a fake Kafka consumer written from the KafkaConsumer Javadoc. The seed picked the consumer configuration (streaming or batching, commit managers, `breakOnFirstError`, `maxPollRecords`, timeouts), records that cannot be deserialized, failed polls and commits, and route restarts and suspends. The invariant "every record sent reaches the route" failed only with `batching=true` and a failed poll on main; with this change those seeds pass. The simulation harness is not part of this PR.
# 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 camel-kafka, 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-25756: camel-kafka - a failed poll must not drop the records of an incomplete batch [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus closed pull request #28071: CAMEL-25756: camel-kafka - a failed poll must not drop the records of an incomplete batch
URL: https://github.com/apache/camel/pull/28071
--
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-25756: camel-kafka - a failed poll must not drop the records of an incomplete batch [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on PR #28071:
URL: https://github.com/apache/camel/pull/28071#issuecomment-6266378110
@allthingssecurity I pushed the follow-up we agreed on to this PR as c643569f9ef0, so it ships together with the first commit. `KafkaRecordBatchingProcessor` now clears the batch, and restarts the poll timer, in a `finally` around `processBatch`. The new `KafkaBatchingConsumerProcessErrorTest` makes the consumer's exception handler throw for a failed batch with `allowManualCommit=true`: without the change the next records fail with `Queue full`. The camel-kafka unit tests pass (223 run, 1 skipped). The PR description is updated.
_Claude Code on behalf of Claus Ibsen_
--
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-25756: camel-kafka - a failed poll must not drop the records of an incomplete batch [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on PR #28071:
URL: https://github.com/apache/camel/pull/28071#issuecomment-6266914649
Thanks a lot for taking care of the follow-up, Claus. c643569f9ef0 looks good to me: clearing the batch and restarting the timer in the `finally` closes the queue-full case, and the new test covers it.
_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-25756: camel-kafka - a failed poll must not drop the records of an incomplete batch [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #28071:
URL: https://github.com/apache/camel/pull/28071#discussion_r4310717013
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java:
##########
@@ -91,6 +91,9 @@ private enum State {
private final BridgeExceptionHandlerToErrorHandler bridge;
private final ReentrantLock lock = new ReentrantLock();
private CommitManager commitManager;
+ // kept while the Kafka consumer is connected: the batching processor holds the records of earlier polls until its
+ // batch is complete, and the consumer does not return them again
+ private KafkaRecordProcessorFacade recordProcessorFacade;
Review Comment:
Non-blocking: the facade now lives as long as the connection, so a fresh facade on every `startPolling()` no longer resets a broken `KafkaRecordBatchingProcessor` by accident. In `processExchange`, `exchangeList.clear()` runs only after `processBatch(...)` returns. If `processBatch` ever throws (it catches processor exceptions, so this is unlikely), the `ArrayBlockingQueue` (capacity `maxPollRecords`) stays full. Then every later `exchangeList.add(...)` throws `IllegalStateException: Queue full` until a reconnect. Moving the `clear()` (and the timer restart) into a `finally` around `processBatch` would close that gap. This could be a follow-up.
--
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-25756: camel-kafka - a failed poll must not drop the records of an incomplete batch [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on code in PR #28071:
URL: https://github.com/apache/camel/pull/28071#discussion_r4310844323
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java:
##########
@@ -91,6 +91,9 @@ private enum State {
private final BridgeExceptionHandlerToErrorHandler bridge;
private final ReentrantLock lock = new ReentrantLock();
private CommitManager commitManager;
+ // kept while the Kafka consumer is connected: the batching processor holds the records of earlier polls until its
+ // batch is complete, and the consumer does not return them again
+ private KafkaRecordProcessorFacade recordProcessorFacade;
Review Comment:
Agreed, thanks. With the facade living as long as the connection, a `processBatch` that throws would leave the queue full until a reconnect. I will move the `clear()` and the timer restart into a `finally` in a small follow-up, so this approved commit stays as it is.
_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-25756: camel-kafka - a failed poll must not drop the records of an incomplete batch [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on PR #28071:
URL: https://github.com/apache/camel/pull/28071#issuecomment-6261562777
Closing and reopening to re-run CI against main: the `CatalogSamplesTest.callingAServiceIsTheHttpComponentNotEnrich` failure in camel-jbang-core came from main and is fixed there (bbd6f750aa00). Nothing to change in this PR.
_Claude Code on behalf of Claus Ibsen_
--
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-25756: camel-kafka - a failed poll must not drop the records of an incomplete batch [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on PR #28071:
URL: https://github.com/apache/camel/pull/28071#issuecomment-6266401168
Thanks a lot for pushing it here, that is cleaner than a separate follow-up. I read c643569f9ef0 and the new test: clearing the batch and restarting the timer in the `finally` closes the `Queue full` case, and the description matches both commits.
_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-25756: camel-kafka - a failed poll must not drop the records of an incomplete batch [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #28071:
URL: https://github.com/apache/camel/pull/28071#issuecomment-6260992483
<!-- ci-tested-modules -->
:test_tube: **CI tested the following changed modules:**
- `components/camel-kafka`
---
<details><summary>:microscope: Scalpel shadow comparison — Scalpel: 8 of 706 tested, 27 compile-only — current: 8 all tested</summary>
[Maveniverse Scalpel](https://github.com/maveniverse/scalpel) detected **8 affected modules** (current approach: 8).
**Skip-tests mode would test 8 modules** (1 direct + 7 downstream), **skip tests for 27** (generated code, meta-modules)
<details><summary>Modules Scalpel would test (8)</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-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/KafkaBatchingConsumerPollErrorTest.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 (35 modules, 3m 10s total)</summary>
**Total reactor time:** 3m 10s
| Module | Duration | Status |
| --- | --- | --- |
| Camel :: YAML DSL | 26.8s | SUCCESS |
| Camel :: YAML DSL :: Validator | 26.0s | SUCCESS |
| Camel :: Catalog :: Camel Catalog | 24.5s | SUCCESS |
| Camel :: Component DSL | 22.1s | SUCCESS |
| Camel :: Kamelet Main | 19.4s | SUCCESS |
| Camel :: Docs | 16.4s | SUCCESS |
| Camel :: YAML DSL :: Deserializers | 8.7s | SUCCESS |
| Camel :: Assembly | 8.2s | SUCCESS |
| Camel :: Catalog :: Camel Route Parser | 7.8s | SUCCESS |
| Camel :: Catalog :: Camel Report Maven Plugin | 7.3s | SUCCESS |
| Camel :: All Components Sync point | 5.7s | SUCCESS |
| Camel :: YAML DSL :: Validator Maven Plugin | 5.0s | SUCCESS |
| Camel :: Catalog :: Maven | 3.5s | SUCCESS |
| Camel :: YAML DSL :: Maven Plugins | 3.2s | SUCCESS |
| Camel :: Catalog :: Suggest (deprecated) | 2.8s | SUCCESS |
| Camel :: Catalog :: Dummy Component | 1.0s | SUCCESS |
| Camel :: Catalog :: Console | 1.0s | SUCCESS |
| Camel :: Endpoint DSL :: Support | 1.0s | SUCCESS |
| Camel :: Coverage | n/a | |
| Camel :: Endpoint DSL | n/a | |
| Camel :: Integration Tests | n/a | |
| Camel :: JBang :: Core | n/a | |
| Camel :: JBang :: Integration tests | n/a | |
| Camel :: JBang :: MCP | n/a | |
| Camel :: JBang :: Main | n/a | |
| Camel :: JBang :: Plugin :: Edit | n/a | |
| Camel :: JBang :: Plugin :: Generate | n/a | |
| Camel :: JBang :: Plugin :: Kubernetes | n/a | |
| Camel :: JBang :: Plugin :: MCP | n/a | |
| Camel :: JBang :: Plugin :: TUI | n/a | |
| Camel :: JBang :: Plugin :: Testing | n/a | |
| Camel :: JBang :: Plugin :: Validate | n/a | |
| Camel :: Kafka | n/a | |
| Camel :: Launcher | n/a | |
| Camel :: Launcher :: Container | n/a | |
**Top 20 slowest modules:**
- `Camel :: YAML DSL` (26.8s)
- `Camel :: YAML DSL :: Validator` (26.0s)
- `Camel :: Catalog :: Camel Catalog` (24.5s)
- `Camel :: Component DSL` (22.1s)
- `Camel :: Kamelet Main` (19.4s)
- `Camel :: Docs` (16.4s)
- `Camel :: YAML DSL :: Deserializers` (8.7s)
- `Camel :: Assembly` (8.2s)
- `Camel :: Catalog :: Camel Route Parser` (7.8s)
- `Camel :: Catalog :: Camel Report Maven Plugin` (7.3s)
- `Camel :: All Components Sync point` (5.7s)
- `Camel :: YAML DSL :: Validator Maven Plugin` (5.0s)
- `Camel :: Catalog :: Maven` (3.5s)
- `Camel :: YAML DSL :: Maven Plugins` (3.2s)
- `Camel :: Catalog :: Suggest (deprecated)` (2.8s)
- `Camel :: Catalog :: Dummy Component` (1.0s)
- `Camel :: Catalog :: Console` (1.0s)
- `Camel :: Endpoint DSL :: Support` (1.0s)
</details>
---
:gear: [View full build and test results](https://github.com/apache/camel/actions/runs/39179588899)
--
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-25756: camel-kafka - a failed poll must not drop the records of an incomplete batch [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #28071:
URL: https://github.com/apache/camel/pull/28071#issuecomment-6260652904
: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-25756: camel-kafka - a failed poll must not drop the records of an incomplete batch [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus merged PR #28071:
URL: https://github.com/apache/camel/pull/28071
--
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