jeffkbkim commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1462074651
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java:
##########
@@ -2691,6 +2716,57 @@ public void testReplayTransactionEndMarkerWithCommit() {
));
}
+ @Test
+ public void
testReplayTransactionEndMarkerKeepsTheMostRecentCommittedOffset() {
+ OffsetMetadataManagerTestContext context = new
OffsetMetadataManagerTestContext.Builder().build();
+
+ // Add pending transactional offset commit for producer id 5.
+ verifyTransactionalReplay(context, 5L, "foo", "bar", 0, new
OffsetAndMetadata(
+ 0L,
+ 100L,
+ OptionalInt.empty(),
+ "small",
+ context.time.milliseconds(),
+ OptionalLong.empty()
+ ));
+
+ // Add regular offset commit.
+ verifyReplay(context, "foo", "bar", 0, new OffsetAndMetadata(
+ 1L,
+ 101L,
+ OptionalInt.empty(),
+ "small",
+ context.time.milliseconds(),
+ OptionalLong.empty()
+ ));
+
+ // Replaying an end marker to commit transaction of producer id 5.
+ context.replayEndTransactionMarker(5L, TransactionResult.COMMIT);
+
+ // The pending offset is removed...
+ assertNull(context.offsetMetadataManager.pendingTransactionalOffset(
+ 5L,
+ "foo",
+ "bar",
+ 0
+ ));
Review Comment:
should we add a check that the pending txn offset exists before replaying
the end txn marker in L2744?
--
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
