Bill Bejeck created KAFKA-21253:
-----------------------------------
Summary: 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: 4.2.2, 4.3.1, 4.1.2, 4.0.0, 3.9.2, 3.8.1, 4.4.0
Reporter: Bill Bejeck
Assignee: Bill Bejeck
Fix For: 4.5.0, 4.4.1
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
(v8.20.10#820010)
