sanghyeok An created KAFKA-21298:
------------------------------------
Summary: Pending offset resets are lost during task recreation,
causing missing processing results under exactly-once semantics
Key: KAFKA-21298
URL: https://issues.apache.org/jira/browse/KAFKA-21298
Project: Kafka
Issue Type: Bug
Reporter: sanghyeok An
Assignee: sanghyeok An
When a transaction is aborted during partial task revocation in exactly-once
mode, processing results can be lost if a retained task is recreated before its
pending offset reset is applied.
Records processed in the aborted transaction must be reprocessed from the last
committed offset. However, if the pending reset information is lost with the
old task instance, the replacement task continues from the consumer’s current
position. A subsequent commit can advance the committed offset past the skipped
records, preventing normal recovery from reprocessing them.
This violates {*}exactly-once semantics{*}, as some input records never have
their processing results committed.
The following scenario illustrates the issue:
# *Task A and Task B run on the same StreamThread.*
## {*}{*}Task B’s committed offset is {{{}100{}}}. It processes records at
offsets {{{}100–109{}}}, advancing the consumer position to {{{}110{}}}.
## The outputs for these records belong to a transaction that has not yet been
committed.
# *The transaction is aborted while only Task A is revoked.*
## {*}{*}A rebalance revokes Task A while retaining Task B. During revocation
handling, {{sendOffsetsToTransaction()}} times out, and the transaction is
aborted.
## Task B’s outputs for offsets {{100–109}} are therefore not committed. Task
B must reprocess those records starting from committed offset {{{}100{}}}.
## Task B is moved into the pending-initialization state, and the partitions
requiring an offset reset are recorded in the existing task instance. The
consumer position remains at {{{}110{}}}.
# *Another assignment is handled before Task B resets its offsets.*
## {*}{*}Assignment handling closes the existing Task B awaiting
initialization and creates a replacement with the same TaskId.
## If the pending reset information is not transferred to the replacement
task, it does not rewind the consumer to offset {{{}100{}}}. Instead, it
continues processing from offset {{{}110{}}}.
# *A subsequent transaction commits a higher offset.*
## {*}{*}The replacement Task B processes the record at offset {{110}} and
commits the next offset, {{{}111{}}}.
## The processing results for offsets {{100–109}} were discarded by the
earlier abort, and those records were never reprocessed. Since the committed
offset has advanced to {{{}111{}}}, subsequent restarts or reassignments do not
replay them.
The expected behavior is for the replacement task to resume from offset
{{{}100{}}}, reprocess the records from the aborted transaction, and commit
their results.
Instead, processing can continue past offsets {{{}100–109{}}}, permanently
omitting their committed processing results despite exactly-once mode being
enabled.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
