[
https://issues.apache.org/jira/browse/CAMEL-25756?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-25756.
---------------------------------
Resolution: Fixed
> 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
> Priority: Major
> Fix For: 4.24.0
>
>
> 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)
