[FLINK-34554] Adding pooling name strategy

Reduce the load on Kafka broker by reusing transaction ids as soon as they have been committed.

Author: AHeise

flink-connector-kafka-e2e-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/KafkaSinkE2ECase.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.tests.util.kafka;
 
+import org.apache.flink.connector.kafka.sink.TransactionNamingStrategy;
 import org.apache.flink.connector.kafka.sink.testutils.KafkaSinkExternalContextFactory;
 import org.apache.flink.connector.kafka.testutils.DockerImageVersions;
 import org.apache.flink.connector.testframe.container.FlinkContainerTestEnvironment;
@@ -62,7 +63,7 @@ public class KafkaSinkE2ECase extends SinkTestSuiteBase<String> {
     // Defines 2 External context Factories, so test cases will be invoked twice using these two
     // kinds of external contexts.
     @TestContext
-    KafkaSinkExternalContextFactory contextFactory =
+    KafkaSinkExternalContextFactory incrementing =
             new KafkaSinkExternalContextFactory(
                     kafka.getContainer(),
                     Arrays.asList(
@@ -77,7 +78,27 @@ public class KafkaSinkE2ECase extends SinkTestSuiteBase<String> {
                             ResourceTestUtils.getResource("flink-connector-testing.jar")
                                     .toAbsolutePath()
                                     .toUri()
-                                    .toURL()));
+                                    .toURL()),
+                    TransactionNamingStrategy.INCREMENTING);
+
+    @TestContext
+    KafkaSinkExternalContextFactory pooling =
+            new KafkaSinkExternalContextFactory(
+                    kafka.getContainer(),
+                    Arrays.asList(
+                            ResourceTestUtils.getResource("kafka-connector.jar")
+                                    .toAbsolutePath()
+                                    .toUri()
+                                    .toURL(),
+                            ResourceTestUtils.getResource("kafka-clients.jar")
+                                    .toAbsolutePath()
+                                    .toUri()
+                                    .toURL(),
+                            ResourceTestUtils.getResource("flink-connector-testing.jar")
+                                    .toAbsolutePath()
+                                    .toUri()
+                                    .toURL()),
+                    TransactionNamingStrategy.POOLING);
 
     public KafkaSinkE2ECase() throws Exception {}
 }

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/ExactlyOnceKafkaWriter.java

@@ -200,6 +200,10 @@ public void initialize() {
 
     private FlinkKafkaInternalProducer<byte[], byte[]> startTransaction(long checkpointId) {
         namingContext.setNextCheckpointId(checkpointId);
+        namingContext.setOngoingTransactions(
+                producerPool.getOngoingTransactions().stream()
+                        .map(CheckpointTransaction::getTransactionalId)
+                        .collect(Collectors.toSet()));
         FlinkKafkaInternalProducer<byte[], byte[]> producer =
                 transactionNamingStrategy.getTransactionalProducer(namingContext);
         namingContext.setLastCheckpointId(checkpointId);

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java

@@ -56,6 +56,7 @@ class KafkaCommitter implements Committer<KafkaCommittable>, Closeable {
                     + "To avoid data loss, the application will restart.";
 
     private final Properties kafkaProducerConfig;
+    private final boolean reusesTransactionalIds;
     private final BiFunction<Properties, String, FlinkKafkaInternalProducer<?, ?>> producerFactory;
     private final WritableBackchannel<TransactionFinished> backchannel;
     @Nullable private FlinkKafkaInternalProducer<?, ?> committingProducer;
@@ -65,8 +66,10 @@ class KafkaCommitter implements Committer<KafkaCommittable>, Closeable {
             String transactionalIdPrefix,
             int subtaskId,
             int attemptNumber,
+            boolean reusesTransactionalIds,
             BiFunction<Properties, String, FlinkKafkaInternalProducer<?, ?>> producerFactory) {
         this.kafkaProducerConfig = kafkaProducerConfig;
+        this.reusesTransactionalIds = reusesTransactionalIds;
         this.producerFactory = producerFactory;
         backchannel =
                 BackchannelFactory.getInstance()
@@ -102,19 +105,7 @@ public void commit(Collection<CommitRequest<KafkaCommittable>> requests)
                         "Encountered retriable exception while committing {}.", transactionalId, e);
                 request.retryLater();
             } catch (ProducerFencedException e) {
-                // initTransaction has been called on this transaction before
-                LOG.error(
-                        "Unable to commit transaction ({}) because its producer is already fenced."
-                                + " This means that you either have a different producer with the same '{}' (this is"
-                                + " unlikely with the '{}' as all generated ids are unique and shouldn't be reused)"
-                                + " or recovery took longer than '{}' ({}ms). In both cases this most likely signals data loss,"
-                                + " please consult the Flink documentation for more details.",
-                        request,
-                        ProducerConfig.TRANSACTIONAL_ID_CONFIG,
-                        KafkaSink.class.getSimpleName(),
-                        ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,
-                        kafkaProducerConfig.getProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG),
-                        e);
+                logFencedRequest(request, e);
                 handleFailedTransaction(producer);
                 request.signalFailedWithKnownReason(e);
             } catch (InvalidTxnStateException e) {
@@ -146,6 +137,40 @@ public void commit(Collection<CommitRequest<KafkaCommittable>> requests)
         }
     }
 
