Federico Mariani created CAMEL-23994:
----------------------------------------
Summary: camel-kafka: consumer offset-handling bugs causing silent
message loss (resume adapter, pollOnError seek, batching commit, offset
repository)
Key: CAMEL-23994
URL: https://issues.apache.org/jira/browse/CAMEL-23994
Project: Camel
Issue Type: Bug
Components: camel-kafka
Reporter: Federico Mariani
Code review of camel-kafka found several independent bugs in the consumer
offset/commit handling that all lead to *silent message loss*. Filing them
together since they are in the same area and interact. All line numbers refer
to current {{main}} (4.22.0-SNAPSHOT).
h3. 1. Resume-state restore is dead code: inverted key-format check in
KafkaResumeAdapter
{{org.apache.camel.component.kafka.consumer.support.resume.KafkaResumeAdapter#deserialize}}
(lines 59-75):
{code:java}
final String[] keyParts = key.split("/");
if (keyParts == null || keyParts.length != 2) { // inverted!
String topic = keyParts[0];
int partition = Integer.parseInt(keyParts[1]);
...
resumeCache.add(new TopicPartition(topic, partition), offset);
} else {
LOG.warn("Unable to deserialize key '{}' because it has in invalid format
and it will be discarded", key);
}
{code}
The condition is inverted: every *valid* key ({{topic/partition}}, 2 parts) is
discarded with an "invalid format" warning, and an actually-invalid key (no
{{/}}) enters the parse branch and throws {{ArrayIndexOutOfBoundsException}}.
({{String.split}} also never returns {{null}}.)
Effect: consumers configured with a {{resumeStrategy}} never restore persisted
offsets on restart — {{resumeCache}} stays empty, {{resume()}} seeks nowhere,
and the consumer silently restarts from the group / {{auto.offset.reset}}
position (duplicates or skipped work). Present since CAMEL-18688
(59046fe140ff); the log messages prove the intent is the opposite of the code.
Fix: {{if (keyParts.length == 2) \{ parse & cache \} else \{ warn \}}}.
h3. 2. pollOnError=DISCARD/ERROR_HANDLER skips one unread record on *every*
assigned partition
{{KafkaFetchRecords#startPolling}} (line ~437) handles any exception from the
poll loop with a hard-coded offset:
{code:java}
// why do we set this to -1
long partitionLastOffset = -1;
pollExceptionStrategy.handle(partitionLastOffset, e);
{code}
Because the offset is always -1, {{SeekUtil.seekToNextOffset}} always takes its
second branch, which seeks {{position(tp) + 1}} for *all* assigned partitions.
Two problems:
* Only (at most) one partition contained the failing record; for every other
partition {{position()}} points at an unread, valid record — seeking {{+1}}
silently drops one good message per partition.
* The caught exception need not be record-related at all (e.g. a transient
{{KafkaException}} from a commit inside the loop) — records are still skipped.
Also, the first branch of {{seekToNextOffset}} ({{partitionLastOffset != -1}})
is unreachable from the only caller, and would be wrong anyway (it seeks every
partition to the same offset taken from one partition).
Fix suggestion: when the exception is a {{RecordDeserializationException}},
seek only {{e.topicPartition()}} to {{e.offset() + 1}}; for other exceptions do
not advance any position.
h3. 3. Batching mode: auto-commit commits offsets of records that are still
buffered/unprocessed
{{KafkaRecordBatchingProcessor}} buffers records across polls (since
CAMEL-20380), but batch completion commits through
{{AsyncCommitManager.commit()}} → no-arg {{consumer.commitAsync()}}, which
commits the consumer's *current position* (end of the last poll) for all
partitions — including records still sitting unprocessed in {{exchangeList}} or
not yet drained from the current poll.
Scenario ({{batching=true}}, {{maxPollRecords=100}}): poll #1 returns 50
records → buffered. Poll #2 returns 100 (position now 150). After adding 50 of
them the batch of 100 is dispatched and committed → *offset 150 is committed*
although records 100-149 are unprocessed. A crash, rebalance, or even graceful
shutdown (buffered exchanges are discarded on stop) permanently loses them. The
same over-commit happens on every {{batchingIntervalMs}} flush, because
{{hasExpiredRecords()}} dispatches before the just-polled records are added.
Introduced by the interaction of CAMEL-19241 (auto-commit for batching) and
CAMEL-20380 (cross-poll buffering).
Fix: track the max offset per {{TopicPartition}} actually contained in the
dispatched batch and commit explicitly via {{commitAsync(Map<TopicPartition,
OffsetAndMetadata>, callback)}}.
h3. 4. Off-by-one when Sync/Async commit managers write to the offsetRepository
The offset-repository contract (see
{{OffsetPartitionAssignmentAdapter#resumeFromOffset}}: "the state contains the
last read offset, so seek from the next one", i.e. seek {{state+1}}) is honored
by {{CommitToOffsetManager}} (writes the raw record offset), but:
* {{SyncCommitManager}} (lines 70-79) writes {{offset + 1}}
* {{AsyncCommitManager#postCommitCallback}} (lines 93-99) writes the committed
offset ({{OffsetAndMetadata.offset()}} = record offset + 1)
Round trip: record at offset 41 processed → repo stores 42 → on restart the
adapter seeks {{42 + 1 = 43}} → the record at offset 42 is never delivered. One
message lost per partition per restart. Introduced by CAMEL-18717 (its ITs
verify the repo write but not the restart/seek round-trip).
Fix: save the record offset ({{partitionLastOffset}}) in both managers,
matching {{CommitToOffsetManager}} semantics.
h3. 5. Batching + DefaultKafkaManualAsyncCommitFactory: manual.commit() never
reaches Kafka
{{DefaultKafkaManualAsyncCommit}} only calls
{{commitManager.recordOffset(...)}} into the {{OffsetCache}}; the cache is
flushed to Kafka only by {{AsyncCommitManager.commit(partition)}}. The
*streaming* facade calls that at the end of each partition batch — the
*batching* facade ({{KafkaRecordBatchingProcessorFacade}}) never does.
Scenario:
{{batching=true&allowManualCommit=true&kafkaManualCommitFactory=#class:...DefaultKafkaManualAsyncCommitFactory&autoCommitEnable=false}}
— the route calls {{manual.commit()}} for every batch, but offsets reach Kafka
only on graceful partition revocation. On crash, the group offset is still at
the initial position → the entire history since startup is redelivered.
Fix: the batching facade should flush recorded offsets after each
{{processPolledRecords}} (call {{commit(partition)}} for the partitions seen in
the poll), mirroring the streaming facade.
----
_This issue was researched and written by Claude Code on behalf of Federico
Mariani (GitHub: Croway). Findings were verified against the source and git
history before filing._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
