MartijnVisser commented on code in PR #314:
URL:
https://github.com/apache/flink-connector-kafka/pull/314#discussion_r3981061914
##########
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:
A fenced commit also means we might be the stale side — an
unreachable-but-alive TM whose committer arrives after the new attempt took the
id over. Then this bumps the epoch and fences the healthy attempt. I checked
both directions against a broker: on `main` a live owner of the id still
commits, here it gets `ProducerFencedException`. Can you show that window is
closed, or bump only when the transaction is provably orphaned?
##########
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:
`max.block.ms` isn't set by `KafkaSinkBuilder`, so this blocks up to the 60s
default per fenced request, serially. `KafkaCommitterTest` pins it to 100 so it
doesn't show. Please set a short one on a copy of the config.
##########
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:
`InterruptException extends KafkaException`, so a cancel here is swallowed
and only surfaces on the next request — `commit()` converts it a few lines up
for that reason. Skip the abort when the thread is already interrupted.
##########
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:
`MockProducer` ignores the id the factory is handed (`super(properties,
TRANS_ID)`, line 316) and `initTransactions()` just sets a flag, so this passes
even with the wrong id. Assert the captured id, and add a broker-level case —
`ExactlyOnceKafkaWriterITCase` has the fixtures.
--
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]
