jolshan commented on code in PR #17698:
URL: https://github.com/apache/kafka/pull/17698#discussion_r1859362179
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -922,6 +946,103 @@ public void testTransactionManagerEnablesV2() {
assertTrue(transactionManager.isTransactionV2Enabled());
}
+ @Test
+ public void testTransactionV2AddPartitionAndOffsets() throws
InterruptedException {
+ initializeTransactionManager(Optional.of(transactionalId), true);
+ doInitTransactions();
+
+ transactionManager.beginTransaction();
+
+ Future<RecordMetadata> responseFuture = appendToAccumulator(tp0);
+
+ assertFalse(responseFuture.isDone());
+
+ prepareProduceResponse(Errors.NONE, producerId, epoch);
+ assertTrue(transactionManager.transactionContainsPartition(tp0));
+ assertTrue(transactionManager.isSendToPartitionAllowed(tp0));
+ assertFalse(responseFuture.isDone());
+ runUntil(responseFuture::isDone);
+
+ // Now, test adding the offsets.
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(tp1, new OffsetAndMetadata(1));
+
+ TransactionalRequestResult addOffsetsResult =
transactionManager.sendOffsetsToTransaction(
+ offsets, new ConsumerGroupMetadata(consumerGroupId));
+
+ assertTrue(transactionManager.hasPendingOffsetCommits());
+
+ // the result doesn't complete until TxnOffsetCommit returns
Review Comment:
Ending up discussing this a bit with Artem offline a little while back. This
could cause an issue if we didn't allow empty -> prepareAbort transition, but
since we have that (and its variants) this won't be a problem here either.
Essentially we always waited for addOffsetsToTxn to return before sending
the offset, so we wouldn't be in a limbo state about the transaction starting
(we set transactionStarted true when the offset is added) but now we don't have
that, so we require the ability to abort on the other states in the same way.
--
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
