jolshan commented on code in PR #13787:
URL: https://github.com/apache/kafka/pull/13787#discussion_r1261385072
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -3057,6 +3057,56 @@ public void
testSenderShouldRetryWithBackoffOnRetriableError() {
assertEquals(RETRY_BACKOFF_MS, time.milliseconds() - request2);
}
+ @Test
+ public void testReceiveFailedBatchTwiceWithTransactions() throws Exception
{
+ ProducerIdAndEpoch producerIdAndEpoch = new
ProducerIdAndEpoch(123456L, (short) 0);
+ apiVersions.update("0",
NodeApiVersions.create(ApiKeys.INIT_PRODUCER_ID.id, (short) 0, (short) 3));
+ TransactionManager txnManager = new TransactionManager(logContext,
"testFailTwice", 60000, 100, apiVersions);
+
+ setupWithTransactionState(txnManager);
+ doInitTransactions(txnManager, producerIdAndEpoch);
+
+ txnManager.beginTransaction();
+ txnManager.maybeAddPartition(tp0);
+ client.prepareResponse(buildAddPartitionsToTxnResponseData(0,
Collections.singletonMap(tp0, Errors.NONE)));
+ sender.runOnce();
+
+ // Send first ProduceRequest
+ Future<RecordMetadata> request1 = appendToAccumulator(tp0);
+ sender.runOnce(); // send request
+
+ Node node = metadata.fetch().nodes().get(0);
+ time.sleep(2000L);
Review Comment:
If we don't sleep, then the future does not complete. There is another test
in the file that explains this only expires the batch if it hits the delivery
timeout
```
// We add 600 millis to expire the first batch but not the second.
// Note deliveryTimeoutMs is 1500.
```
--
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
