[
https://issues.apache.org/jira/browse/KAFKA-21253?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Bill Bejeck updated KAFKA-21253:
--------------------------------
Fix Version/s: 4.2.3
> Failed transaction abort in RecordCollectorImpl.closeDirty() skips clearing
> changelog offsets, so a revived task can checkpoint a stale offset
> ----------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-21253
> URL: https://issues.apache.org/jira/browse/KAFKA-21253
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 3.8.1, 3.9.2, 4.0.0, 4.1.2, 4.3.1, 4.4.0, 4.2.2
> Reporter: Bill Bejeck
> Assignee: Bill Bejeck
> Priority: Critical
> Fix For: 4.5.0, 4.4.1, 4.3.3, 4.2.3
>
>
> With exactly-once enabled, a commitTransaction() call that times out
> (max.block.ms) leaves the producer with a pending commit that was never
> acknowledged. Any later abortTransaction() on that producer throws
> IllegalStateException: "Cannot attempt operation abortTransaction because the
> previous call to commitTransaction timed out and must be retried".
> RecordCollectorImpl.closeDirty() calls streamsProducer.abortTransaction()
> before close(). StreamsProducer.abortTransaction() doesn't catch
> IllegalStateException, so the exception escapes closeDirty(). As a result,
> close() is skipped and the collector's offsets map is not cleared.
> StreamTask.close() swallows the exception in
> TaskManager.executeAndMaybeSwallow(), which logs only at DEBUG and doesn't
> include the exception.
> A task revived after the dirty close reuses the same record collector. It
> restores its wiped store from the start of the changelog, while
> recordCollector.offsets() still holds the last changelog offset written
> before the failure. If the task is closed clean before restoration finishes,
> that stale offset is written to the checkpoint. On the next initialization,
> restoration starts from that checkpoint and skips the missing records, which
> leaves the store empty or incomplete.
> Trigger path:
> 1. A commit TimeoutException raises TaskCorruptedException.
> 2. TaskManager.handleCorruption() commits the open transaction before closing
> the corrupted tasks.
> 3. If that EndTxn times out and the commit contains no task offsets, the
> timeout is logged but not rethrown.
> 4. closeDirtyAndRevive() then calls closeDirty() on a producer that still has
> the pending commit.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