+    private void logFencedRequest(
+            CommitRequest<KafkaCommittable> request, ProducerFencedException e) {
+        if (reusesTransactionalIds) {
+            // If checkpoint 1 succeeds, checkpoint 2 is aborted, and checkpoint 3 may reuse the id
+            // of checkpoint 1. A recovery of checkpoint 1 would show that the transaction has been
+            // fenced.
+            LOG.warn(
+                    "Unable to commit transaction ({}) because its producer is already fenced."
+                            + " If this warning appears as part of the recovery of a checkpoint, it is expected in some cases (e.g., aborted checkpoints in previous attempt)."
+                            + " If it's outside of recovery, this means that you either have a different sink with the same '{}'"
+                            + " or recovery took longer than '{}' ({}ms). In both cases this most likely signals data loss,"
+                            + " please consult the Flink documentation for more details.",
+                    request,
+                    ProducerConfig.TRANSACTIONAL_ID_CONFIG,
+                    ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,
+                    kafkaProducerConfig.getProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG),
+                    e);
+        } else {
+            // initTransaction has been called on this transaction before
+            LOG.error(
+                    "Unable to commit transaction ({}) because its producer is already fenced."
+                            + " This means that you either have a different producer with the same '{}' (this is"
+                            + " unlikely with the '{}' as all generated ids are unique and shouldn't be reused)"
+                            + " or recovery took longer than '{}' ({}ms). In both cases this most likely signals data loss,"
+                            + " please consult the Flink documentation for more details.",
+                    request,
+                    ProducerConfig.TRANSACTIONAL_ID_CONFIG,
+                    KafkaSink.class.getSimpleName(),
+                    ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,
+                    kafkaProducerConfig.getProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG),
+                    e);
+        }
+    }
+
     private void handleFailedTransaction(FlinkKafkaInternalProducer<?, ?> producer) {
         if (producer == null) {
             return;

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSink.java

@@ -115,6 +115,7 @@ public Committer<KafkaCommittable> createCommitter(CommitterInitContext context)
                 transactionalIdPrefix,
                 context.getTaskInfo().getIndexOfThisSubtask(),
                 context.getTaskInfo().getAttemptNumber(),
+                transactionNamingStrategy == TransactionNamingStrategy.POOLING,
                 FlinkKafkaInternalProducer::new);
     }
 

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilder.java

@@ -21,6 +21,7 @@
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.java.ClosureCleaner;
 import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.kafka.lineage.KafkaDatasetFacetProvider;
 
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
@@ -200,6 +201,12 @@ private void sanityCheck() {
             checkState(
                     transactionalIdPrefix != null,
                     "EXACTLY_ONCE delivery guarantee requires a transactionalIdPrefix to be set to provide unique transaction names across multiple KafkaSinks writing to the same Kafka cluster.");
+            if (transactionNamingStrategy.getImpl().requiresKnownTopics()) {
+                checkState(
+                        recordSerializer instanceof KafkaDatasetFacetProvider,
+                        "For %s naming strategy, the recordSerializer needs to expose the target topics though implementing KafkaDatasetFacetProvider.",
+                        transactionNamingStrategy);
+            }
         }
     }
 

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/TransactionNamingStrategy.java

