Andrea Cosentino created CAMEL-24783:
----------------------------------------
Summary: camel-kafka - async batch producer: in-flight callbacks
can mutate a continued/recycled exchange after a mid-batch dispatch failure
Key: CAMEL-24783
URL: https://issues.apache.org/jira/browse/CAMEL-24783
Project: Camel
Issue Type: Bug
Components: camel-kafka
Reporter: Andrea Cosentino
h3. Problem
In the asynchronous batch/iterator producer path
({{KafkaProducer.processIterableAsync}} -> {{doSend}}), records are dispatched
one at a time. If a *later* element fails to dispatch -
{{kafkaProducer.send()}} can throw synchronously (buffer exhaustion /
{{max.block.ms}} timeout, serialization error, producer closed), or
{{KeyValueHolderIterator.next()}} can throw (bad
{{CamelKafkaOverrideTimestamp}} conversion, or header serialization when
{{batchWithIndividualHeaders=true}}) - the catch in {{process()}} does
{{exchange.setException(e); callback.done(true);}} and routing continues.
However, the records already dispatched have in-flight Kafka callbacks. When
those later complete on the Kafka sender thread,
{{KafkaProducerCallBack.onCompletion}} runs {{setException(exchange, ...)}} and
{{recordMetadataList.add(...)}} on an exchange that has *already continued*
down the route (and, with exchange pooling, may have been reset/reused). This
is a data race / use-after-continue that can corrupt headers or exception state
on a continued or recycled exchange. (The completion counter never reaches 0
because {{allSent()}} is skipped on the failure path, so there is no double
{{done()}} - but the mutation race remains.)
h3. Proposed direction
On a mid-batch dispatch failure, do not continue routing until the
already-dispatched sends have completed (or otherwise prevent late callbacks
from mutating a continued/recycled exchange). Needs careful handling of the
completion counter and exchange pooling, plus a failure-injection test.
Design-first; filing to track. Related: CAMEL-24779 (single-message path),
CAMEL-24780.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
