This is an automated email from the ASF dual-hosted git repository.
Croway 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 45f4138f5e16 CAMEL-24797: camel-core - Splitter unwraps Exchange parts
like Message parts
45f4138f5e16 is described below
commit 45f4138f5e160590157cf6da67e49b5a4f0c3836
Author: croway <[email protected]>
AuthorDate: Thu Sep 17 16:15:35 2026 +0200
CAMEL-24797: camel-core - Splitter unwraps Exchange parts like Message parts
Splitting a List<Exchange> body (camel-kafka batching consumer, grouped
exchange aggregation strategy) now copies the body and headers of each
part into the child exchange, instead of making the part Exchange itself
the child body with no headers. The message is copied, not adopted, so
the exchanges owned by a batching consumer stay intact after the split.
Updates KafkaConsumerAsyncManualCommitIT and the Kafka batch headers docs
for the new shape, and documents the behaviour change in the 4.23 upgrade
guide.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
---
.../apache/camel/catalog/docs/kafka-component.adoc | 10 +-
.../camel-kafka/src/main/docs/kafka-component.adoc | 10 +-
.../KafkaConsumerAsyncManualCommitIT.java | 4 +-
.../java/org/apache/camel/processor/Splitter.java | 4 +
.../camel/processor/SplitListOfExchangesTest.java | 107 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 25 +++++
6 files changed, 148 insertions(+), 12 deletions(-)
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 ba38a3af68de..9a9b669a78ec 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
@@ -761,16 +761,16 @@ 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:
+`CamelKafkaTopic` and `CamelKafkaPartition`. This makes it possible to read
them on the batch itself, before
+splitting it. When the batch is split, each child exchange carries the body
and the headers of its own record:
-._Java-only: reading the topic from the batch before splitting_
+._Java-only: reading the topic from the batch and the offset from each record_
[source,java]
----
from("kafka:topic?groupId=myGroup&batching=true&maxPollRecords=10")
- .setVariable("topic", header(KafkaConstants.TOPIC))
+ .log("Received a batch of ${body.size} records from
${header.CamelKafkaTopic}")
.split(body())
- .log("Record from ${variable.topic}")
+ .log("Record ${body} at offset ${header.CamelKafkaOffset}")
.end();
----
diff --git a/components/camel-kafka/src/main/docs/kafka-component.adoc
b/components/camel-kafka/src/main/docs/kafka-component.adoc
index ba38a3af68de..9a9b669a78ec 100644
--- a/components/camel-kafka/src/main/docs/kafka-component.adoc
+++ b/components/camel-kafka/src/main/docs/kafka-component.adoc
@@ -761,16 +761,16 @@ 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:
+`CamelKafkaTopic` and `CamelKafkaPartition`. This makes it possible to read
them on the batch itself, before
+splitting it. When the batch is split, each child exchange carries the body
and the headers of its own record:
-._Java-only: reading the topic from the batch before splitting_
+._Java-only: reading the topic from the batch and the offset from each record_
[source,java]
----
from("kafka:topic?groupId=myGroup&batching=true&maxPollRecords=10")
- .setVariable("topic", header(KafkaConstants.TOPIC))
+ .log("Received a batch of ${body.size} records from
${header.CamelKafkaTopic}")
.split(body())
- .log("Record from ${variable.topic}")
+ .log("Record ${body} at offset ${header.CamelKafkaOffset}")
.end();
----
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAsyncManualCommitIT.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAsyncManualCommitIT.java
index 0d45195830c2..02bf9e90c9a4 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAsyncManualCommitIT.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerAsyncManualCommitIT.java
@@ -93,8 +93,8 @@ public class KafkaConsumerAsyncManualCommitIT extends
BaseKafkaTestSupport {
.aggregationStrategy(AggregationStrategies.groupedExchange())
.split().body()
.process(e -> {
- KafkaManualCommit manual =
e.getMessage().getBody(Exchange.class)
-
.getMessage().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
+ KafkaManualCommit manual
+ =
e.getMessage().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
assertNotNull(manual);
try {
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java
index 78a89bdc33ba..050872386d75 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java
@@ -422,6 +422,10 @@ public class Splitter extends MulticastProcessor {
}
if (part instanceof Message message) {
newExchange.setIn(message);
+ } else if (part instanceof Exchange partExchange) {
+ // copy (do not adopt) the message, so the
exchange owning the part, such as the batch
+ // exchange of a batching consumer, is left
untouched when the child exchange completes
+
newExchange.getIn().copyFrom(partExchange.getMessage());
} else {
Message in = newExchange.getIn();
in.setBody(part);
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/SplitListOfExchangesTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitListOfExchangesTest.java
new file mode 100644
index 000000000000..5510530c2116
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitListOfExchangesTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.processor;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedCamelContext;
+import org.apache.camel.Message;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.engine.PooledExchangeFactory;
+import org.apache.camel.impl.engine.PooledProcessorExchangeFactory;
+import org.apache.camel.support.DefaultExchange;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Splitting a body of type {@code List<Exchange>}, as produced by batching
consumers and the grouped exchange
+ * aggregation strategy, should unwrap each part into the child exchange, like
splitting a {@code List<Message>} does.
+ * Exchange pooling is enabled because a pooled child exchange resets its
message on completion, which would wipe the
+ * parts if their message was adopted instead of copied.
+ */
+public class SplitListOfExchangesTest extends ContextTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ ExtendedCamelContext ecc = camelContext.getCamelContextExtension();
+ ecc.setExchangeFactory(new PooledExchangeFactory());
+ ecc.setProcessorExchangeFactory(new PooledProcessorExchangeFactory());
+ return camelContext;
+ }
+
+ @Test
+ public void testSplitListOfExchanges() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:split");
+ mock.expectedBodiesReceived("A", "B", "C");
+ mock.expectedHeaderValuesReceivedInAnyOrder("foo", 1, 2, 3);
+
+ List<Exchange> parts = createParts();
+ template.sendBody("direct:start", parts);
+
+ assertMockEndpointsSatisfied();
+
+ for (Exchange child : mock.getReceivedExchanges()) {
+ assertEquals("bar", child.getMessage().getHeader("common"));
+ // the part is unwrapped, the child body is not the part exchange
itself
+ assertNull(child.getMessage().getBody(Exchange.class));
+ // exchange properties of the part are not carried over (same as
for Message parts)
+ assertNull(child.getProperty("partProperty"));
+ }
+
+ // the owner of the parts (e.g. a batching consumer committing after
the route) still sees them intact
+ for (int i = 0; i < parts.size(); i++) {
+ Message part = parts.get(i).getMessage();
+ assertEquals(List.of("A", "B", "C").get(i), part.getBody());
+ assertEquals(i + 1, part.getHeader("foo"));
+ assertEquals("bar", part.getHeader("common"));
+ }
+ }
+
+ private List<Exchange> createParts() {
+ List<Exchange> parts = new ArrayList<>();
+ int i = 1;
+ for (String body : List.of("A", "B", "C")) {
+ Exchange part = new DefaultExchange(context);
+ Message message = part.getMessage();
+ message.setBody(body);
+ message.setHeader("foo", i++);
+ message.setHeader("common", "bar");
+ part.setProperty("partProperty", "notCopied");
+ parts.add(part);
+ }
+ return parts;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from("direct:start")
+ .split(body())
+ .to("mock:split");
+ }
+ };
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index 943072cffb7d..ea7421392502 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -478,6 +478,31 @@ The XML schemas (`camel-spring.xsd`, `camel-xml-io.xsd`)
and the YAML schema no
ran. A bean that is neither scripted nor built still needs a `type`, and the
error for a missing one
is now an `IllegalArgumentException` naming the bean instead of a
`NullPointerException`.
+=== camel-core - Splitter unwraps Exchange parts
+
+The Splitter EIP now unwraps parts of type `org.apache.camel.Exchange`, the
same way it already unwrapped parts of
+type `org.apache.camel.Message`. When splitting a body of type
`List<Exchange>`, as produced by the camel-kafka
+batching consumer (`batching=true`) and by the grouped exchange aggregation
strategy, each child exchange now
+carries the body and the headers of the corresponding part. Previously the
child body was the part `Exchange`
+itself, with no headers.
+
+Code that read the part from the child body must be updated, for example when
performing a Kafka manual commit
+after a split:
+
+[source,java]
+----
+// before
+KafkaManualCommit manual = exchange.getMessage().getBody(Exchange.class)
+ .getMessage().getHeader(KafkaConstants.MANUAL_COMMIT,
KafkaManualCommit.class);
+
+// now
+KafkaManualCommit manual =
exchange.getMessage().getHeader(KafkaConstants.MANUAL_COMMIT,
KafkaManualCommit.class);
+----
+
+The message of the part is copied into the child exchange, so the exchanges in
the original list are left
+untouched and remain usable after the split, for example by the batching
consumer that owns them. Exchange
+properties of the part are not carried over to the child exchange, as is
already the case for `Message` parts.
+
=== Component deprecation
==== camel-minio
