[
https://issues.apache.org/jira/browse/FLINK-40626?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Sylwester Lachiewicz updated FLINK-40626:
-----------------------------------------
Description:
With {{TransactionNamingStrategy.POOLING}}, a committed transactional id is
recycled for a later checkpoint; {{ProducerPoolImpl.getTransactionalProducer}}
calls {{initTransactions()}} on reuse, which bumps the epoch. If the job then
fails and recovers from the earlier checkpoint, the committer re-commits the id
at the old epoch and gets {{ProducerFencedException}}. That is the documented
benign case (the transaction was already committed), but the newer transaction
the id was recycled for is still ONGOING on the broker, and nobody aborts it:
{{TransactionAbortStrategyImpl.LISTING}} skips the id because it is in the
recovered precommitted set
([TransactionAbortStrategyImpl.java:145|https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/TransactionAbortStrategyImpl.java#L145]),
and the committer drops the request
([KafkaCommitter.java:109|https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/KafkaCommitter.java#L109]).
The open transaction pins the last stable offset, so {{read_committed}}
consumers see nothing the recovered job writes. With an unchanged prefix and
parallelism the stall is short: {{POOLING}} takes the id back at the next
checkpoint and {{initTransactions}} aborts it. The transaction survives until
{{transaction.timeout.ms}} whenever the id is never recycled again: after a
downscale, because new transactions only use the first owned subtask id and an
orphan under a secondary owned subtask id is never reused, and after a change
of {{transactionalIdPrefix}} (supported; the writer logs it and adds the old
prefix to the abort list). The downscale case needs no user error.
Observed in CI as
[FLINK-40585|https://issues.apache.org/jira/browse/FLINK-40585] (run
30181683657, {{rescaleListing[3->5]}}): job 3 lists {{secondPrefix-0-0}} as
open, skips it as precommitted, the committer's re-commit at epoch 11 is
fenced, and the drain stops after job 2's records.
Fix ([PR #314|https://github.com/apache/flink-connector-kafka/pull/314]): abort
the transaction at recovery, in {{TransactionAbortStrategyImpl.LISTING}}, which
is the one point where it is provably ownerless. Before the new attempt holds
any id, an open transaction under a precommitted id whose broker epoch is newer
than the one the writer snapshotted (or whose producer id differs) can only
belong to the failed attempt or to a zombie, and the newest attempt is the side
entitled to fence. A committer-side epoch bump on the fenced commit was
considered first and rejected: a fenced commit cannot tell that orphan from a
live owner of the reused id (both show epoch plus one, ONGOING), so a late
commit from an unreachable-but-alive task manager would have fenced the healthy
attempt.
To make the comparison possible, {{CheckpointTransaction}} records producer id
and epoch and {{KafkaWriterStateSerializer}} moves to version 3; version 2
state reads back with unknown epoch and keeps the previous skip. Recovery calls
{{Admin.describeTransactions}} (broker 3.0+, {{Describe}} on the transactional
id, implied by {{Write}}). Backports: 5.0 and 4.0; {{POOLING}} is not on 3.4.
was:
With {{TransactionNamingStrategy.POOLING}}, a committed transactional id is
recycled for a later checkpoint; {{ProducerPoolImpl.getTransactionalProducer}}
calls {{initTransactions()}} on reuse, which bumps the epoch. If the job then
fails and recovers from the earlier checkpoint, the committer re-commits the id
at the old epoch and gets {{ProducerFencedException}}. That is the documented
benign case (the transaction was already committed), but the newer transaction
the id was recycled for is still ONGOING on the broker, and nobody aborts it:
{{TransactionAbortStrategyImpl.LISTING}} skips the id because it is in the
recovered precommitted set
([TransactionAbortStrategyImpl.java:145|https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/TransactionAbortStrategyImpl.java#L145]),
and the committer drops the request
([KafkaCommitter.java:109|https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/KafkaCommitter.java#L109]).
The open transaction pins the last stable offset until
{{transaction.timeout.ms}} (1 h by default), so {{read_committed}} consumers
see nothing the recovered job writes. With an unchanged prefix the writer
eventually reuses the id and the epoch bump aborts it (a stall of about two
checkpoint intervals). With a changed {{transactionalIdPrefix}} (supported; the
writer logs it and adds the old prefix to the abort list) the id is never
touched again and the stall lasts until the timeout.
Observed in CI as
[FLINK-40585|https://issues.apache.org/jira/browse/FLINK-40585] (run
30181683657, {{rescaleListing[3->5]}}): job 3 lists {{secondPrefix-0-0}} as
open, skips it as precommitted, the committer's re-commit at epoch 11 is
fenced, and the drain stops after job 2's records.
Proposed fix: on a fenced commit with reused ids, the committer bumps the epoch
of that id ({{initTransactions}} on a short-lived producer) before dropping the
request, which aborts the orphaned transaction.
> KafkaCommitter leaves the newer open transaction under a reused transactional
> id after a fenced commit
> ------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40626
> URL: https://issues.apache.org/jira/browse/FLINK-40626
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Affects Versions: kafka-4.0.1, kafka-5.0.0, kafka-5.1.0
> Reporter: Sylwester Lachiewicz
> Assignee: Sylwester Lachiewicz
> Priority: Major
> Labels: pull-request-available
>
> With {{TransactionNamingStrategy.POOLING}}, a committed transactional id is
> recycled for a later checkpoint;
> {{ProducerPoolImpl.getTransactionalProducer}} calls {{initTransactions()}} on
> reuse, which bumps the epoch. If the job then fails and recovers from the
> earlier checkpoint, the committer re-commits the id at the old epoch and gets
> {{ProducerFencedException}}. That is the documented benign case (the
> transaction was already committed), but the newer transaction the id was
> recycled for is still ONGOING on the broker, and nobody aborts it:
> {{TransactionAbortStrategyImpl.LISTING}} skips the id because it is in the
> recovered precommitted set
> ([TransactionAbortStrategyImpl.java:145|https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/TransactionAbortStrategyImpl.java#L145]),
> and the committer drops the request
> ([KafkaCommitter.java:109|https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/KafkaCommitter.java#L109]).
> The open transaction pins the last stable offset, so {{read_committed}}
> consumers see nothing the recovered job writes. With an unchanged prefix and
> parallelism the stall is short: {{POOLING}} takes the id back at the next
> checkpoint and {{initTransactions}} aborts it. The transaction survives until
> {{transaction.timeout.ms}} whenever the id is never recycled again: after a
> downscale, because new transactions only use the first owned subtask id and
> an orphan under a secondary owned subtask id is never reused, and after a
> change of {{transactionalIdPrefix}} (supported; the writer logs it and adds
> the old prefix to the abort list). The downscale case needs no user error.
> Observed in CI as
> [FLINK-40585|https://issues.apache.org/jira/browse/FLINK-40585] (run
> 30181683657, {{rescaleListing[3->5]}}): job 3 lists {{secondPrefix-0-0}} as
> open, skips it as precommitted, the committer's re-commit at epoch 11 is
> fenced, and the drain stops after job 2's records.
> Fix ([PR #314|https://github.com/apache/flink-connector-kafka/pull/314]):
> abort the transaction at recovery, in
> {{TransactionAbortStrategyImpl.LISTING}}, which is the one point where it is
> provably ownerless. Before the new attempt holds any id, an open transaction
> under a precommitted id whose broker epoch is newer than the one the writer
> snapshotted (or whose producer id differs) can only belong to the failed
> attempt or to a zombie, and the newest attempt is the side entitled to fence.
> A committer-side epoch bump on the fenced commit was considered first and
> rejected: a fenced commit cannot tell that orphan from a live owner of the
> reused id (both show epoch plus one, ONGOING), so a late commit from an
> unreachable-but-alive task manager would have fenced the healthy attempt.
> To make the comparison possible, {{CheckpointTransaction}} records producer
> id and epoch and {{KafkaWriterStateSerializer}} moves to version 3; version 2
> state reads back with unknown epoch and keeps the previous skip. Recovery
> calls {{Admin.describeTransactions}} (broker 3.0+, {{Describe}} on the
> transactional id, implied by {{Write}}). Backports: 5.0 and 4.0; {{POOLING}}
> is not on 3.4.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
