[
https://issues.apache.org/jira/browse/KAFKA-21298?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias J. Sax updated KAFKA-21298:
------------------------------------
Labels: (was: client streams)
> 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
> Components: streams
> Reporter: sanghyeok An
> Assignee: sanghyeok An
> Priority: Major
>
> 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.*
> ## 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)
