[GitHub] [kafka] CalvinConfluent commented on a diff in pull request #14321: KAFKA-14499: [6/N] Add MemberId and MemberEpoch to OffsetFetchRequest

["CalvinConfluent (via GitHub)" <gi...@apache.org>]

CalvinConfluent commented on code in PR #14321:
URL: https://github.com/apache/kafka/pull/14321#discussion_r1313307656


##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -538,17 +537,46 @@ public void validateOffsetCommit(
         if (memberEpoch < 0 && members().isEmpty()) return;
 
         final ConsumerGroupMember member = getOrMaybeCreateMember(memberId, false);
-        if (memberEpoch != member.memberEpoch()) {
-            throw Errors.STALE_MEMBER_EPOCH.exception();
-        }
+        validateMemberEpoch(memberEpoch, member.memberEpoch());
     }
 
     /**
      * Validates the OffsetFetch request.
+     *
+     * @param memberId              The member id for consumer groups.
+     * @param memberEpoch           The member epoch for consumer groups.
+     * @param lastCommittedOffset   The last committed offsets in the timeline.
      */
     @Override
-    public void validateOffsetFetch() {
-        // Nothing.
+    public void validateOffsetFetch(
+        String memberId,
+        int memberEpoch,
+        long lastCommittedOffset
+    ) throws UnknownMemberIdException, StaleMemberEpochException {
+        // When the member epoch is -1, the request comes from the admin client. In this case,
+        // the request can commit offsets if the group is empty.

Review Comment:
   Should we check the group is empty here? The code seems misaligned with the comment.



##########
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:
   Maybe add the case for admin client?



-- 
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-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org