shashank created CAMEL-25756:
--------------------------------
Summary: camel-kafka - batching consumer loses the records of an
incomplete batch when a poll fails (e.g. a record that cannot be deserialized,
pollOnError=ERROR_HANDLER): they are never routed and the next batch commits
past them
Key: CAMEL-25756
URL: https://issues.apache.org/jira/browse/CAMEL-25756
Project: Camel
Issue Type: Bug
Components: camel-kafka
Reporter: shashank
Assignee: shashank
With {{batching=true}} the Kafka consumer keeps the records of several polls in
{{KafkaRecordBatchingProcessor.exchangeList}} until the batch is complete
({{maxPollRecords}} records), the poll timeout expires or
{{batchingIntervalMs}} passes (since CAMEL-20380). The processor lives in the
record processor facade, and {{KafkaFetchRecords.startPolling()}} (main
ddca15435112, line 333) creates a new facade every time it is entered:
{code:java}
final KafkaRecordProcessorFacade recordProcessorFacade =
createRecordProcessor();
{code}
{{startPolling()}} is entered again, on the same connected Kafka consumer,
after every exception of the polling loop that the poll exception strategy
continues on (line 396 and the do-while at line 174): a failed {{poll}} with
the default {{pollOnError=ERROR_HANDLER}}, and with {{DISCARD}} or {{RETRY}}.
The new processor starts with an empty batch, so the records held for the
incomplete batch are dropped. The Kafka consumer does not return them again
(its position is already past them), and the next complete batch commits its
highest offset per partition, which is past them. Once a later batch with
records of the same partition is committed, they are never redelivered, even
after a restart or a rebalance. Nothing is logged about them. (If no later
batch of that partition is committed before the consumer stops or the partition
is revoked, they are polled again from the committed offset: redelivery, not
loss. With {{allowManualCommit}} the commit past them is the route's own
{{KafkaManualCommit}} of a later batch.)
A realistic trigger: a record that cannot be deserialized by a deserializer
that can fail (Avro, JSON schema, a custom one; the default
{{StringDeserializer}} does not fail). {{poll}} throws
{{RecordDeserializationException}}, the default error handler skips that one
record ({{SeekUtil}}, CAMEL-23994), and the good records polled before it, on
every partition, are lost with it. Up to {{maxPollRecords - 1}} records are
lost per failed poll. Other poll exceptions the strategy continues on (for
example a {{KafkaException}} for a corrupt record batch) drop the held records
the same way. An authentication or authorization error ends the consumer with
{{ERROR_HANDLER}}; nothing is committed past the held records then.
Example ({{batching=true&maxPollRecords=10}}, one partition): a poll returns
offsets 0-2 (held, the batch is not complete); the next poll fails on the
record at offset 3; the error handler seeks to 4; the next poll returns 4-13;
the batch 4-13 is routed and offset 14 is committed. Offsets 0-2 are never
routed.
*Fix* (one class): {{KafkaFetchRecords}} creates the record processor facade
when the Kafka consumer is connected (and reconnected) and keeps it while the
consumer is connected, so a failed poll keeps the held records and they are
routed and committed with the next batch. After a reconnect a new Kafka
consumer starts from the committed offsets, so a new facade (and an empty
batch) is still created then. The streaming facade holds no state between
polls; it is unaffected.
*Tests*: new {{KafkaBatchingConsumerPollErrorTest}} (2 tests, a real route and
Kafka's {{MockConsumer}}; {{batching=true&maxPollRecords=10}}): three records
are polled and held, a poll fails with a {{KafkaException}} (first test) or a
{{RecordDeserializationException}} for offset 3 (second test), then ten more
records arrive. Both fail on main ({{expected: <[a0, a1, a2, b0, b1, b2, b3,
b4, b5, b6]> but was: <[b0, b1, b2, b3, b4, b5, b6, b7, b8, b9]>}}) and pass
with the fix; the camel-kafka unit tests pass (222 run, 0 failures, 1 skipped).
*How it was found*: deterministic simulation testing with a fake Kafka world
(the camel-kafka consumer, its fetcher loop, commit managers and rebalance
listener real; a fake {{Consumer}} written from the KafkaConsumer Javadoc of
kafka-clients 4.3, bound through {{kafkaClientFactory}}). The seed picked the
configuration (streaming or batching, {{autoCommitEnable}}, manual commit with
the sync / async / no-op commit manager, {{breakOnFirstError}},
{{maxPollRecords}} 1 to 50, {{pollTimeoutMs}}, {{batchingIntervalMs}}), poison
records, failed polls and commits, partial fetches, route restarts and
suspends. The invariant "every record sent reaches the route" failed in 103 of
1000 seeds on main, all with {{batching=true}} and a failed poll (first failing
seed 7); with the fix, 0 of these 1000 seeds fail it with {{batching=true}}.
*Affected*: main, camel-4.23.x, camel-4.22.x, camel-4.18.x and camel-4.14.x
(same {{createRecordProcessor()}} call in {{startPolling()}} and the same batch
kept across polls; read, not run). On camel-4.18.x and camel-4.14.x the error
handler's seek is the one before CAMEL-23994, the held records are dropped the
same way.
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
