[
https://issues.apache.org/jira/browse/CAMEL-26598?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-26598 started by shashank.
----------------------------------------
> camel-kafka - breakOnFirstError: a failed record is skipped when the consumer
> is paused in the same poll (e.g. by ThrottlingExceptionRoutePolicy) and its
> partition has no committed offset, also with exactlyOnce
> ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-26598
> URL: https://issues.apache.org/jira/browse/CAMEL-26598
> Project: Camel
> Issue Type: Bug
> Components: camel-kafka
> Reporter: shashank
> Assignee: shashank
> Priority: Major
>
> Found with a TLA+ model of the camel-kafka consumer whose traces were
> validated against the real consumer (Specula-style trace validation: the real
> {{KafkaFetchRecords}} driven by Kafka's {{MockConsumer}}, each trace replayed
> in the model), then reproduced on the real classes.
> With {{breakOnFirstError=true}}, a failed record is skipped for good when the
> consumer is paused in the same poll loop iteration as the break and its
> partition has no committed offset yet. This also happens with
> {{exactlyOnce=true}}, whose documentation says that
> {{breakOnFirstError=true}} "is required so a failed record is replayed from
> the last committed offset".
> *How*
> {{KafkaFetchRecords.startPolling()}} processes the polled records, then
> applies a pending pause or resume request ({{updateTaskState()}}), then
> handles the break:
> {code:java}
> updateTaskState();
> // when breakOnFirstError we want to unsubscribe from Kafka
> if (result != null && result.isBreakOnErrorHit() &&
> !this.state.get().equals(State.PAUSED)) {
> // force re-connect
> setReconnect(true);
> setConnected(false);
> }
> {code}
> When the pause was requested while the failed record was in the route,
> {{updateTaskState()}} pauses the consumer and the break is dropped: no new
> Kafka consumer, and the fetch position stays after the whole poll. On resume,
> {{updateTaskState()}} seeks only the partitions for which
> {{consumer.committed(...)}} returns an offset. A partition without one (a new
> group, a new partition, the first records a group consumes, or committed
> offsets that expired) is not seeked, so the consumer goes on after the poll:
> the failed record and the records after it in the same poll are never routed.
> The next successful record commits past them. This matters with
> {{autoOffsetReset=earliest}} (the setting that makes a new group read the
> records it has not committed): without a pause, the break creates a new Kafka
> consumer, which polls such a partition from the beginning again. With
> {{autoOffsetReset=latest}} (the default) a new Kafka consumer starts such a
> partition at its end, so a break without a pause skips the records as well;
> that is not changed here. The processing log still says "Will seek consumer
> to offset N on partition P and start polling again."
> The pause does not need a race. {{ThrottlingExceptionRoutePolicy}} suspends
> the consumer from {{onExchangeDone}} of the failed exchange, which runs on
> the fetcher thread while the record is processed, so with this policy the
> pause and the break always come in the same iteration.
> {code:java}
> from("kafka:in?groupId=new-group&autoOffsetReset=earliest&autoCommitEnable=false&allowManualCommit=true"
> +
> "&kafkaManualCommitFactory=#class:org.apache.camel.component.kafka.consumer.DefaultKafkaManualCommitFactory"
> + "&breakOnFirstError=true")
> .routePolicy(new ThrottlingExceptionRoutePolicy(1, 60000, 1000, null,
> false))
> .process(e -> { /* fails once on the first record */ })
> .process(e -> e.getMessage().getHeader(KafkaConstants.MANUAL_COMMIT,
> KafkaManualCommit.class).commit())
> .to("mock:result");
> // records a, b, c in a partition the group never committed: a fails, the
> circuit opens and half opens,
> // mock:result receives nothing; expected a, b, c
> {code}
> Model trace (exactly-once, 3 records, 2 per poll, a partition without a
> committed offset starts at 0): partition assigned with no committed offset,
> poll returns records 0 and 1, record 0 fails, the pause is applied, no
> reconnect, resume without a seek, the next poll returns record 2, its
> transaction commits offset 3: records 0 and 1 are lost. The same holds with
> the synchronous manual commit manager and with the default
> {{NoopCommitManager}}. With a committed offset, the resume seek replays the
> failed record (model and tests).
> On main (03ed6ff1a3a1) a MockConsumer test on a simulated log fails for the
> circuit breaker (synchronous manual commit and exactly-once) and for a pause
> requested while the failed record is processed: the records are never routed
> again. The same test without a pause, or with a committed offset on the
> partition, gets all records.
> *History*
> CAMEL-18588 added the {{!PAUSED}} condition: then a new Kafka consumer was
> not paused, so the break would have resumed a suspended route; CAMEL-18327
> added the seek to the committed offset on resume instead. CAMEL-18760
> (circuit breaker loses the message that opened it) was closed with the advice
> to use manual commit, which still loses it when the partition has no
> committed offset. Since CAMEL-25313 a new Kafka consumer of a suspended
> consumer is paused again (when it is created and when partitions are assigned
> to it), so the condition is no longer needed.
> *Proposal*
> Handle the break also when the consumer is paused: a new Kafka consumer is
> created, paused again (CAMEL-25313), and on resume polls from the committed
> offset, or from {{autoOffsetReset}} when there is none, as on a break without
> a pause. In the model with this change (a partition without a committed
> offset starts at 0, as with {{autoOffsetReset=earliest}}) no record is lost
> and exactly-once holds (exactly-once, synchronous and no-op commit managers,
> with and without a fenced zombie). The rebalance listener pauses the
> partitions of the new Kafka consumer; partitions assigned without a rebalance
> ({{assign()}} in a custom {{SubscribeAdapter}}) must be paused when the
> consumer is created too, or the suspended route routes the records again (and
> fails and reconnects again in a loop).
> Cost: the new consumer leaves and rejoins the group, which the group handles
> as a rebalance, once for each failed exchange that suspends the consumer,
> also when the partition has a committed offset (where the resume seek already
> replays the record today). Keeping the paused consumer and seeking instead
> would have to move back every partition with unprocessed records of the poll
> (the break also skips the records of the other partitions), and for a
> partition without a committed offset it would have to re-implement each
> {{autoOffsetReset}} strategy (earliest, latest, by_duration) per partition;
> the reconnect is the path a break without a pause already takes. Worth an
> upgrade-guide entry.
> Not covered: {{autoCommitEnable=true}}, where the Kafka client commits
> positions by itself.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
