jolshan commented on code in PR #14321: URL: https://github.com/apache/kafka/pull/14321#discussion_r1313618740
########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java: ########## @@ -1409,6 +1444,82 @@ public void testFetchAllOffsetsAtDifferentCommittedOffset() { ), context.fetchAllOffsets("group", Long.MAX_VALUE)); } + @Test + public void testConsumerGroupOffsetFetchWithMemberIdAndEpoch() { + OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build(); + // Create consumer group. + ConsumerGroup group = context.groupMetadataManager.getOrMaybeCreateConsumerGroup("group", true); + // Create member. + group.getOrMaybeCreateMember("member", true); + // Commit offset. + context.commitOffset("group", "foo", 0, 100L, 1); + + // Fetch offsets case. + List<OffsetFetchRequestData.OffsetFetchRequestTopics> topics = Collections.singletonList( + new OffsetFetchRequestData.OffsetFetchRequestTopics() + .setName("foo") + .setPartitionIndexes(Collections.singletonList(0)) + ); + + assertEquals(Collections.singletonList( + new OffsetFetchResponseData.OffsetFetchResponseTopics() + .setName("foo") + .setPartitions(Collections.singletonList( + mkOffsetPartitionResponse(0, 100L, 1, "metadata") + )) + ), context.fetchOffsets("group", "member", 0, topics, Long.MAX_VALUE)); + + // Fetch all offsets case. + assertEquals(Collections.singletonList( + new OffsetFetchResponseData.OffsetFetchResponseTopics() + .setName("foo") + .setPartitions(Collections.singletonList( + mkOffsetPartitionResponse(0, 100L, 1, "metadata") + )) + ), context.fetchAllOffsets("group", "member", 0, Long.MAX_VALUE)); + } + + @Test + public void testConsumerGroupOffsetFetchWithUnknownMemberId() { + OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build(); + context.groupMetadataManager.getOrMaybeCreateConsumerGroup("group", true); + + // Fetch offsets case. + List<OffsetFetchRequestData.OffsetFetchRequestTopics> topics = Collections.singletonList( + new OffsetFetchRequestData.OffsetFetchRequestTopics() + .setName("foo") + .setPartitionIndexes(Collections.singletonList(0)) + ); + + assertThrows(UnknownMemberIdException.class, + () -> context.fetchOffsets("group", "member", 0, topics, Long.MAX_VALUE)); + + // Fetch all offsets case. + assertThrows(UnknownMemberIdException.class, + () -> context.fetchAllOffsets("group", "member", 0, Long.MAX_VALUE)); + } + + @Test + public void testConsumerGroupOffsetFetchWithStaleMemberEpoch() { Review Comment: This is the member epoch -1 right? -- 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