jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1528996611


##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -3523,6 +3524,128 @@ public void 
testForegroundInvalidStateTransitionIsRecoverable() {
         assertFalse(transactionManager.hasOngoingTransaction());
     }
 
+    @Test
+    public void testAbortableTransactionExceptionInInitProducerId() {
+        TransactionalRequestResult initPidResult = 
transactionManager.initializeTransactions();
+        prepareFindCoordinatorResponse(Errors.NONE, false, 
CoordinatorType.TRANSACTION, transactionalId);
+        runUntil(() -> 
transactionManager.coordinator(CoordinatorType.TRANSACTION) != null);
+        assertEquals(brokerNode, 
transactionManager.coordinator(CoordinatorType.TRANSACTION));
+
+        prepareInitPidResponse(Errors.ABORTABLE_TRANSACTION, false, 
producerId, RecordBatch.NO_PRODUCER_EPOCH);
+        runUntil(transactionManager::hasError);
+        assertTrue(initPidResult.isCompleted());
+        assertFalse(initPidResult.isSuccessful());
+        assertThrows(AbortableTransactionException.class, 
initPidResult::await);
+        assertAbortableError(AbortableTransactionException.class);
+    }
+
+    @Test
+    public void testAbortableTransactionExceptionInAddPartitions() {
+        final TopicPartition tp = new TopicPartition("foo", 0);
+
+        doInitTransactions();
+
+        transactionManager.beginTransaction();
+        transactionManager.maybeAddPartition(tp);
+
+        prepareAddPartitionsToTxn(tp, Errors.ABORTABLE_TRANSACTION);
+        runUntil(transactionManager::hasError);
+        assertTrue(transactionManager.lastError() instanceof 
AbortableTransactionException);
+
+        assertAbortableError(AbortableTransactionException.class);
+    }
+
+    @Test
+    public void testAbortableTransactionExceptionInFindCoordinator() {
+        doInitTransactions();
+
+        transactionManager.beginTransaction();
+        TransactionalRequestResult sendOffsetsResult = 
transactionManager.sendOffsetsToTransaction(
+                singletonMap(new TopicPartition("foo", 0), new 
OffsetAndMetadata(39L)), new ConsumerGroupMetadata(consumerGroupId));
+
+        prepareAddOffsetsToTxnResponse(Errors.NONE, consumerGroupId, 
producerId, epoch);
+        runUntil(() -> !transactionManager.hasPartitionsToAdd());
+
+        prepareFindCoordinatorResponse(Errors.ABORTABLE_TRANSACTION, false, 
CoordinatorType.GROUP, consumerGroupId);
+        runUntil(transactionManager::hasError);
+        assertTrue(transactionManager.lastError() instanceof 
AbortableTransactionException);
+
+        runUntil(sendOffsetsResult::isCompleted);
+        assertFalse(sendOffsetsResult.isSuccessful());
+        assertTrue(sendOffsetsResult.error() instanceof 
AbortableTransactionException);
+
+        assertAbortableError(AbortableTransactionException.class);
+    }
+
+    @Test
+    public void testAbortableTransactionExceptionInEndTxn() throws 
InterruptedException {
+        doInitTransactions();
+
+        transactionManager.beginTransaction();
+        transactionManager.maybeAddPartition(tp0);
+        TransactionalRequestResult commitResult = 
transactionManager.beginCommit();
+
+        Future<RecordMetadata> responseFuture = appendToAccumulator(tp0);
+
+        assertFalse(responseFuture.isDone());
+        prepareAddPartitionsToTxnResponse(Errors.NONE, tp0, epoch, producerId);
+        prepareProduceResponse(Errors.NONE, producerId, epoch);
+        prepareEndTxnResponse(Errors.ABORTABLE_TRANSACTION, 
TransactionResult.COMMIT, producerId, epoch);
+
+        runUntil(commitResult::isCompleted);
+        runUntil(responseFuture::isDone);
+
+        assertThrows(KafkaException.class, commitResult::await);
+        assertFalse(commitResult.isSuccessful());
+        assertTrue(commitResult.isAcked());
+
+        assertAbortableError(AbortableTransactionException.class);
+        // make sure the exception was thrown directly from the follow-up 
calls.

Review Comment:
   nit: was this comment meant to be before further checks?



-- 
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-unsubscr...@kafka.apache.org

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

Reply via email to