Savonitar commented on code in PR #314:
URL:
https://github.com/apache/flink-connector-kafka/pull/314#discussion_r4039525893
##########
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:
I think here we have a tricky issue:
With unchanged prefix (the common/default case), this method frees the ID
and `POOLING` reuses it immediately, however the restored committer's fenced
commit still can send notification `TransactionFinished.erroneously (ID)` and
`recycleByTransactionId` that matches by transactional id.
Can that stale notification land on the newly reused transaction? If yes,
it means in that case we will have a dataloss.
To improve the robustness of this repository, I raised a separate
**independent** PR with an additional test coverage
https://github.com/apache/flink-connector-kafka/pull/322
It is a general test (not dedicated for this PR/ticket). But it catches this
issue too.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/TransactionAbortStrategyContextImpl.java:
##########
@@ -111,7 +136,50 @@ public Set<String> getPrefixesToAbort() {
@Override
public Set<String> getPrecommittedTransactionalIds() {
- return precommittedTransactionIds;
+ return Collections.unmodifiableSet(precommittedTransactions.keySet());
+ }
+
+ @Override
+ public boolean isPrecommittedTransactionSuperseded(String transactionalId)
{
Review Comment:
Is this method tested only by IT case tests?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/TransactionAbortStrategyContextImpl.java:
##########
@@ -111,7 +136,50 @@ public Set<String> getPrefixesToAbort() {
@Override
public Set<String> getPrecommittedTransactionalIds() {
- return precommittedTransactionIds;
+ return Collections.unmodifiableSet(precommittedTransactions.keySet());
+ }
+
+ @Override
+ public boolean isPrecommittedTransactionSuperseded(String transactionalId)
{
+ CheckpointTransaction transaction =
precommittedTransactions.get(transactionalId);
+ if (transaction == null || !transaction.hasKnownEpoch()) {
+ // state written before v3 did not record the epoch; keep the old
behavior
+ return false;
+ }
+ TransactionDescription description =
describePrecommitted().get(transactionalId);
+ if (description == null) {
+ // the broker does not know the id any more; nothing to abort
+ return false;
+ }
+ boolean superseded =
+ description.producerId() != transaction.getProducerId()
+ || description.producerEpoch() >
transaction.getEpoch();
+ if (superseded) {
+ LOG.info(
+ "Recovered transaction {} was opened with producer id {}
and epoch {}, but the broker now holds producer id {} and epoch {} in state {}",
+ transactionalId,
+ transaction.getProducerId(),
+ transaction.getEpoch(),
+ description.producerId(),
+ description.producerEpoch(),
+ description.state());
+ }
+ return superseded;
+ }
+
+ @Override
+ public void abandonPrecommittedTransaction(String transactionalId) {
+ precommittedTransactions.remove(transactionalId);
+ precommittedTransactionAbandoner.accept(transactionalId);
+ }
+
+ private Map<String, TransactionDescription> describePrecommitted() {
+ if (precommittedDescriptions == null) {
+ precommittedDescriptions =
+ AdminUtils.describeTransactions(
Review Comment:
just to confirm, the frequency of calling this method is not high?
(according to PR description it is once). Is it once per subtask restart?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/TransactionAbortStrategyContextImpl.java:
##########
@@ -111,7 +136,50 @@ public Set<String> getPrefixesToAbort() {
@Override
public Set<String> getPrecommittedTransactionalIds() {
- return precommittedTransactionIds;
+ return Collections.unmodifiableSet(precommittedTransactions.keySet());
+ }
+
+ @Override
+ public boolean isPrecommittedTransactionSuperseded(String transactionalId)
{
+ CheckpointTransaction transaction =
precommittedTransactions.get(transactionalId);
+ if (transaction == null || !transaction.hasKnownEpoch()) {
+ // state written before v3 did not record the epoch; keep the old
behavior
+ return false;
+ }
+ TransactionDescription description =
describePrecommitted().get(transactionalId);
+ if (description == null) {
+ // the broker does not know the id any more; nothing to abort
+ return false;
+ }
+ boolean superseded =
+ description.producerId() != transaction.getProducerId()
+ || description.producerEpoch() >
transaction.getEpoch();
+ if (superseded) {
+ LOG.info(
+ "Recovered transaction {} was opened with producer id {}
and epoch {}, but the broker now holds producer id {} and epoch {} in state {}",
Review Comment:
Should we use "the broker reported" instead of "the broker now holds" here
and avoid calling the transaction "open" in the subsequent warning?
The `ONGOING` filter applies to the earlier `listTransactions()` result,
while `describeTransactions()` is a separate call whose results are cached. The
coordinator could abort the transaction between these observations or before
cleanup, so **neither log guarantees** it is still **open**.
Or am I missing something?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/ProducerPool.java:
##########
@@ -40,9 +40,23 @@ public interface ProducerPool extends AutoCloseable {
FlinkKafkaInternalProducer<byte[], byte[]> getTransactionalProducer(
String transactionalId, long checkpointId);
- /** Returns a snapshot of all ongoing transactions. */
+ /**
+ * Returns a snapshot of all ongoing transactions. Transactions opened by
this pool carry the
+ * producer id and epoch of their producer; transactions restored from
state keep whatever the
+ * state recorded.
+ */
Collection<CheckpointTransaction> getOngoingTransactions();
+ /**
+ * Stops tracking a transaction that was restored from state without
touching a producer. Used
+ * on recovery when the broker no longer holds the restored transaction
under its transactional
+ * id because the id was reused for a later checkpoint, so that the id can
be aborted and
+ * reused.
Review Comment:
> reused
Maybe this comment should be clarified a bit after you check
https://github.com/apache/flink-connector-kafka/pull/314/changes#r4039525893
(because then it can not be reused ASAP).
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriterStateSerializer.java:
##########
@@ -31,11 +32,17 @@
import static org.apache.flink.connector.kafka.sink.KafkaWriterState.UNKNOWN;
-/** A serializer used to serialize {@link KafkaWriterState}. */
+/**
+ * A serializer used to serialize {@link KafkaWriterState}.
+ *
+ * <p>Version 3 adds the producer id and epoch of every precommitted
transaction, so that a recovery
Review Comment:
Is a downgrade/rollback path considered or we expect forward-only for this
bump? (I see PR says "backwards compatible on read").
##########
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:
> the new writer gets a new prefix,
Do we test only changed prefix here? So, the reused prefix is never
excerciced. Should we add a same-prefix test case with the restored commiter?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/ProducerPoolImpl.java:
##########
@@ -254,7 +255,39 @@ public FlinkKafkaInternalProducer<byte[], byte[]>
getTransactionalProducer(
@Override
public Collection<CheckpointTransaction> getOngoingTransactions() {
- return new ArrayList<>(transactionalIdsByCheckpoint.keySet());
+ List<CheckpointTransaction> ongoing = new
ArrayList<>(transactionalIdsByCheckpoint.size());
+ for (Map.Entry<CheckpointTransaction, String> entry :
+ transactionalIdsByCheckpoint.entrySet()) {
+ CheckpointTransaction transaction = entry.getKey();
+ ProducerEntry producerEntry =
producerByTransactionalId.get(entry.getValue());
+ FlinkKafkaInternalProducer<byte[], byte[]> producer =
+ producerEntry == null ? null : producerEntry.getProducer();
+ if (producer == null) {
+ // restored from state; keep what the state knows
+ ongoing.add(transaction);
+ } else {
+ ongoing.add(
+ new CheckpointTransaction(
+ transaction.getTransactionalId(),
+ transaction.getCheckpointId(),
+ producer.getProducerId(),
+ producer.getEpoch()));
+ }
+ }
+ return ongoing;
+ }
+
+ @Override
+ public void abandonTransaction(String transactionalId) {
+ ProducerEntry producerEntry =
producerByTransactionalId.get(transactionalId);
+ checkState(
+ producerEntry != null && producerEntry.getProducer() == null,
+ "Transaction %s is not a restored transaction without a
producer: %s",
+ transactionalId,
+ producerEntry);
+ producerByTransactionalId.remove(transactionalId);
Review Comment:
could we keep the "superseeded" transactionalId reserved after abandoning?
Removing this entry allows `POOLING` to reuse it during writer initialization,
meanwhile a completion notification for the old transaction may still be
queued.
##########
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(
Review Comment:
we have this WARN log and INFO log from [public boolean
isPrecommittedTransactionSuperseded(String transactionalId)
{](https://github.com/apache/flink-connector-kafka/pull/314/changes#diff-2b2784802b7feb204128c1f72e9523037a202f417cb74ff3698cd2fe6ce45017R143)
> LOG.info(
"Recovered transaction {} was opened with producer id {}
and epoch {}, but the broker now holds producer id {} and epoch {} in state {}",
do we need to keep both?
--
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]
