shashank created CAMEL-26599:
--------------------------------
Summary: 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
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)
