[jira] [Updated] (CAMEL-25756) 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

["Claus Ibsen (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CAMEL-25756?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Claus Ibsen updated CAMEL-25756:
--------------------------------
    Fix Version/s: 4.24.0

> 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)