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