nym3r0s commented on a change in pull request #9280:
URL: https://github.com/apache/kafka/pull/9280#discussion_r500166517
##########
File path:
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
##########
@@ -2358,6 +2359,41 @@ public void testForceShutdownWithIncompleteTransaction()
{
}
}
+ @Test
+ public void testTransactionAbortedExceptionOnAbortWithoutError() throws
InterruptedException, ExecutionException {
+ ProducerIdAndEpoch producerIdAndEpoch = new
ProducerIdAndEpoch(123456L, (short) 0);
+ TransactionManager txnManager = new TransactionManager(logContext,
"testTransactionAbortedExceptionOnAbortWithoutError", 60000, 100, apiVersions,
false);
+
+ setupWithTransactionState(txnManager, false, null);
+ doInitTransactions(txnManager, producerIdAndEpoch);
+ // Begin the transaction
+ txnManager.beginTransaction();
+ txnManager.maybeAddPartitionToTransaction(tp0);
+ client.prepareResponse(new AddPartitionsToTxnResponse(0,
Collections.singletonMap(tp0, Errors.NONE)));
+ // Run it once so that the partition is added to the transaction.
+ sender.runOnce();
+ // Append a record to the accumulator.
+ FutureRecordMetadata metadata1 = appendToAccumulator(tp0,
time.milliseconds(), "key1", "value1");
+ // Now abort the transaction manually.
+ txnManager.beginAbort();
+ // Try to send.
+ // This should abort the existing transaction and
+ // drain all the unsent batches with a TransactionAbortedException.
+ sender.runOnce();
+ // Now attempt to fetch the result for the record.
+ try {
+ // This should fail since we aborted the transaction.
+ metadata1.get();
Review comment:
@hachikuji - Thank you - I've updated the test 👍
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