@@ -39,8 +39,24 @@ public enum TransactionNamingStrategy {
      * on the broker.
      *
      * <p>This is exactly the same behavior as in flink-connector-kafka 3.X.
+     *
+     * <p>Switching to this strategy from {@link #POOLING} is not supported.
+     */
+    INCREMENTING(TransactionNamingStrategyImpl.INCREMENTING, TransactionAbortStrategyImpl.PROBING),
+
+    /**
+     * This strategy reuses transaction names. It is more resource-friendly than {@link
+     * #INCREMENTING} on the Kafka broker.
+     *
+     * <p>It's a new strategy introduced in flink-connector-kafka 4.X. It requires Kafka 3.0+ and
+     * additional read permissions on the target topics.
+     *
+     * <p>The recommended way to switch to this strategy is to first take a checkpoint with
+     * flink-connector-kafka 4.X and then switch to this strategy. This will ensure that no
+     * transactions are left open from the previous run. Alternatively, you can use a savepoint from
+     * any version.
      */
-    INCREMENTING(TransactionNamingStrategyImpl.INCREMENTING, TransactionAbortStrategyImpl.PROBING);
+    POOLING(TransactionNamingStrategyImpl.POOLING, TransactionAbortStrategyImpl.LISTING);
 
     /**
      * The default transaction naming strategy. Currently set to {@link #INCREMENTING}, which is the

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/TransactionNamingStrategyContextImpl.java

@@ -20,13 +20,17 @@
 
 import org.apache.flink.annotation.Internal;
 
+import java.util.Collection;
+import java.util.Set;
+
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /** Implementation of {@link TransactionNamingStrategyImpl.Context}. */
 @Internal
 public class TransactionNamingStrategyContextImpl implements TransactionNamingStrategyImpl.Context {
     private final String transactionalIdPrefix;
     private final int subtaskId;
+    private Set<String> ongoingTransactions;
     private final ProducerPool producerPool;
     private long lastCheckpointId;
     private long nextCheckpointId;
@@ -63,6 +67,15 @@ public void setLastCheckpointId(long lastCheckpointId) {
         this.lastCheckpointId = lastCheckpointId;
     }
 
+    @Override
+    public Set<String> getOngoingTransactions() {
+        return ongoingTransactions;
+    }
+
+    public void setOngoingTransactions(Collection<String> ongoingTransactions) {
+        this.ongoingTransactions = Set.copyOf(ongoingTransactions);
+    }
+
     @Override
     public long getLastCheckpointId() {
         return lastCheckpointId;

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/TransactionNamingStrategyImpl.java

@@ -21,12 +21,14 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.connector.kafka.sink.TransactionNamingStrategy;
 
+import java.util.Set;
+
 import static org.apache.flink.util.Preconditions.checkState;
 
 /** Implementation of {@link TransactionNamingStrategy}. */
 @Internal
 public enum TransactionNamingStrategyImpl {
-    INCREMENTING(TransactionOwnership.IMPLICIT_BY_SUBTASK_ID) {
+    INCREMENTING(TransactionOwnership.IMPLICIT_BY_SUBTASK_ID, false) {
         /**
          * For each checkpoint we create new {@link FlinkKafkaInternalProducer} so that new
          * transactions will not clash with transactions created during previous checkpoints ({@code
@@ -55,12 +57,36 @@ public FlinkKafkaInternalProducer<byte[], byte[]> getTransactionalProducer(
             }
             return context.getProducer(context.buildTransactionalId(expectedCheckpointId));
         }
+    },
+    POOLING(TransactionOwnership.EXPLICIT_BY_WRITER_STATE, true) {
+        @Override
+        public FlinkKafkaInternalProducer<byte[], byte[]> getTransactionalProducer(
+                Context context) {
+            Set<String> usedTransactionalIds = context.getOngoingTransactions();
+            for (int offset = 0; ; offset++) {
+                String transactionalIdCandidate = context.buildTransactionalId(offset);
+                if (usedTransactionalIds.contains(transactionalIdCandidate)) {
+                    continue;
+                }
+                return context.getProducer(transactionalIdCandidate);
+            }
+        }
     };
 
     private final TransactionOwnership ownership;
+    private final boolean requiresKnownTopics;
 
-    TransactionNamingStrategyImpl(TransactionOwnership ownership) {
+    TransactionNamingStrategyImpl(TransactionOwnership ownership, boolean requiresKnownTopics) {
         this.ownership = ownership;
+        this.requiresKnownTopics = requiresKnownTopics;
+    }
+
+    public boolean requiresKnownTopics() {
+        return requiresKnownTopics;
+    }
+
+    public TransactionOwnership getOwnership() {
+        return ownership;
     }
 
     /**
@@ -70,16 +96,14 @@ public FlinkKafkaInternalProducer<byte[], byte[]> getTransactionalProducer(
     public abstract FlinkKafkaInternalProducer<byte[], byte[]> getTransactionalProducer(
             Context context);
 
-    public TransactionOwnership getOwnership() {
-        return ownership;
-    }
-
     /** Context for the transaction naming strategy. */
     public interface Context {
         String buildTransactionalId(long offset);
 
         long getNextCheckpointId();
 
+        Set<String> getOngoingTransactions();
+
         long getLastCheckpointId();
 
         FlinkKafkaInternalProducer<byte[], byte[]> getProducer(String transactionalId);

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/TransactionalIdFactory.java

@@ -44,11 +44,11 @@ public static String buildTransactionalId(
                 + checkpointOffset;
     }
 
-    public static long extractSubtaskId(String name) {
+    public static int extractSubtaskId(String name) {
         int lastSep = name.lastIndexOf("-");
         int secondLastSep = name.lastIndexOf("-", lastSep - 1);
         String subtaskString = name.substring(secondLastSep + 1, lastSep);
-        return Long.parseLong(subtaskString);
+        return Integer.parseInt(subtaskString);
     }
 
     public static String extractPrefix(String name) {