[
https://issues.apache.org/jira/browse/CAMEL-24797?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18116310#comment-18116310
]
Claus Ibsen edited comment on CAMEL-24797 at 9/17/26 11:44 AM:
---------------------------------------------------------------
Reviewed against the current code on main. The report checks out: {{Splitter}}
only unwraps {{Message}} parts ({{newExchange.setIn(message)}}, from CAMEL-1597
/ CAMEL-9252); an {{Exchange}} part falls into the generic branch and becomes
the child body, with no headers. And the Kafka "Batch Headers" docs from
CAMEL-20428 do recommend {{.split(body())}} over that shape, so we currently
document a split the Splitter does not really support.
Claus agrees with option 1: the Splitter should unwrap {{Exchange}} parts. The
CAMEL-9323 rationale was "use {{GroupedMessageAggregationStrategy}} instead",
i.e. the route author picks the shape. With batching consumers (camel-kafka
today, camel-sjms next per CAMEL-16039 / CAMEL-20368) the *component* picks
{{List<Exchange>}} and the user cannot opt into {{List<Message>}}. Falling back
to a {{.process()}} with an {{instanceof Exchange}} loop instead of the
Splitter EIP is not what we want to recommend.
A few things the fix should take into account:
# *It is a behaviour change with existing code relying on the old shape.*
{{KafkaConsumerAsyncManualCommitIT}} does
{{aggregate(groupedExchange()).split().body().process(e ->
e.getMessage().getBody(Exchange.class).getMessage().getHeader(MANUAL_COMMIT,
...))}}. After the change {{getBody(Exchange.class)}} returns null there. Since
this has been the behaviour for a decade, users will have copied that pattern,
so it needs an upgrade guide entry (4.23) and that IT updated (it gets simpler:
the header ends up directly on the child).
# *Copy the message rather than adopt it.* The {{Message}} branch does
{{setIn(message)}}, which re-parents the message. For {{Exchange}} parts use
{{newExchange.getIn().copyFrom(part.getMessage())}} instead. The batching
consumer keeps using the original exchanges after the route has run (offset
computation, manual commit), and with exchange pooling
{{DefaultPooledExchange.done()}} calls {{in.reset()}}, which would wipe the
Kafka record headers off the batch's exchanges as each child completes. Copying
avoids that.
# *Exchange properties on the part are not carried over* (same as the
{{Message}} branch today). Kafka manual commit and offsets are header based so
that is fine, but state it in the upgrade note.
# The Kafka "Batch Headers" example can then drop the variable workaround, and
the camel-sjms batching consumer should use the same {{List<Exchange>}} shape
so it is covered by the same fix.
Suggest: main only, no backport (behaviour change), plus a Splitter unit test
for {{List<Exchange>}} covering body and headers.
_Claude Code on behalf of davsclaus_
was (Author: davsclaus):
Reviewed against the current code on main. The report checks out: {{Splitter}}
only unwraps {{Message}} parts ({{newExchange.setIn(message)}}, from CAMEL-1597
/ CAMEL-9252); an {{Exchange}} part falls into the generic branch and becomes
the child body, with no headers. And the Kafka "Batch Headers" docs from
CAMEL-20428 do recommend {{.split(body())}} over that shape, so we currently
document a split the Splitter does not really support.
I agree option 1 makes sense now. The CAMEL-9323 rationale was "use
{{GroupedMessageAggregationStrategy}} instead", i.e. the route author picks the
shape. With batching consumers (camel-kafka today, camel-sjms next per
CAMEL-16039 / CAMEL-20368) the *component* picks {{List<Exchange>}} and the
user cannot opt into {{List<Message>}}. Falling back to a {{.process()}} with
an {{instanceof Exchange}} loop instead of the Splitter EIP is not what we want
to recommend.
A few things the fix should take into account:
# *It is a behaviour change with existing code relying on the old shape.*
{{KafkaConsumerAsyncManualCommitIT}} does
{{aggregate(groupedExchange()).split().body().process(e ->
e.getMessage().getBody(Exchange.class).getMessage().getHeader(MANUAL_COMMIT,
...))}}. After the change {{getBody(Exchange.class)}} returns null there. Since
this has been the behaviour for a decade, users will have copied that pattern,
so it needs an upgrade guide entry (4.23) and that IT updated (it gets simpler:
the header ends up directly on the child).
# *Copy the message rather than adopt it.* The {{Message}} branch does
{{setIn(message)}}, which re-parents the message. For {{Exchange}} parts use
{{newExchange.getIn().copyFrom(part.getMessage())}} instead. The batching
consumer keeps using the original exchanges after the route has run (offset
computation, manual commit), and with exchange pooling
{{DefaultPooledExchange.done()}} calls {{in.reset()}}, which would wipe the
Kafka record headers off the batch's exchanges as each child completes. Copying
avoids that.
# *Exchange properties on the part are not carried over* (same as the
{{Message}} branch today). Kafka manual commit and offsets are header based so
that is fine, but state it in the upgrade note.
# The Kafka "Batch Headers" example can then drop the variable workaround, and
the camel-sjms batching consumer should use the same {{List<Exchange>}} shape
so it is covered by the same fix.
Suggest: main only, no backport (behaviour change), plus a Splitter unit test
for {{List<Exchange>}} covering body and headers.
_Claude Code on behalf of davsclaus_
> camel-core - Splitter should unwrap Exchange parts like Message parts
> (List<Exchange> from batching consumers)
> --------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-24797
> URL: https://issues.apache.org/jira/browse/CAMEL-24797
> Project: Camel
> Issue Type: Improvement
> Components: camel-core, camel-kafka
> Reporter: Federico Mariani
> Priority: Major
>
> Splitting a body of type {{List<Exchange>}} (the shape produced by the
> camel-kafka batching consumer, {{batching=true}}, and by
> {{GroupedExchangeAggregationStrategy}}) creates child exchanges whose *body
> is the {{Exchange}} object itself* and whose *headers are empty*. Only
> {{Message}} parts are unwrapped by the Splitter today (the branch added for
> CAMEL-9252).
> Verified on 4.22.0 with a plain {{from("direct:a").split(body())}} route:
> ||Body shape||Child {{${body}}}||Child headers||
> |{{List<Exchange>}}|{{Exchange[]}}|none|
> |{{List<Message>}}|the record body|preserved|
> So {{.split(body()).to("kafka:out")}} after a batching consumer silently
> publishes {{Exchange[]}} as record value, and any filter/transform on the
> record headers sees nothing.
> *Why this matters now*: the Kafka batching docs added in 4.22 (CAMEL-20428,
> PR #25006, "Batch Headers" section) explicitly recommend {{.split(body())}}
> on the batch exchange, and work around the header loss by storing the topic
> in a variable before the split. The batching consumer proposed for camel-sjms
> (dev list thread, CAMEL-16039 / CAMEL-20368) will copy the same
> {{List<Exchange>}} shape, so the gap gets copied with it.
> CAMEL-9323 asked for this in 2015 and was closed as Won't Fix, pointing to
> {{List<Message>}} as the supported shape. That predates a core component
> producing {{List<Exchange>}} itself and documenting a split over it.
> *Proposed outcomes* (one of):
> # Splitter unwraps {{Exchange}} parts like it does {{Message}} parts: set the
> child's message from {{part.getMessage()}} (a few lines next to the existing
> {{instanceof Message}} branch in {{Splitter}}), plus an upgrade-guide note
> since {{getBody()}} on the child changes for that case.
> # Or: keep the Splitter as is, and change the Kafka batching docs to stop
> recommending {{split}} and show reading each child with
> {{getBody(Exchange.class)}} instead; future batching consumers document the
> same.
> _Claude Code on behalf of Federico Mariani_
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
