[FLINK-37356] Recreate producer on error

Reusing a producer in erroneous state can result in further issues that are outside the planned state machine.

These known error cases can be found in KafkaCommitter and contain at this time:
* ProducerFencedException - when a commit cannot be completed independent on how often we retry.
* InvalidTxnStateException - was mostly caused by empty transactions in the past. Again we can just abandon this transactions.
* UnknownProducerIdException - usually happening after recovery for 7 days where the metadata about the transaction has been deleted.

Author: AHeise

flink-connector-kafka/archunit-violations/c0d94764-76a0-4c50-b617-70b1754c4612

@@ -24,11 +24,12 @@ Method <org.apache.flink.connector.kafka.dynamic.source.reader.DynamicKafkaSourc
 Method <org.apache.flink.connector.kafka.sink.ExactlyOnceKafkaWriter.getProducerPool()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (ExactlyOnceKafkaWriter.java:0)
 Method <org.apache.flink.connector.kafka.sink.ExactlyOnceKafkaWriter.getTransactionalIdPrefix()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (ExactlyOnceKafkaWriter.java:0)
 Method <org.apache.flink.connector.kafka.sink.KafkaCommitter.getBackchannel()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (KafkaCommitter.java:0)
-Method <org.apache.flink.connector.kafka.sink.KafkaSink.addPostCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.getCoLocationGroupKey()> in (KafkaSink.java:168)
-Method <org.apache.flink.connector.kafka.sink.KafkaSink.addPostCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.getInputs()> in (KafkaSink.java:171)
-Method <org.apache.flink.connector.kafka.sink.KafkaSink.addPostCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.getOutputType()> in (KafkaSink.java:167)
-Method <org.apache.flink.connector.kafka.sink.KafkaSink.addPostCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.setCoLocationGroupKey(java.lang.String)> in (KafkaSink.java:170)
-Method <org.apache.flink.connector.kafka.sink.KafkaSink.addPostCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> checks instanceof <org.apache.flink.streaming.api.connector.sink2.CommittableMessageTypeInfo> in (KafkaSink.java:167)
+Method <org.apache.flink.connector.kafka.sink.KafkaCommitter.getCommittingProducer()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (KafkaCommitter.java:0)
+Method <org.apache.flink.connector.kafka.sink.KafkaSink.addPostCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.getCoLocationGroupKey()> in (KafkaSink.java:170)
+Method <org.apache.flink.connector.kafka.sink.KafkaSink.addPostCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.getInputs()> in (KafkaSink.java:173)
+Method <org.apache.flink.connector.kafka.sink.KafkaSink.addPostCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.getOutputType()> in (KafkaSink.java:169)
+Method <org.apache.flink.connector.kafka.sink.KafkaSink.addPostCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.setCoLocationGroupKey(java.lang.String)> in (KafkaSink.java:172)
+Method <org.apache.flink.connector.kafka.sink.KafkaSink.addPostCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> checks instanceof <org.apache.flink.streaming.api.connector.sink2.CommittableMessageTypeInfo> in (KafkaSink.java:169)
 Method <org.apache.flink.connector.kafka.sink.KafkaSink.addPostCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> has generic parameter type <org.apache.flink.streaming.api.datastream.DataStream<org.apache.flink.streaming.api.connector.sink2.CommittableMessage<org.apache.flink.connector.kafka.sink.KafkaCommittable>>> with type argument depending on <org.apache.flink.streaming.api.connector.sink2.CommittableMessage> in (KafkaSink.java:0)
 Method <org.apache.flink.connector.kafka.sink.KafkaSink.getKafkaProducerConfig()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (KafkaSink.java:0)
 Method <org.apache.flink.connector.kafka.sink.KafkaSinkBuilder.setRecordSerializer(org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema)> calls method <org.apache.flink.api.java.ClosureCleaner.clean(java.lang.Object, org.apache.flink.api.common.ExecutionConfig$ClosureCleanerLevel, boolean)> in (KafkaSinkBuilder.java:152)
