Fix onClose, remove TransactionState monitoring attempt - not needed

astubbs · astubbs · commit 42e502cd5606 · 2020-11-11T17:13:20.000Z
diff --git a/parallel-consumer-core/src/main/java/io/confluent/TransactionState.java b/parallel-consumer-core/src/main/java/io/confluent/TransactionState.java
diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/AbstractOffsetCommitter.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/AbstractOffsetCommitter.java
@@ -31,8 +31,6 @@ public void commit() {
             ConsumerGroupMetadata groupMetadata = consumer.groupMetadata();
 
             commitOffsets(offsetsToSend, groupMetadata);
-
-            afterCommit();
         }
     }
 
@@ -42,7 +40,4 @@ protected void onOffsetCommitSuccess(final Map<TopicPartition, OffsetAndMetadata
 
     protected abstract void commitOffsets(final Map<TopicPartition, OffsetAndMetadata> offsetsToSend, final ConsumerGroupMetadata groupMetadata);
 
-    protected void afterCommit() {
-        // noop
-    }
 }
diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ConsumerOffsetCommitter.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ConsumerOffsetCommitter.java
@@ -35,12 +35,13 @@ public ConsumerOffsetCommitter(final Consumer<K, V> newConsumer, final WorkManag
     protected void commitOffsets(final Map<TopicPartition, OffsetAndMetadata> offsetsToSend, final ConsumerGroupMetadata groupMetadata) {
         if (isSync) {
             consumer.commitSync(offsetsToSend);
+            commitBarrier.countDown(); // todo use Condition instead?
         } else {
             consumer.commitAsync(offsetsToSend, (offsets, exception) -> {
-                if (exception == null) {
+                if (exception != null) {
                     log.error("Error committing offsets", exception);
+                    // todo keep work in limbo until async response is received?
                 }
-                commitBarrier.countDown();
             });
         }
     }
diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ParallelConsumerOptions.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ParallelConsumerOptions.java
@@ -45,7 +45,7 @@ public enum ProcessingOrder {
      * // TODO we could just auto detect this from the Producers state. However, forcing it to be specified makes the choice more verbose?
      */
     @Builder.Default
-    private final boolean usingTransactionalProducer = false;
+    private final boolean usingTransactionalProducer = true;
 
     /**
      * Don't have more than this many uncommitted messages in process
diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ProducerManager.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ProducerManager.java
@@ -1,6 +1,5 @@
 package io.confluent.parallelconsumer;
 
-import io.confluent.TransactionState;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.Consumer;
@@ -12,10 +11,10 @@
 import org.apache.kafka.common.TopicPartition;
 
 import java.lang.reflect.Field;
+import java.lang.reflect.Method;
 import java.time.Duration;
 import java.util.Map;
 import java.util.concurrent.Future;
-import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 import static io.confluent.csid.utils.StringUtils.msg;
@@ -26,8 +25,6 @@ public class ProducerManager<K, V> extends AbstractOffsetCommitter<K, V> impleme
     protected final Producer<K, V> producer;
     private final ParallelConsumerOptions options;
 
-    protected final TransactionState ts = new TransactionState();
-
     /**
      * The {@link KafkaProducer) isn't actually completely thread safe, at least when using it transactionally. We must
      * be careful not to send messages to the producer, while we are committing a transaction - "Cannot call send in
@@ -57,7 +54,6 @@ private void initProducer(final Producer<K, V> newProducer) {
                 log.debug("Initialising producer transaction session...");
                 producer.initTransactions();
                 producer.beginTransaction();
-                ts.setInTransaction();
             } catch (KafkaException e) {
                 log.error("Make sure your producer is setup for transactions - specifically make sure it's {} is set.", ProducerConfig.TRANSACTIONAL_ID_CONFIG, e);
                 throw e;
@@ -76,16 +72,21 @@ private void initProducer(final Producer<K, V> newProducer) {
      */
     @SneakyThrows
     private boolean getProducerIsTransactional(final Producer<K, V> newProducer) {
-        Field coordinatorField = newProducer.getClass().getDeclaredField("transactionManager");
-        coordinatorField.setAccessible(true);
-        TransactionManager transactionManager = (TransactionManager) coordinatorField.get(newProducer);
+        TransactionManager transactionManager = getTransactionManager(newProducer);
         if (transactionManager == null) {
             return false;
         } else {
             return transactionManager.isTransactional();
         }
     }
 
+    private TransactionManager getTransactionManager(final Producer<K, V> newProducer) throws NoSuchFieldException, IllegalAccessException {
+        Field coordinatorField = newProducer.getClass().getDeclaredField("transactionManager");
+        coordinatorField.setAccessible(true);
+        TransactionManager transactionManager = (TransactionManager) coordinatorField.get(newProducer);
+        return transactionManager;
+    }
+
     /**
      * Produce a message back to the broker.
      * <p>
@@ -152,15 +153,31 @@ protected void commitOffsets(final Map<TopicPartition, OffsetAndMetadata> offset
                     ReentrantReadWriteLock.WriteLock writeLock = producerCommitLock.writeLock();
                     writeLock.lock();
                     try {
-                        producer.commitTransaction();
+                        boolean retrying = retryCount > 0;
+                        if (retrying) {
+                            if (isTransactionCompleting()) {
+                                // try wait again
+                                producer.commitTransaction();
+                            }
+                            if (isTransactionReady()) {
+                                // tx has completed since we last tried, start a new one
+                                producer.beginTransaction();
+                            }
+                            boolean ready = (lastErrorSavedForRethrow != null) ? !lastErrorSavedForRethrow.getMessage().contains("Invalid transition attempted from state READY to state COMMITTING_TRANSACTION") : true;
+                            if (ready) {
+                                // try again
+                                log.error("Was already ready - tx completed between interrupt and retry");
+                            }
+                        } else {
+                            // happy path
+                            producer.commitTransaction();
+                            producer.beginTransaction();
+                        }
                     } finally {
                         writeLock.unlock();
                     }
-
                 }
 
-                ts.setNotInTransaction();
-
                 onOffsetCommitSuccess(offsetsToSend);
 
                 committed = true;
@@ -175,18 +192,42 @@ protected void commitOffsets(final Map<TopicPartition, OffsetAndMetadata> offset
         }
     }
 
-    @Override
-    protected void afterCommit() {
-        // begin tx for next cycle
-        producer.beginTransaction();
-        ts.setInTransaction();
+    /**
+     * TODO talk about alternatives to this brute force approach for retrying committing transactions
+     */
+    @SneakyThrows
+    private boolean isTransactionCompleting() {
+        TransactionManager transactionManager = getTransactionManager(producer);
+        Method method = transactionManager.getClass().getDeclaredMethod("isCompleting");
+        method.setAccessible(true);
+        return (boolean) method.invoke(transactionManager);
+    }
+
+    /**
+     * TODO talk about alternatives to this brute force approach for retrying committing transactions
+     */
+    @SneakyThrows
+    private boolean isTransactionReady() {
+        TransactionManager transactionManager = getTransactionManager(producer);
+        Method method = transactionManager.getClass().getDeclaredMethod("isReady");
+        method.setAccessible(true);
+        return (boolean) method.invoke(transactionManager);
     }
 
+    /**
+     * Assumes the system is drained at this point, or draining is not desired.
+     */
     public void close(final Duration timeout) {
         log.debug("Closing producer, assuming no more in flight...");
-        if (options.isUsingTransactionalProducer() && ts.isInTransaction()) {
-            // close started after tx began, but before work was done, otherwise a tx wouldn't have been started
-            producer.abortTransaction();
+        if (options.isUsingTransactionalProducer() && !isTransactionReady()) {
+            ReentrantReadWriteLock.WriteLock writeLock = producerCommitLock.writeLock();
+            writeLock.lock();
+            try {
+                // close started after tx began, but before work was done, otherwise a tx wouldn't have been started
+                producer.abortTransaction();
+            } finally {
+                writeLock.unlock();
+            }
         }
         producer.close(timeout);
     }
diff --git a/parallel-consumer-examples/parallel-consumer-example-core/src/test/java/io/confluent/parallelconsumer/examples/core/Bug25AppTest.java b/parallel-consumer-examples/parallel-consumer-example-core/src/test/java/io/confluent/parallelconsumer/examples/core/Bug25AppTest.java
@@ -6,6 +6,7 @@
 
 import io.confluent.parallelconsumer.ParallelConsumerOptions;
 import io.confluent.parallelconsumer.integrationTests.KafkaTest;
+import lombok.AllArgsConstructor;
 import lombok.RequiredArgsConstructor;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
@@ -28,22 +29,35 @@
 @Slf4j
 public class Bug25AppTest extends KafkaTest<String, String> {
 
-    @SneakyThrows
+    int DEAFULT_MAX_POLL_RECORDS_CONFIG = 500;
+
+    @Test
+    public void testTransactionalDefaultMaxPoll() {
+        boolean tx = true;
+        runTest(tx, DEAFULT_MAX_POLL_RECORDS_CONFIG);
+    }
+
+    @Test
+    public void testNonTransactionalDefaultMaxPoll() {
+        boolean tx = false;
+        runTest(tx, DEAFULT_MAX_POLL_RECORDS_CONFIG);
+    }
+
     @Test
     public void testTransactional() {
         boolean tx = true;
-        runTest(tx);
+        runTest(tx, 1);  // Sometimes causes test to fail (default 500)
     }
 
-    @SneakyThrows
     @Test
     public void testNonTransactional() {
         boolean tx = false;
-        runTest(tx);
+        runTest(tx, 1);  // Sometimes causes test to fail (default 500)
     }
 
-    private void runTest(final boolean tx) {
-        AppUnderTest coreApp = new AppUnderTest(tx, ParallelConsumerOptions.builder().ordering(KEY).usingTransactionalProducer(tx).build());
+    @SneakyThrows
+    private void runTest(boolean tx, int maxPoll) {
+        AppUnderTest coreApp = new AppUnderTest(tx, ParallelConsumerOptions.builder().ordering(KEY).usingTransactionalProducer(tx).build(), maxPoll);
 
         ensureTopic(coreApp.inputTopic, 1);
         ensureTopic(coreApp.outputTopic, 1);
@@ -68,16 +82,17 @@ private void runTest(final boolean tx) {
         coreApp.close();
     }
 
-    @RequiredArgsConstructor
+    @AllArgsConstructor
     class AppUnderTest extends CoreApp {
 
         final boolean tx;
         final ParallelConsumerOptions options;
+        int MAX_POLL_RECORDS_CONFIG;
 
         @Override
         Consumer<String, String> getKafkaConsumer() {
             Properties props = kcu.props;
-            props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1); // Sometimes causes test to fail (default 500)
+            props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, MAX_POLL_RECORDS_CONFIG);
             props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10000);
             return new KafkaConsumer<>(props);
         }

Original file line number	Diff line number	Diff line change
`@@ -31,8 +31,6 @@ public void commit() {`
`31`	`31`	`ConsumerGroupMetadata groupMetadata = consumer.groupMetadata();`
`32`	`32`
`33`	`33`	`commitOffsets(offsetsToSend, groupMetadata);`
`34`		`-`
`35`		`- afterCommit();`
`36`	`34`	`}`
`37`	`35`	`}`
`38`	`36`
`@@ -42,7 +40,4 @@ protected void onOffsetCommitSuccess(final Map<TopicPartition, OffsetAndMetadata`
`42`	`40`
`43`	`41`	`protected abstract void commitOffsets(final Map<TopicPartition, OffsetAndMetadata> offsetsToSend, final ConsumerGroupMetadata groupMetadata);`
`44`	`42`
`45`		`- protected void afterCommit() {`
`46`		`- // noop`
`47`		`- }`
`48`	`43`	`}`
Original file line number	Diff line number	Diff line change
`@@ -35,12 +35,13 @@ public ConsumerOffsetCommitter(final Consumer<K, V> newConsumer, final WorkManag`
`35`	`35`	`protected void commitOffsets(final Map<TopicPartition, OffsetAndMetadata> offsetsToSend, final ConsumerGroupMetadata groupMetadata) {`
`36`	`36`	`if (isSync) {`
`37`	`37`	`consumer.commitSync(offsetsToSend);`
	`38`	`+ commitBarrier.countDown(); // todo use Condition instead?`
`38`	`39`	`} else {`
`39`	`40`	`consumer.commitAsync(offsetsToSend, (offsets, exception) -> {`
`40`		`- if (exception == null) {`
	`41`	`+ if (exception != null) {`
`41`	`42`	`log.error("Error committing offsets", exception);`
	`43`	`+ // todo keep work in limbo until async response is received?`
`42`	`44`	`}`
`43`		`- commitBarrier.countDown();`
`44`	`45`	`});`
`45`	`46`	`}`
`46`	`47`	`}`