[
https://issues.apache.org/jira/browse/KAFKA-21298?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
sanghyeok An updated KAFKA-21298:
---------------------------------
Description:
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.
was:
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.
> 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
> Priority: Major
> Labels: client, streams
>
> 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)
