slachiewicz commented on code in PR #314:
URL:
https://github.com/apache/flink-connector-kafka/pull/314#discussion_r4054034633
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/TransactionAbortStrategyImpl.java:
##########
@@ -143,6 +143,18 @@ public void abortTransactions(Context context) {
TransactionAborter transactionAborter =
context.getTransactionAborter();
for (String name : openTransactionsForSubtask) {
if (context.getPrecommittedTransactionalIds().contains(name)) {
+ if (context.isPrecommittedTransactionSuperseded(name)) {
+ // The broker holds a later transaction under this id
than the one the
+ // committer is about to commit. That commit will be
fenced, and nobody
+ // owns the open transaction; abort it so that it does
not pin the last
+ // stable offset until the transaction timeout.
+ LOG.warn(
+ "Aborting open transaction {}: the recovered
transaction under this id was superseded by a newer epoch",
+ name);
+ context.abandonPrecommittedTransaction(name);
Review Comment:
Yes, it could, and #322 catches exactly that: on the previous revision only
record 0 survived the restore. Fixed by keeping the id reserved; details in the
thread on `ProducerPoolImpl` below.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/ExactlyOnceKafkaWriterITCase.java:
##########
@@ -275,6 +275,61 @@ void shouldNotAbortPrecommittedTransactions(int
numCheckpointed) throws Exceptio
}
}
+ /**
+ * With {@code POOLING}, a committed transactional id is reused for a
later checkpoint under a
+ * newer epoch. A recovery from the earlier checkpoint still lists the id
as precommitted, and
+ * the committer's commit will be fenced. The open transaction under the
newer epoch has no
+ * owner and must be aborted on recovery, also when the prefix changed and
the id is never
+ * reused again (FLINK-40626).
+ */
+ @Test
+ void shouldAbortSupersededPrecommittedTransactionOnRecovery() throws
Exception {
+ final KafkaWriterState stateOfCheckpoint1;
+ final CheckpointTransaction precommitted;
+ try (final ExactlyOnceKafkaWriter<Integer> failedWriter =
+ createWriter(this::withPooling, createInitContext())) {
+ Tuple2<KafkaWriterState, KafkaCommittable> checkpoint1 =
+ onCheckpointBarrier(failedWriter, 1);
+ stateOfCheckpoint1 = checkpoint1.f0;
+ precommitted =
+
Iterables.getOnlyElement(stateOfCheckpoint1.getPrecommittedTransactionalIds());
+ assertThat(precommitted.hasKnownEpoch()).isTrue();
+
assertThat(precommitted.getEpoch()).isEqualTo(checkpoint1.f1.getEpoch());
+
+ // the committer commits checkpoint 1 and hands the id back to the
pool
+ checkpoint1.f1.getProducer().get().commitTransaction();
+ try (WritableBackchannel<TransactionFinished> backchannel =
+ getBackchannel(failedWriter)) {
+
backchannel.send(TransactionFinished.successful(precommitted.getTransactionalId()));
+ }
+ onCheckpointBarrier(failedWriter, 2);
+ // checkpoint 3 reuses the id of checkpoint 1 under a bumped epoch
+ KafkaCommittable checkpoint3 = onCheckpointBarrier(failedWriter,
3).f1;
+ assertThat(checkpoint3.getTransactionalId())
+ .isEqualTo(precommitted.getTransactionalId());
+
assertThat(checkpoint3.getEpoch()).isGreaterThan(precommitted.getEpoch());
+ // the job fails here; the transactions of checkpoints 2 and 3
linger on the broker
+ }
+
+ try (AdminClient admin =
AdminClient.create(getKafkaClientConfiguration())) {
+ assertThat(AdminUtils.getOpenTransactionsForTopics(admin,
Collections.singleton(topic)))
+ .hasSize(2);
+
+ // recovery from checkpoint 1; the new writer gets a new prefix,
so the old id is never
Review Comment:
Changed: the test now restores with the same prefix, which is what makes
POOLING contend for the reserved id, and it asserts that the first new
transaction does not take it and that the fenced report leaves the new
committable intact. The restored committer end to end is
`KafkaSinkRecoveryITCase` from #322.
--
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]
