[
https://issues.apache.org/jira/browse/CAMEL-26599?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-26599.
---------------------------------
Resolution: Fixed
Merged: https://github.com/apache/camel/pull/29288 (main)
_Claude Code on behalf of davsclaus_
> camel-aws2-kinesis - KCL consumer checkpoints past a record whose exchange
> failed (no at-least-once option)
> -----------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-26599
> URL: https://issues.apache.org/jira/browse/CAMEL-26599
> Project: Camel
> Issue Type: Bug
> Components: camel-aws2-kinesis
> Reporter: shashank
> Assignee: shashank
> Priority: Major
> Fix For: 4.24.0
>
>
> Found with a TLA+ model of the KCL consumer, compared with traces of the real
> record processor ({{KclKinesis2Consumer.CamelKinesisRecordProcessor}} behind
> a real route, driven by an in-JVM stand-in of the KCL shard consumer), then
> reproduced on the real classes.
> With {{useKclConsumers=true}}, a record whose exchange fails is checkpointed
> and never consumed again. CAMEL-24191 meant to stop this ("avoid
> checkpointing past a failed batch"), and the code says so ("Checkpoint only
> after the whole batch has been processed successfully, so that a processing
> failure leaves the records to be redelivered from the last checkpoint"), but
> it does not hold, for two reasons.
> *1. A failed route does not throw.* {{processRecords}} calls
> {{processor.process(createExchange(record, shardId))}} for each record and
> then {{checkpointer.checkpoint()}}; only an exception thrown out of the loop
> skips the checkpoint. The processor of a route returns normally when the
> route fails and sets the exception on the exchange, so the catch block is not
> reached and the batch is checkpointed right after the failure. The unit test
> of CAMEL-24191 mocks a processor that throws, which a route does not do.
> *2. KCL does not deliver a batch again.* Even when the checkpoint is skipped,
> KCL goes on with the next batch while the consumer holds the lease of the
> shard (on a throw out of {{processRecords}} it logs "Skipping over the
> following data records"). {{checkpoint()}} checkpoints at the last record
> delivered so far ({{largestPermittedCheckpointValue}}, set before
> {{processRecords}} is called), so the next batch, or {{shutdownRequested}}
> when the route stops, checkpoints past the failed records. They are only read
> again from a checkpoint by the next holder of the lease, and by then the
> checkpoint is past them.
> {code:java}
> from("aws2-kinesis://mystream?useKclConsumers=true&asyncClient=true®ion=eu-west-1")
> .process(e -> { throw new IllegalStateException("downstream not
> available"); }) // fails for one record
> .to("mock:result");
> // the failed record is checkpointed with its batch and not consumed again
> after a restart
> {code}
> Model trace (one shard, 4 records, batches of 2): batch [0,1] delivered,
> record 0 fails (exception on the exchange), record 1 succeeds, checkpoint 2:
> record 0 is lost. Also: batch [0,1] fails, the checkpoint is skipped, batch
> [2,3] succeeds and checkpoints 4; or the route stops and
> {{shutdownRequested}} checkpoints 2. All of these were also traces of the
> real consumer.
> On main (a14dd5068f17) a unit test with a real route ({{processor.process}}
> of the route's consumer processor, a mocked {{RecordProcessorCheckpointer}})
> shows {{checkpoint()}} called after a failed record, in the failing batch and
> at shutdown.
> *Proposal*
> Keep today's default (a failed record is not consumed again, as with the
> plain {{Kinesis2Consumer}} and camel-kafka without {{breakOnFirstError}}),
> and add an opt-in endpoint option for at-least-once:
> * New option {{kclStopCheckpointOnFailure}} (boolean, default {{false}}).
> When enabled, once a record has failed this record processor does not move
> the checkpoint past it while it holds the lease: it checkpoints only the
> records before it ({{checkpoint(sequenceNumber, subSequenceNumber)}} of the
> last good record), and does not checkpoint in later batches or in
> {{shutdownRequested}}. The records from the failed one on are consumed again
> when the lease is taken again (the route is started again, or the lease moves
> to another worker). One WARN with the sequence number of the first failed
> record. In the model with the option no record below the checkpoint was left
> unprocessed, and a failed record is replayed after a restart.
> * Cost of the option, in its documentation: the records after the failed one
> are still routed by the current lease holder, so they can be routed twice and
> out of order (the route must be idempotent); only the WARN with the sequence
> number of the first failed record shows that the checkpoint stays behind (the
> lag metrics do not); a record that fails each time keeps the checkpoint of
> its shard before it, so each new lease holder (a restart, or a KCL 3 lease
> handoff) routes it and everything after it again, up to the stream retention.
> At the end of a shard KCL needs the end checkpointed before it goes on with
> the child shards, so {{shardEnded}} still checkpoints, with a WARN.
> * With the option off (default): an exchange that ends with an exception is
> reported to the exception handler of the consumer (as the pubsub, pulsar and
> plain Kinesis consumers do) and the batch is checkpointed as today; the
> component docs say that a failed record is not consumed again unless the
> route handles it (dead letter channel, redelivery) or the option is on.
> * Either way, catch KCL's retryable checkpoint exceptions
> ({{ThrottlingException}}, {{KinesisClientLibDependencyException}}) with a
> WARN: the next checkpoint covers the records.
> * The comment of CAMEL-24191 ("records to be redelivered from the last
> checkpoint") is not true for KCL and should say what happens.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