@@ -51,4 +52,4 @@ Method <org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptionsU
 Method <org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptionsUtil.createValueFormatProjection(org.apache.flink.configuration.ReadableConfig, org.apache.flink.table.types.DataType)> calls method <org.apache.flink.table.types.logical.utils.LogicalTypeChecks.getFieldCount(org.apache.flink.table.types.logical.LogicalType)> in (KafkaConnectorOptionsUtil.java:569)
 Method <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSink.createSerialization(org.apache.flink.table.connector.sink.DynamicTableSink$Context, org.apache.flink.table.connector.format.EncodingFormat, [I, java.lang.String)> calls method <org.apache.flink.table.types.utils.DataTypeUtils.stripRowPrefix(org.apache.flink.table.types.DataType, java.lang.String)> in (KafkaDynamicSink.java:401)
 Method <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSink.getFieldGetters(java.util.List, [I)> has return type <[Lorg.apache.flink.table.data.RowData$FieldGetter;> in (KafkaDynamicSink.java:0)
-Method <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSource.createDeserialization(org.apache.flink.table.connector.source.DynamicTableSource$Context, org.apache.flink.table.connector.format.DecodingFormat, [I, java.lang.String)> calls method <org.apache.flink.table.types.utils.DataTypeUtils.stripRowPrefix(org.apache.flink.table.types.DataType, java.lang.String)> in (KafkaDynamicSource.java:566)
+Method <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSource.createDeserialization(org.apache.flink.table.connector.source.DynamicTableSource$Context, org.apache.flink.table.connector.format.DecodingFormat, [I, java.lang.String)> calls method <org.apache.flink.table.types.utils.DataTypeUtils.stripRowPrefix(org.apache.flink.table.types.DataType, java.lang.String)> in (KafkaDynamicSource.java:580)

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

@@ -26,6 +26,7 @@
 import org.apache.flink.connector.kafka.sink.internal.ProducerPool;
 import org.apache.flink.connector.kafka.sink.internal.ProducerPoolImpl;
 import org.apache.flink.connector.kafka.sink.internal.ReadableBackchannel;
+import org.apache.flink.connector.kafka.sink.internal.TransactionFinished;
 import org.apache.flink.connector.kafka.sink.internal.TransactionalIdFactory;
 import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
 import org.apache.flink.util.FlinkRuntimeException;
@@ -70,7 +71,7 @@ class ExactlyOnceKafkaWriter<IN> extends KafkaWriter<IN> {
      * Establishing the channel happens during recovery. Thus, it is only safe to poll in checkpoint
      * related methods.
      */
-    private final ReadableBackchannel<String> backchannel;
+    private final ReadableBackchannel<TransactionFinished> backchannel;
 
     /**
      * Constructor creating a kafka writer.
@@ -164,9 +165,10 @@ public Collection<KafkaCommittable> prepareCommit() {
     @Override
     public List<KafkaWriterState> snapshotState(long checkpointId) throws IOException {
         // recycle committed producers
-        String finishedTransactionalId;
-        while ((finishedTransactionalId = backchannel.poll()) != null) {
-            producerPool.recycleByTransactionId(finishedTransactionalId);
+        TransactionFinished finishedTransaction;
+        while ((finishedTransaction = backchannel.poll()) != null) {
+            producerPool.recycleByTransactionId(
+                    finishedTransaction.getTransactionId(), finishedTransaction.isSuccess());
         }
         currentProducer = startTransaction(checkpointId + 1);
         return Collections.singletonList(kafkaWriterState);

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

@@ -21,6 +21,7 @@
 import org.apache.flink.api.connector.sink2.Committer;
 import org.apache.flink.connector.kafka.sink.internal.BackchannelFactory;
 import org.apache.flink.connector.kafka.sink.internal.FlinkKafkaInternalProducer;
+import org.apache.flink.connector.kafka.sink.internal.TransactionFinished;
 import org.apache.flink.connector.kafka.sink.internal.WritableBackchannel;
 import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.IOUtils;
@@ -40,6 +41,7 @@
 import java.util.Collection;
 import java.util.Optional;
 import java.util.Properties;
+import java.util.function.BiFunction;
 
 /**
  * Committer implementation for {@link KafkaSink}
@@ -54,25 +56,34 @@ class KafkaCommitter implements Committer<KafkaCommittable>, Closeable {
                     + "To avoid data loss, the application will restart.";
 
     private final Properties kafkaProducerConfig;
-    private final WritableBackchannel<String> backchannel;
+    private final BiFunction<Properties, String, FlinkKafkaInternalProducer<?, ?>> producerFactory;
+    private final WritableBackchannel<TransactionFinished> backchannel;
     @Nullable private FlinkKafkaInternalProducer<?, ?> committingProducer;
 
     KafkaCommitter(
             Properties kafkaProducerConfig,
             String transactionalIdPrefix,
             int subtaskId,
-            int attemptNumber) {
+            int attemptNumber,
+            BiFunction<Properties, String, FlinkKafkaInternalProducer<?, ?>> producerFactory) {
         this.kafkaProducerConfig = kafkaProducerConfig;
+        this.producerFactory = producerFactory;
         backchannel =
                 BackchannelFactory.getInstance()
                         .getWritableBackchannel(subtaskId, attemptNumber, transactionalIdPrefix);
     }
 
     @VisibleForTesting
-    public WritableBackchannel<String> getBackchannel() {
+    public WritableBackchannel<TransactionFinished> getBackchannel() {
         return backchannel;
     }
 
+    @Nullable
+    @VisibleForTesting
+    FlinkKafkaInternalProducer<?, ?> getCommittingProducer() {
+        return committingProducer;
+    }
+
     @Override
     public void commit(Collection<CommitRequest<KafkaCommittable>> requests)
             throws IOException, InterruptedException {
@@ -81,11 +92,11 @@ public void commit(Collection<CommitRequest<KafkaCommittable>> requests)
             final String transactionalId = committable.getTransactionalId();
             LOG.debug("Committing Kafka transaction {}", transactionalId);
             Optional<FlinkKafkaInternalProducer<?, ?>> writerProducer = committable.getProducer();
-            FlinkKafkaInternalProducer<?, ?> producer;
+            FlinkKafkaInternalProducer<?, ?> producer = null;
             try {
                 producer = writerProducer.orElseGet(() -> getProducer(committable));
                 producer.commitTransaction();
-                backchannel.send(committable.getTransactionalId());
+                backchannel.send(TransactionFinished.successful(committable.getTransactionalId()));
             } catch (RetriableException e) {
                 LOG.warn(
                         "Encountered retriable exception while committing {}.", transactionalId, e);
@@ -104,7 +115,7 @@ public void commit(Collection<CommitRequest<KafkaCommittable>> requests)
                         ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,
                         kafkaProducerConfig.getProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG),
                         e);
-                backchannel.send(committable.getTransactionalId());
+                handleFailedTransaction(producer);
                 request.signalFailedWithKnownReason(e);
             } catch (InvalidTxnStateException e) {
                 // This exception only occurs when aborting after a commit or vice versa.
@@ -114,14 +125,14 @@ public void commit(Collection<CommitRequest<KafkaCommittable>> requests)
                                 + "Most likely the transaction has been aborted for some reason. Please check the Kafka logs for more details.",
                         request,
                         e);
-                backchannel.send(committable.getTransactionalId());
+                handleFailedTransaction(producer);
                 request.signalFailedWithKnownReason(e);
             } catch (UnknownProducerIdException e) {
                 LOG.error(
                         "Unable to commit transaction ({}) " + UNKNOWN_PRODUCER_ID_ERROR_MESSAGE,
                         request,
                         e);
-                backchannel.send(committable.getTransactionalId());
+                handleFailedTransaction(producer);
                 request.signalFailedWithKnownReason(e);
             } catch (Exception e) {
                 LOG.error(
@@ -134,6 +145,17 @@ public void commit(Collection<CommitRequest<KafkaCommittable>> requests)
         }
     }
 
+    private void handleFailedTransaction(FlinkKafkaInternalProducer<?, ?> producer) {
+        if (producer == null) {
+            return;
+        }
+        backchannel.send(TransactionFinished.erroneously(producer.getTransactionalId()));
+        if (producer == this.committingProducer) {
+            this.committingProducer.close();
+            this.committingProducer = null;
+        }
+    }
+
     @Override
     public void close() throws IOException {
         try {
@@ -150,8 +172,7 @@ public void close() throws IOException {
     private FlinkKafkaInternalProducer<?, ?> getProducer(KafkaCommittable committable) {
         if (committingProducer == null) {
             committingProducer =
-                    new FlinkKafkaInternalProducer<>(
-                            kafkaProducerConfig, committable.getTransactionalId());
+                    producerFactory.apply(kafkaProducerConfig, committable.getTransactionalId());
         } else {
             committingProducer.setTransactionId(committable.getTransactionalId());
         }

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

@@ -30,6 +30,7 @@
 import org.apache.flink.connector.kafka.lineage.LineageUtil;
 import org.apache.flink.connector.kafka.lineage.TypeDatasetFacet;
 import org.apache.flink.connector.kafka.lineage.TypeDatasetFacetProvider;
+import org.apache.flink.connector.kafka.sink.internal.FlinkKafkaInternalProducer;
 import org.apache.flink.core.io.SimpleVersionedSerializer;
 import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
 import org.apache.flink.streaming.api.connector.sink2.CommittableMessageTypeInfo;
@@ -110,7 +111,8 @@ public Committer<KafkaCommittable> createCommitter(CommitterInitContext context)
                 kafkaProducerConfig,
                 transactionalIdPrefix,
                 context.getTaskInfo().getIndexOfThisSubtask(),
-                context.getTaskInfo().getAttemptNumber());
+                context.getTaskInfo().getAttemptNumber(),
+                FlinkKafkaInternalProducer::new);
     }
 
     @Internal

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

@@ -24,13 +24,13 @@ public interface ProducerPool extends AutoCloseable {
      * Notify the pool that a transaction has finished. The producer with the given transactional id
      * can be recycled.
      */
-    void recycleByTransactionId(String transactionalId);
+    void recycleByTransactionId(String transactionalId, boolean success);
 
     /**
      * Get a producer for the given transactional id and checkpoint id. The producer is not recycled
      * until it is passed to the committer, the committer commits the transaction, and {@link
-     * #recycleByTransactionId(String)} is called. Alternatively, the producer can be recycled by
-     * {@link #recycle(FlinkKafkaInternalProducer)}.
+     * #recycleByTransactionId(String, boolean)} is called. Alternatively, the producer can be
+     * recycled by {@link #recycle(FlinkKafkaInternalProducer)}.
      */
     FlinkKafkaInternalProducer<byte[], byte[]> getTransactionalProducer(
             String transactionalId, long checkpointId);

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

@@ -32,6 +32,7 @@
 import java.util.Deque;
 import java.util.Map;
 import java.util.NavigableMap;
+import java.util.Objects;
 import java.util.Properties;
 import java.util.TreeMap;
 import java.util.function.Consumer;
@@ -105,17 +106,22 @@ public ProducerPoolImpl(
     }
 
     @Override
-    public void recycleByTransactionId(String transactionalId) {
+    public void recycleByTransactionId(String transactionalId, boolean success) {
         ProducerEntry producerEntry = producerByTransactionalId.remove(transactionalId);
         LOG.debug("Transaction {} finished, producer {}", transactionalId, producerEntry);
         if (producerEntry == null) {
             // during recovery, the committer may finish transactions that are not yet ongoing from
             // the writer's perspective
+            // these transaction will be closed by the second half of this method eventually
             return;
         }
 
         transactionalIdsByCheckpoint.remove(producerEntry.getCheckpointedTransaction());
-        recycleProducer(producerEntry.getProducer());
+        if (success) {
+            recycleProducer(producerEntry.getProducer());
+        } else {
+            closeProducer(producerEntry.getProducer());
+        }
 
         // In rare cases (only for non-chained committer), some transactions may not be detected to
         // be finished.
@@ -131,12 +137,18 @@ public void recycleByTransactionId(String transactionalId) {
         if (!earlierTransactions.isEmpty()) {
             for (String id : earlierTransactions.values()) {
                 ProducerEntry entry = producerByTransactionalId.remove(id);
-                recycleProducer(entry.getProducer());
+                closeProducer(entry.getProducer());
             }
             earlierTransactions.clear();
         }
     }
 
+    private void closeProducer(@Nullable FlinkKafkaInternalProducer<byte[], byte[]> producer) {
+        if (producer != null) {
+            producer.close();
+        }
+    }
+
     @Override
     public void recycle(FlinkKafkaInternalProducer<byte[], byte[]> producer) {
         recycleProducer(producer);
@@ -151,6 +163,7 @@ private void recycleProducer(@Nullable FlinkKafkaInternalProducer<byte[], byte[]
         if (producer == null) {
             return;
         }
+
         // For non-chained committer, we have a split brain scenario:
         // Both the writer and the committer have a producer representing the same transaction.
         // The committer producer has finished the transaction while the writer producer is still in
@@ -212,6 +225,7 @@ public void close() throws Exception {
                         closeAll(
                                 producerByTransactionalId.values().stream()
                                         .map(ProducerEntry::getProducer)
+                                        .filter(Objects::nonNull)
                                         .collect(Collectors.toList())),
                 producerPool::clear,
                 producerByTransactionalId::clear);