[FLINK-37356] Remove transaction reset

Added in a recent commit that introduces ProducerPool, the method was supposed to cleanup a producer that is still in transactional state for reuse. Since we now don't reuse producers for abnormal states anymore, we can now simply double-commit before reuse avoiding at least some reflection hacks.

Author: AHeise

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

@@ -162,26 +162,6 @@ public long getProducerId() {
         return (long) getField(producerIdAndEpoch, "producerId");
     }
 
-    /**
-     * Sets the transaction manager state to uninitialized.
-     *
-     * <p>Can only be called if the producer is in a transaction. Its main purpose is to resolve the
-     * split brain scenario between writer and committer.
-     */
-    public void resetTransactionState() {
-        checkState(inTransaction, "Not in transactional state");
-        this.inTransaction = false;
-        this.hasRecordsInTransaction = false;
-        Object transactionManager = getTransactionManager();
-        synchronized (transactionManager) {
-            setField(transactionManager, "transactionalId", transactionalId);
-            setField(
-                    transactionManager,
-                    "currentState",
-                    getTransactionManagerState("UNINITIALIZED"));
-        }
-    }
-
     /**
      * Sets the transactional id and sets the transaction manager state to uninitialized.
      *

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

@@ -167,10 +167,11 @@ private void recycleProducer(@Nullable FlinkKafkaInternalProducer<byte[], byte[]
         // 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
-        // transaction. In this case, we forcibly complete the transaction, such that we can
-        // initialize it.
+        // transaction.
         if (producer.isInTransaction()) {
-            producer.resetTransactionState();
+            // Here we just double-commit the same transaction which succeeds in all cases
+            // because the producer shares the same epoch as the committer's producer
+            producer.commitTransaction();
         }
         producerPool.add(producer);