slachiewicz commented on code in PR #314:
URL:
https://github.com/apache/flink-connector-kafka/pull/314#discussion_r3981860534
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/KafkaCommitter.java:
##########
@@ -108,6 +109,9 @@ public void
commit(Collection<CommitRequest<KafkaCommittable>> requests)
request.retryLater();
} catch (ProducerFencedException e) {
logFencedRequest(request, e);
+ if (reusesTransactionalIds) {
+ abortNewerTransaction(transactionalId);
Review Comment:
You are right, and the window is not closable from here. The orphan and the
zombie look identical to the broker: same producer id, epoch plus one, ONGOING.
In the orphan case `getTransactionalProducer` re-initialised the id for a later
checkpoint; in your case the new attempt's committer committed at the old
epoch, the backchannel freed the id, pooling reused it and `initTransactions`
bumped it. So `describeTransactions` in the committer cannot prove anything
either.
Moved the abort to where it is provable:
`ExactlyOnceKafkaWriter.initialize()` at recovery, inside
`TransactionAbortStrategyImpl.LISTING`, before the new attempt holds any id.
There, 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 dead attempt or to a zombie, and the newest attempt is the side
that is supposed to fence. This also covers the changed-prefix case from
FLINK-40585, which the committer-side bump would not have.
Cost: the writer state did not carry the epoch, so `CheckpointTransaction`
now records producer id and epoch and `KafkaWriterStateSerializer` is at v3; v2
state reads back with unknown epoch and keeps the previous skip.
`describeTransactions` needs broker 3.0+ like `listTransactions` that LISTING
already uses, and `Describe` on the transactional id, which `Write` implies.
Regression guard is a changed-prefix recovery in
`ExactlyOnceKafkaWriterITCase` asserting no ONGOING transaction survives; it
fails without the strategy change. Trailer and Release Note added. The inline
threads below are outdated by the move; I have replied on each.
*This comment was created with AI assistance.*
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/KafkaCommitter.java:
##########
@@ -184,6 +188,33 @@ private void logFencedRequest(
}
}
+ /**
+ * With reused transactional ids, a fenced commit during recovery usually
means that the
+ * transaction was committed before the failure and its id was recycled
for a later checkpoint,
+ * whose transaction is still open on the broker under a newer epoch.
Nobody owns that
+ * transaction any more: the writer skips the id as precommitted and this
committer cannot
+ * commit it. Bumping the epoch aborts it, so that it does not block
read_committed consumers
+ * until the transaction timeout expires.
+ */
+ private void abortNewerTransaction(String transactionalId) {
+ FlinkKafkaInternalProducer<?, ?> producer =
+ producerFactory.apply(kafkaProducerConfig, transactionalId);
+ try {
+ producer.initTransactions();
+ LOG.info(
+ "Aborted open transaction of a newer epoch under {} after
its commit was fenced.",
+ transactionalId);
+ } catch (KafkaException e) {
+ LOG.warn(
+ "Could not abort the open transaction under {}; it stays
open until '{}' expires.",
+ transactionalId,
+ ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,
+ e);
+ } finally {
+ producer.close();
+ }
+ }
Review Comment:
Superseded: the committer-side bump is gone in 05b8847f; the abort now
happens at recovery in `TransactionAbortStrategyImpl.LISTING` (see the thread
on line 113), so this code no longer exists.
*This comment was created with AI assistance.*
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/KafkaCommitter.java:
##########
@@ -184,6 +188,33 @@ private void logFencedRequest(
}
}
+ /**
+ * With reused transactional ids, a fenced commit during recovery usually
means that the
+ * transaction was committed before the failure and its id was recycled
for a later checkpoint,
+ * whose transaction is still open on the broker under a newer epoch.
Nobody owns that
+ * transaction any more: the writer skips the id as precommitted and this
committer cannot
+ * commit it. Bumping the epoch aborts it, so that it does not block
read_committed consumers
+ * until the transaction timeout expires.
+ */
+ private void abortNewerTransaction(String transactionalId) {
+ FlinkKafkaInternalProducer<?, ?> producer =
+ producerFactory.apply(kafkaProducerConfig, transactionalId);
+ try {
+ producer.initTransactions();
Review Comment:
Superseded: the committer-side bump is gone in 05b8847f; the abort now
happens at recovery in `TransactionAbortStrategyImpl.LISTING` (see the thread
on line 113), so this code no longer exists.
*This comment was created with AI assistance.*
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/KafkaCommitter.java:
##########
@@ -184,6 +188,33 @@ private void logFencedRequest(
}
}
+ /**
+ * With reused transactional ids, a fenced commit during recovery usually
means that the
+ * transaction was committed before the failure and its id was recycled
for a later checkpoint,
+ * whose transaction is still open on the broker under a newer epoch.
Nobody owns that
+ * transaction any more: the writer skips the id as precommitted and this
committer cannot
+ * commit it. Bumping the epoch aborts it, so that it does not block
read_committed consumers
+ * until the transaction timeout expires.
+ */
+ private void abortNewerTransaction(String transactionalId) {
+ FlinkKafkaInternalProducer<?, ?> producer =
+ producerFactory.apply(kafkaProducerConfig, transactionalId);
+ try {
+ producer.initTransactions();
+ LOG.info(
+ "Aborted open transaction of a newer epoch under {} after
its commit was fenced.",
+ transactionalId);
+ } catch (KafkaException e) {
Review Comment:
Superseded: the committer-side bump is gone in 05b8847f; the abort now
happens at recovery in `TransactionAbortStrategyImpl.LISTING` (see the thread
on line 113), so this code no longer exists.
*This comment was created with AI assistance.*
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/internal/KafkaCommitterTest.java:
##########
@@ -219,6 +221,49 @@ private AtomicBoolean interruptOnMessage(Thread
mainThread, ServerSocket serverS
return interrupting;
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testFencedCommitAbortsNewerTransactionOnlyIfIdsAreReused(
+ boolean reusesTransactionalIds) throws IOException,
InterruptedException {
+ Properties properties = getProperties();
+ List<MockProducer> createdProducers = new ArrayList<>();
+ BiFunction<Properties, String, FlinkKafkaInternalProducer<?, ?>>
fencingFactory =
+ (props, transactionalId) -> {
+ MockProducer producer =
+ new MockProducer(props, new
ProducerFencedException("test"));
+ createdProducers.add(producer);
+ return producer;
+ };
+ try (final KafkaCommitter committer =
+ new KafkaCommitter(
+ properties,
+ TRANS_ID,
+ SUB_ID,
+ ATTEMPT,
+ reusesTransactionalIds,
+ fencingFactory);
+ ReadableBackchannel<TransactionFinished> backchannel =
+ BackchannelFactory.getInstance()
+ .getReadableBackchannel(SUB_ID, ATTEMPT,
TRANS_ID)) {
+ // committable restored from state: the committer resumes it with
its own producer
+ final MockCommitRequest<KafkaCommittable> request =
+ new MockCommitRequest<>(
+ new KafkaCommittable(PRODUCER_ID, EPOCH, TRANS_ID,
null));
+ committer.commit(Collections.singletonList(request));
+
+ assertThat(backchannel).has(transactionFinished(false));
+ assertThat(committer.getCommittingProducer()).isNull();
+
assertThat(createdProducers).allMatch(FlinkKafkaInternalProducer::isClosed);
+ // with reused ids the fenced id may hold a newer, orphaned
transaction; bumping the
+ // epoch through initTransactions aborts it
+ long abortedIds =
+ createdProducers.stream()
+ .filter(MockProducer::isTransactionsInitialized)
+ .count();
+ assertThat(abortedIds).isEqualTo(reusesTransactionalIds ? 1 : 0);
Review Comment:
Superseded: that test went with the committer change in 05b8847f. The guard
is now
`ExactlyOnceKafkaWriterITCase#shouldAbortSupersededPrecommittedTransactionOnRecovery`
against the broker: checkpoint 1 committed, its id reused for checkpoint 3
under a bumped epoch, recovery from checkpoint 1 with a new prefix, then
`AdminUtils.getOpenTransactionsForTopics` must be empty. Without the strategy
change it fails with the reused id still ONGOING.
*This comment was created with AI assistance.*
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
