This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 4b9ccda91de1 CAMEL-20428: camel-kafka - expose the batch-wide topic
and partition on the batch exchange (#25006)
4b9ccda91de1 is described below
commit 4b9ccda91de1044d8e2cb073f9e3e058a60365b1
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Jul 23 13:24:59 2026 +0200
CAMEL-20428: camel-kafka - expose the batch-wide topic and partition on the
batch exchange (#25006)
The batch exchange now also carries CamelKafkaTopic and
CamelKafkaPartition, but only when every record in the batch agrees on the
value, so they can be read before splitting the batch.
Co-authored-by: Claude Fable 5 <[email protected]>
---
.../apache/camel/catalog/docs/kafka-component.adoc | 20 ++++
.../camel-kafka/src/main/docs/kafka-component.adoc | 20 ++++
.../batching/KafkaRecordBatchingProcessor.java | 43 +++++++
...kaRecordBatchingProcessorCommonHeadersTest.java | 123 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 13 +++
5 files changed, 219 insertions(+)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kafka-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kafka-component.adoc
index d45a41e25a44..8519f6a16845 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kafka-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kafka-component.adoc
@@ -758,6 +758,26 @@ receiving new messages from the broker. Camel is not
active during this processi
has been configured with a high value, then Camel cannot trigger batch timeout
or interval completion ahead
of time. Therefore, it's recommended to keep this value as default.
+==== Batch Headers
+
+The exchange carrying the batch also exposes the record metadata headers that
every record in the batch agrees on:
+`CamelKafkaTopic` and `CamelKafkaPartition`. This makes it possible to read
them before splitting the batch, for
+example to store the topic in a variable and reuse it after the split:
+
+._Java-only: reading the topic from the batch before splitting_
+[source,java]
+----
+from("kafka:topic?groupId=myGroup&batching=true&maxPollRecords=10")
+ .setVariable("topic", header(KafkaConstants.TOPIC))
+ .split(body())
+ .log("Record from ${variable.topic}")
+ .end();
+----
+
+A header is only set on the batch when *every* record in the batch carries the
same value for it. If the batch spans
+multiple topics or partitions, the corresponding header is not set, since no
single value would be correct for the
+batch as a whole. Per-record headers that naturally differ, such as
`CamelKafkaOffset`, are never set on the batch —
+read those from the individual exchanges in the body.
==== Automatic Commits
diff --git a/components/camel-kafka/src/main/docs/kafka-component.adoc
b/components/camel-kafka/src/main/docs/kafka-component.adoc
index d45a41e25a44..8519f6a16845 100644
--- a/components/camel-kafka/src/main/docs/kafka-component.adoc
+++ b/components/camel-kafka/src/main/docs/kafka-component.adoc
@@ -758,6 +758,26 @@ receiving new messages from the broker. Camel is not
active during this processi
has been configured with a high value, then Camel cannot trigger batch timeout
or interval completion ahead
of time. Therefore, it's recommended to keep this value as default.
+==== Batch Headers
+
+The exchange carrying the batch also exposes the record metadata headers that
every record in the batch agrees on:
+`CamelKafkaTopic` and `CamelKafkaPartition`. This makes it possible to read
them before splitting the batch, for
+example to store the topic in a variable and reuse it after the split:
+
+._Java-only: reading the topic from the batch before splitting_
+[source,java]
+----
+from("kafka:topic?groupId=myGroup&batching=true&maxPollRecords=10")
+ .setVariable("topic", header(KafkaConstants.TOPIC))
+ .split(body())
+ .log("Record from ${variable.topic}")
+ .end();
+----
+
+A header is only set on the batch when *every* record in the batch carries the
same value for it. If the batch spans
+multiple topics or partitions, the corresponding header is not set, since no
single value would be correct for the
+batch as a whole. Per-record headers that naturally differ, such as
`CamelKafkaOffset`, are never set on the batch —
+read those from the individual exchanges in the body.
==== Automatic Commits
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/batching/KafkaRecordBatchingProcessor.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/batching/KafkaRecordBatchingProcessor.java
index ad81b501732b..8efa0eeeea7b 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/batching/KafkaRecordBatchingProcessor.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/batching/KafkaRecordBatchingProcessor.java
@@ -45,6 +45,12 @@ final class KafkaRecordBatchingProcessor extends
KafkaRecordProcessor {
private static final Logger LOG =
LoggerFactory.getLogger(KafkaRecordBatchingProcessor.class);
+ /**
+ * Record metadata headers that identify where a batch came from, and are
therefore worth exposing on the batch
+ * exchange itself when the whole batch agrees on them.
+ */
+ private static final List<String> COMMON_BATCH_HEADERS =
List.of(KafkaConstants.TOPIC, KafkaConstants.PARTITION);
+
private final KafkaConfiguration configuration;
private final Processor processor;
private final CommitManager commitManager;
@@ -194,6 +200,42 @@ final class KafkaRecordBatchingProcessor extends
KafkaRecordProcessor {
return timeout || interval;
}
+ /**
+ * Copies the record metadata headers that are identical across every
record in the batch onto the batch exchange,
+ * so a route can read them - for example to store the topic in a variable
- before splitting the batch.
+ * <p>
+ * A header is only propagated when every record in the batch carries the
same value for it. Headers that vary
+ * within the batch are left off, because no single value would be correct
for the batch as a whole.
+ */
+ static void propagateCommonHeaders(List<Exchange> exchanges, Message
batchMessage) {
+ for (String header : COMMON_BATCH_HEADERS) {
+ Object common = commonHeaderValue(exchanges, header);
+ if (common != null) {
+ batchMessage.setHeader(header, common);
+ }
+ }
+ }
+
+ /**
+ * Returns the value of the given header when every exchange in the batch
carries the same non-null value for it,
+ * otherwise null.
+ */
+ private static Object commonHeaderValue(List<Exchange> exchanges, String
header) {
+ Object common = null;
+ for (Exchange exchange : exchanges) {
+ Object value = exchange.getMessage().getHeader(header);
+ if (value == null) {
+ return null;
+ }
+ if (common == null) {
+ common = value;
+ } else if (!common.equals(value)) {
+ return null;
+ }
+ }
+ return common;
+ }
+
private Map<TopicPartition, Long> computeBatchOffsets(List<Exchange>
exchanges) {
Map<TopicPartition, Long> offsets = new HashMap<>();
for (Exchange ex : exchanges) {
@@ -217,6 +259,7 @@ final class KafkaRecordBatchingProcessor extends
KafkaRecordProcessor {
Message message = exchange.getMessage();
var exchanges = exchangeList.stream().toList();
message.setBody(exchanges);
+ propagateCommonHeaders(exchanges, message);
ProcessingResult result = ProcessingResult.newUnprocessed();
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/batching/KafkaRecordBatchingProcessorCommonHeadersTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/batching/KafkaRecordBatchingProcessorCommonHeadersTest.java
new file mode 100644
index 000000000000..88f0d6553327
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/batching/KafkaRecordBatchingProcessorCommonHeadersTest.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka.consumer.support.batching;
+
+import java.util.List;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.component.kafka.KafkaConstants;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Verifies which record metadata headers are propagated from the records in a
batch onto the batch exchange.
+ */
+class KafkaRecordBatchingProcessorCommonHeadersTest {
+
+ private CamelContext context;
+
+ @BeforeEach
+ void setUp() {
+ context = new DefaultCamelContext();
+ }
+
+ @AfterEach
+ void tearDown() {
+ context.stop();
+ }
+
+ private Exchange record(String topic, Integer partition, Long offset) {
+ Exchange exchange = new DefaultExchange(context);
+ Message message = exchange.getMessage();
+ message.setHeader(KafkaConstants.TOPIC, topic);
+ message.setHeader(KafkaConstants.PARTITION, partition);
+ message.setHeader(KafkaConstants.OFFSET, offset);
+ return exchange;
+ }
+
+ private Message batchMessageFor(List<Exchange> records) {
+ Exchange batch = new DefaultExchange(context);
+ KafkaRecordBatchingProcessor.propagateCommonHeaders(records,
batch.getMessage());
+ return batch.getMessage();
+ }
+
+ @Test
+ void uniformTopicAndPartitionArePropagated() {
+ Message batch = batchMessageFor(List.of(
+ record("orders", 0, 1L),
+ record("orders", 0, 2L)));
+
+ assertEquals("orders", batch.getHeader(KafkaConstants.TOPIC));
+ assertEquals(0, batch.getHeader(KafkaConstants.PARTITION));
+ }
+
+ @Test
+ void offsetIsNeverPropagatedBecauseItVariesPerRecord() {
+ Message batch = batchMessageFor(List.of(
+ record("orders", 0, 1L),
+ record("orders", 0, 2L)));
+
+ assertNull(batch.getHeader(KafkaConstants.OFFSET));
+ }
+
+ @Test
+ void mixedTopicIsNotPropagated() {
+ Message batch = batchMessageFor(List.of(
+ record("orders", 0, 1L),
+ record("shipments", 0, 2L)));
+
+ assertNull(batch.getHeader(KafkaConstants.TOPIC));
+ // the partition is still uniform across the batch, so it is still
propagated
+ assertEquals(0, batch.getHeader(KafkaConstants.PARTITION));
+ }
+
+ @Test
+ void mixedPartitionIsNotPropagated() {
+ Message batch = batchMessageFor(List.of(
+ record("orders", 0, 1L),
+ record("orders", 1, 2L)));
+
+ assertEquals("orders", batch.getHeader(KafkaConstants.TOPIC));
+ assertNull(batch.getHeader(KafkaConstants.PARTITION));
+ }
+
+ @Test
+ void missingHeaderOnAnyRecordIsNotPropagated() {
+ Message batch = batchMessageFor(List.of(
+ record("orders", 0, 1L),
+ record("orders", null, 2L)));
+
+ assertEquals("orders", batch.getHeader(KafkaConstants.TOPIC));
+ assertNull(batch.getHeader(KafkaConstants.PARTITION));
+ }
+
+ @Test
+ void emptyBatchSetsNoHeaders() {
+ Message batch = batchMessageFor(List.of());
+
+ assertNull(batch.getHeader(KafkaConstants.TOPIC));
+ assertNull(batch.getHeader(KafkaConstants.PARTITION));
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index 1a8ceba558ae..db480bb442a8 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -811,6 +811,19 @@ affects code constructing the dispatcher directly; the
client must be started be
unchanged — redirects were already never followed for push webhooks, and still
are not, since a redirect would be
resolved by the client and escape the validation applied to the registered URL.
+=== camel-kafka - batch exchange carries the batch-wide topic and partition
+
+When using the batching consumer (`batching=true`), the exchange carrying the
batch now also has the
+`CamelKafkaTopic` and `CamelKafkaPartition` headers set, so they can be read
before the batch is split — for example
+to store the topic in a variable and reuse it after the split.
+
+A header is only set when *every* record in the batch has the same value for
it; if the batch spans multiple topics
+or partitions, that header is left unset, because no single value would be
correct for the batch. Per-record headers
+that naturally differ, such as `CamelKafkaOffset`, are not set on the batch
exchange.
+
+This is additive — these headers were previously absent from the batch
exchange, so nothing that worked before
+changes. The headers on the individual record exchanges in the body are
unchanged.
+
=== camel-core - Multicast UseOriginalAggregationStrategy fix
The Multicast EIP now correctly honors `UseOriginalAggregationStrategy`,
consistent with the Splitter
