[
https://issues.apache.org/jira/browse/CAMEL-23994?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen updated CAMEL-23994:
--------------------------------
Fix Version/s: 4.22.0
> 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
> Assignee: Claus Ibsen
> Priority: Major
> Fix For: 4.22.0
>
>
> 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)
