[GitHub] [kafka] dajac commented on a change in pull request #11452: KAFKA-13412; Ensure initTransactions() safe for retry after timeout

[GitBox <gi...@apache.org>]

dajac commented on a change in pull request #11452:
URL: https://github.com/apache/kafka/pull/11452#discussion_r750335078



##########
File path: clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
##########
@@ -903,6 +903,47 @@ public void testPartitionsForWithNullTopic() {
         }
     }
 
+    @Test
+    public void testInitTransactionsResponseAfterTimeout() throws Exception {
+        int maxBlockMs = 500;
+
+        Map<String, Object> configs = new HashMap<>();
+        configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "bad-transaction");
+        configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMs);
+        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
+
+        Time time = new MockTime();
+        MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
+        ProducerMetadata metadata = newMetadata(0, Long.MAX_VALUE);
+        metadata.updateWithCurrentRequestVersion(initialUpdateResponse, false, time.milliseconds());
+
+        MockClient client = new MockClient(time, metadata);
+
+        ExecutorService executor = Executors.newFixedThreadPool(1);
+
+        Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
+            new StringSerializer(), metadata, client, null, time);
+        try {
+            client.prepareResponse(
+                request -> request instanceof FindCoordinatorRequest &&
+                    ((FindCoordinatorRequest) request).data().keyType() == FindCoordinatorRequest.CoordinatorType.TRANSACTION.id(),
+                FindCoordinatorResponse.prepareResponse(Errors.NONE, "bad-transaction", host1));
+
+            Future<?> future = executor.submit(producer::initTransactions);
+            TestUtils.waitForCondition(client::hasInFlightRequests, "blah blah");

Review comment:
       nit: I suppose that you planned to change the message here :)

##########
File path: clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
##########
@@ -903,6 +903,47 @@ public void testPartitionsForWithNullTopic() {
         }
     }
 
+    @Test
+    public void testInitTransactionsResponseAfterTimeout() throws Exception {
+        int maxBlockMs = 500;
+
+        Map<String, Object> configs = new HashMap<>();
+        configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "bad-transaction");
+        configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMs);
+        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
+
+        Time time = new MockTime();
+        MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1));
+        ProducerMetadata metadata = newMetadata(0, Long.MAX_VALUE);
+        metadata.updateWithCurrentRequestVersion(initialUpdateResponse, false, time.milliseconds());
+
+        MockClient client = new MockClient(time, metadata);
+
+        ExecutorService executor = Executors.newFixedThreadPool(1);
+
+        Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
+            new StringSerializer(), metadata, client, null, time);
+        try {
+            client.prepareResponse(
+                request -> request instanceof FindCoordinatorRequest &&
+                    ((FindCoordinatorRequest) request).data().keyType() == FindCoordinatorRequest.CoordinatorType.TRANSACTION.id(),
+                FindCoordinatorResponse.prepareResponse(Errors.NONE, "bad-transaction", host1));
+
+            Future<?> future = executor.submit(producer::initTransactions);
+            TestUtils.waitForCondition(client::hasInFlightRequests, "blah blah");
+
+            time.sleep(maxBlockMs);
+            TestUtils.assertFutureThrows(future, TimeoutException.class);
+
+            client.respond(initProducerIdResponse(1L, (short) 5, Errors.NONE));
+
+            Thread.sleep(1000);
+            producer.initTransactions();
+        } finally {
+            producer.close(Duration.ZERO);
+        }
+    }
+

Review comment:
       Should we also add new tests for `commitTransaction` and `abortTransaction` which both seem to suffer from the same bug?

##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
##########
@@ -1184,15 +1186,19 @@ private TxnOffsetCommitHandler txnOffsetCommitHandler(TransactionalRequestResult
     }
 
     private TransactionalRequestResult handleCachedTransactionRequestResult(
-            Supplier<TransactionalRequestResult> transactionalRequestResultSupplier,
-            State targetState) {
+        Supplier<TransactionalRequestResult> transactionalRequestResultSupplier,
+        State transientState
+    ) {
         ensureTransactional();
 
-        if (pendingResult != null && currentState == targetState) {
+        if (pendingResult != null && !pendingResult.isAcked()) {
             TransactionalRequestResult result = pendingResult;
-            if (result.isCompleted())
+            if (result.isCompleted() && priorState == transientState) {
                 pendingResult = null;
-            return result;
+                return result;
+            } else if (currentState == transientState) {
+                return result;

Review comment:
       Why don't we set `pendingResult` to `null` when `isCompleted()` is `true` here? I suppose that we assume that the `currentState` which transition to the next state whenever the response is received so we should never have a completed request here. Therefore, the "prior state" check will handle all the cases. Is my reasoning correct?




-- 
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: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org