oscerd opened a new pull request, #25006:
URL: https://github.com/apache/camel/pull/25006

   ### Motivation
   
   [CAMEL-20428](https://issues.apache.org/jira/browse/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.
   
   ```java
   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](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]

Reply via email to