CAMEL-20428: camel-kafka - expose the batch-wide topic and partition on the batch exchange

[oscerd]

Motivation

CAMEL-20428. With the batching consumer, the exchange carrying the batch had only the record list as its body (plus the manual-commit header). The record metadata was reachable only by digging into the child exchanges, which makes it awkward to use the topic in the route before the batch is split — the issue's example being to store the topic in a variable and reuse it after the split.

Changes

KafkaRecordBatchingProcessor now propagates the record metadata headers that every record in the batch agrees on onto the batch exchange: CamelKafkaTopic and CamelKafkaPartition.

The "common" rule is the important part:

• A header is set only when all records in the batch carry the same non-null value for it.
• If the batch spans multiple topics or partitions, that header is left unset — no single value would be correct for the batch as a whole, and setting an arbitrary one (e.g. the first record's) would be quietly misleading.
• Per-record headers that naturally differ, such as CamelKafkaOffset, are therefore never set on the batch. Those stay readable on the individual exchanges in the body.

This is additive: these headers were previously absent from the batch exchange, so nothing that worked before changes, and the headers on the individual record exchanges are untouched.

from("kafka:topic?groupId=myGroup&batching=true&maxPollRecords=10")
    .setVariable("topic", header(KafkaConstants.TOPIC))
    .split(body())
        .log("Record from ${variable.topic}")
    .end();

Testing

• 6 new unit tests covering uniform topic+partition, mixed topic, mixed partition, a header missing on one record, offset never propagated, and the empty batch. The propagation logic is a package-private static method so it is testable without a broker (the existing batching tests are all ITs).
• mvn test -Dtest=KafkaRecordBatchingProcessorCommonHeadersTest: 6/6 pass.
• Full reactor mvn clean install -DskipTests from root: success, no stale generated files.

Note: the tests use JUnit assertions rather than AssertJ — AssertJ is not a test dependency of camel-kafka, and the module's existing tests use JUnit assertions throughout, so adding a dependency purely for this test did not seem justified.

Docs: new Batch Headers section in the Kafka component docs, plus an upgrade-guide entry. Main-only (4.22.0), no backport.

Claude Code on behalf of Andrea Cosentino (@oscerd).

🤖 Generated with Claude Code

[gnodet]

Claude Code on behalf of gnodet

✅ LGTM — Well-designed batch header propagation

The "all records must agree" rule is the right design choice — setting a header from the first record would be silently misleading when a batch spans multiple topics or partitions. The implementation is clean and correct:

• commonHeaderValue() — simple linear scan, null on first disagreement or missing value. Correct for the expected batch sizes.
• COMMON_BATCH_HEADERS — List.of(TOPIC, PARTITION) makes it trivially extensible if more headers qualify in the future.
• Fully additive — these headers were previously absent from the batch exchange, so zero backward compatibility risk.
• Offset implicitly excluded by not being in COMMON_BATCH_HEADERS (rather than special-cased in the logic) — clean separation.

Tests: 6 targeted tests covering the key scenarios (uniform, mixed topic, mixed partition, missing header, offset exclusion, empty batch). Testing via the package-private static method without a Kafka broker is the right call for a pure logic function.

Docs: both the component doc (new "Batch Headers" section with example) and upgrade guide entry are well-written and thorough.

JUnit assertions are appropriate here since camel-kafka doesn't have an AssertJ test dependency.

No concerns.

[davsclaus]

merge conflict to resolve

[davsclaus]

Clean, well-thought-out PR. The approach is conservative — only propagating headers when all records in the batch agree — and fully additive (previously absent headers are now conditionally set). No existing behavior changes.

Confirmed: No Issues

• Code logic — commonHeaderValue() correctly handles all edge cases: empty list, single record, uniform values, mixed values, and null values. The COMMON_BATCH_HEADERS constant is an immutable List.of(), thread-safe as a static final field.
• Call site placement — propagateCommonHeaders() is called in processBatch() right after message.setBody(exchanges), before processing begins. Both call paths (expired-batch and full-batch) will propagate headers.
• Test coverage — 6 unit tests covering uniform topic+partition, mixed topic, mixed partition, null header on one record, offset never propagated, and empty batch.
• Assertion style — JUnit assertions are correct here since AssertJ is not a test dependency in camel-kafka.
• Documentation — New "Batch Headers" section and upgrade guide entry are well-placed and clear.
• Conventions — Branch name, commit format, test visibility, and doc conventions all followed.

This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.

Claude Code on behalf of Claus Ibsen (@davsclaus).

[orpiske]

Nice work on this, Andrea — clean implementation and solid test coverage.

Review summary:

• The commonHeaderValue logic correctly handles all edge cases (empty batch, uniform values, mixed values, null on any record).
• The call site in processBatch() is correctly placed — headers are set before the route processes the exchange, and there's no interference with the breakOnFirstError handling (which reads from child exchanges, not the batch message).
• The "all-or-nothing" propagation approach is well-designed: setting an arbitrary value (e.g., the first record's) when the batch spans multiple topics/partitions would be misleading.
• Test coverage is thorough: 6 unit tests covering uniform topic+partition, mixed topic, mixed partition, null header, offset never propagated, and empty batch.

One optional suggestion (non-blocking): Recent commits (CAMEL-23789) have been replacing Java constant references with string header names and adding multi-DSL tabs in the component docs. The new "Batch Headers" section uses KafkaConstants.TOPIC in the example and is labeled "Java-only", which is honest — but it could optionally use string header names or add YAML/XML tabs to stay consistent with that direction.

This review does not replace specialized AI review tools (CodeRabbit, Sourcery) or static analysis (SonarCloud).

This review was generated by an AI agent (Claude Code on behalf of orpiske) and may contain inaccuracies. Please verify all suggestions before applying.

[github-actions]

🌟 Thank you for your contribution to the Apache Camel project! 🌟
🤖 CI automation will test this PR automatically.

🐫 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 have access to the summary.

⚠️ Be careful when sharing logs. Review their contents before sharing them publicly.

[gnodet]

Clean, well-implemented additive feature that propagates common batch metadata headers (topic, partition) onto the batch exchange when all records agree. Good test coverage, proper documentation, and correct backward compatibility.

Notable positives:

• Good design decision to use the "common value" approach — setting a header only when all records agree avoids misleading values when a batch spans multiple topics/partitions
• Tests correctly reside in the same package for unit testing without a Kafka broker
• Upgrade guide entry is clear and notes backward compatibility
• Correctly limits to TOPIC and PARTITION (not OFFSET/TIMESTAMP/KEY which naturally vary per record)

This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.

Claude Code on behalf of Guillaume Nodet

[github-actions]

🧪 CI tested the following changed modules:

• catalog/camel-catalog
• components/camel-kafka
• docs

---

🔬 Scalpel shadow comparison — Scalpel: 11 tested, 27 compile-only — current: 9 all tested

Maveniverse Scalpel detected 38 affected modules (current approach: 9).

⚠️ Modules only in Scalpel (29)

• apache-camel
• camel-allcomponents
• camel-catalog
• camel-catalog-console
• camel-catalog-lucene
• camel-catalog-maven
• camel-catalog-suggest
• camel-componentdsl
• camel-csimple-maven-plugin
• 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

Skip-tests mode would test 11 modules (3 direct + 8 downstream), skip tests for 27 (generated code, meta-modules)

Modules Scalpel would test (11)

• camel-catalog
• camel-jbang-mcp
• camel-jbang-plugin-mcp
• camel-jbang-plugin-route-parser
• camel-jbang-plugin-tui
• camel-jbang-plugin-validate
• camel-kafka
• camel-launcher-container
• camel-yaml-dsl-validator
• camel-yaml-dsl-validator-maven-plugin
• docs

Modules with tests skipped (27)

• apache-camel
• camel-allcomponents
• camel-catalog-console
• camel-catalog-lucene
• camel-catalog-maven
• camel-catalog-suggest
• camel-componentdsl
• camel-csimple-maven-plugin
• 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
• dummy-component

ℹ️ Shadow mode — Scalpel observes but does not affect test execution. Learn more

⚠️ 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

All tested modules (38 modules)

• Camel :: All Components Sync point
• Camel :: Assembly
• Camel :: Catalog :: CSimple Maven Plugin (deprecated)
• Camel :: Catalog :: Camel Catalog
• Camel :: Catalog :: Camel Report Maven Plugin
• Camel :: Catalog :: Camel Route Parser
• Camel :: Catalog :: Console
• Camel :: Catalog :: Dummy Component
• Camel :: Catalog :: Lucene (deprecated)
• Camel :: Catalog :: Maven
• Camel :: Catalog :: Suggest
• Camel :: Component DSL
• Camel :: Coverage
• Camel :: Docs
• Camel :: Endpoint DSL
• Camel :: Endpoint DSL :: Support
• Camel :: Integration Tests
• Camel :: JBang :: Core
• Camel :: JBang :: Integration tests
• Camel :: JBang :: MCP
• Camel :: JBang :: Main
• Camel :: JBang :: Plugin :: Edit
• Camel :: JBang :: Plugin :: Generate
• Camel :: JBang :: Plugin :: Kubernetes
• Camel :: JBang :: Plugin :: MCP
• Camel :: JBang :: Plugin :: Route Parser
• Camel :: JBang :: Plugin :: TUI
• Camel :: JBang :: Plugin :: Testing
• Camel :: JBang :: Plugin :: Validate
• Camel :: Kafka
• Camel :: Kamelet Main
• Camel :: Launcher
• Camel :: Launcher :: Container
• Camel :: YAML DSL
• Camel :: YAML DSL :: Deserializers
• Camel :: YAML DSL :: Maven Plugins
• Camel :: YAML DSL :: Validator
• Camel :: YAML DSL :: Validator Maven Plugin

---

⚙️ View full build and test results

[gnodet]

Re-review: the only new commit (ddb5dba) is a clean rebase resolving the merge conflict that davsclaus flagged. The diff is identical to the previously reviewed and approved code — same 5 files, 219 additions, same tests and documentation. Our prior approval still stands.

This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.

Claude Code on behalf of Guillaume Nodet