MartijnVisser commented on code in PR #314:
URL:
https://github.com/apache/flink-connector-kafka/pull/314#discussion_r3981061938
##########
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:
`org.apache.kafka.common.errors.InterruptException extends KafkaException`,
so a cancel arriving during the epoch bump is swallowed here and only surfaces
on the next request in the batch. `commit()` converts that exception into
`InterruptedException` a few lines up precisely because the producer is left
inconsistent.
Skip the abort when the thread is already interrupted.
##########
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:
Blocking, as a question. `ProducerFencedException` means someone bumped the
epoch under this id. This reads that as "our own recycled transaction, now
orphaned" and bumps it again — but the warning right below names two other
readings, and under either of those the id has a live owner.
I checked both directions against a broker. On `main`: the orphan stays
`Ongoing`, and a live owner of the same id commits fine. On this branch: the
orphan is aborted, and the live owner gets `ProducerFencedException: There is a
newer producer with the same transactionalId which fences the current one.` on
its commit.
The case I can't rule out is a task manager that has lost its heartbeat but
is still running: the job restarts elsewhere, the new attempt's writer takes
the pooled id over, and the old committer's `commit()` then arrives and is
fenced. Today that committer loses, which is what the fencing is for; with this
change it wins and fails the healthy attempt over.
Please either show that window is closed, or bump only where the transaction
is provably orphaned — the committable carries `(producerId, epoch)` and
`Admin.describeTransactions` returns the broker's current epoch.
##########
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:
`initTransactions()` blocks up to `max.block.ms`. `KafkaSinkBuilder` sets
`transaction.timeout.ms` but never `max.block.ms`, so the default 60s applies
per fenced request, serially through the request loop; a scale-down that
concentrates ids on one committer multiplies it. `KafkaCommitterTest` doesn't
see this because `getProperties()` pins it to 100.
Please copy `kafkaProducerConfig` and set a short `max.block.ms` for this
throwaway producer.
##########
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` is constructed as `super(properties,
KafkaCommitterTest.TRANS_ID)` (line 316), so it ignores the id the factory is
handed, and `initTransactions()` only sets a flag. This asserts that some
producer was re-inited — not that the right id was, and not that anything was
aborted; passing the wrong id to `producerFactory` would still pass.
Assert the id the factory captured, and add a broker-level case.
`ExactlyOnceKafkaWriterITCase` already has the container fixture and
`AdminUtils.getOpenTransactionsForTopics`, which is enough to assert no
`ONGOING` transaction survives.
--
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]
