dajac commented on code in PR #15183:
URL: https://github.com/apache/kafka/pull/15183#discussion_r1462083032
##########
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:
`verifyTransactionalReplay` replays and verifies. Is it what you want?
--
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
