jolshan commented on code in PR #15155:
URL: https://github.com/apache/kafka/pull/15155#discussion_r1448139696
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java:
##########
@@ -1856,6 +1888,91 @@ public void testFetchOffsetsAtDifferentCommittedOffset()
{
), context.fetchOffsets("group", request, Long.MAX_VALUE));
}
+ @Test
+ public void testFetchOffsetsWithPendingTransactionalOffsets() {
+ OffsetMetadataManagerTestContext context = new
OffsetMetadataManagerTestContext.Builder().build();
+
+ context.groupMetadataManager.getOrMaybeCreateConsumerGroup("group",
true);
+
+ context.commitOffset("group", "foo", 0, 100L, 1);
+ context.commitOffset("group", "foo", 1, 110L, 1);
+ context.commitOffset("group", "bar", 0, 200L, 1);
+
+ context.commit();
+
+ assertEquals(3, context.lastWrittenOffset);
+ assertEquals(3, context.lastCommittedOffset);
+
+ context.commitOffset(10L, "group", "foo", 1, 111L, 1,
context.time.milliseconds());
+ context.commitOffset(10L, "group", "bar", 0, 201L, 1,
context.time.milliseconds());
+ // Note that bar-1 does not exist in the initial commits.
UNSTABLE_OFFSET_COMMIT errors
+ // must be returned in this case too.
+ context.commitOffset(10L, "group", "bar", 1, 211L, 1,
context.time.milliseconds());
+
+ // Always use the same request.
+ List<OffsetFetchRequestData.OffsetFetchRequestTopics> request =
Arrays.asList(
+ new OffsetFetchRequestData.OffsetFetchRequestTopics()
+ .setName("foo")
+ .setPartitionIndexes(Arrays.asList(0, 1)),
+ new OffsetFetchRequestData.OffsetFetchRequestTopics()
+ .setName("bar")
+ .setPartitionIndexes(Arrays.asList(0, 1))
+ );
+
+ // Fetching offsets with "require stable" (Long.MAX_VALUE) should
return the UNSTABLE_OFFSET_COMMIT
+ // errors for foo-1, bar-0 and bar-1.
Review Comment:
should we mention foo-0 will be returned in this comment and also the
comment below?
--
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
