KAFKA-20695: Extend "describe group" to return "task offset" metadata

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java

@@ -1308,7 +1308,11 @@ public StreamsGroupDescribeResponseData.DescribedGroup asDescribedGroup(
         members.entrySet(committedOffset).forEach(
             entry -> describedGroup.members().add(
                 entry.getValue().asStreamsGroupDescribeMember(
-                    targetAssignment.get(entry.getValue().memberId(), committedOffset)
+                    targetAssignment.get(entry.getValue().memberId(), committedOffset),
+                    // Task (end-)offsets are transient, unpersisted telemetry, so unlike the rest of the member state
+                    // they are read from the latest in-memory value rather than at committedOffset. Members that have
+                    // not reported any yield MemberTaskOffsets.EMPTY.
+                    taskOffsets(entry.getValue().memberId())
                 )
             )
         );

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java

@@ -19,6 +19,7 @@
 import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
 import org.apache.kafka.coordinator.group.generated.StreamsGroupCurrentMemberAssignmentValue;
 import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataValue;
+import org.apache.kafka.coordinator.group.streams.assignor.TaskId;
 
 import org.slf4j.Logger;
 
@@ -364,10 +365,17 @@ public boolean isReconciledTo(int targetAssignmentEpoch) {
      * Creates a member description for the streams group describe response from this member.
      *
      * @param targetAssignment The target assignment of this member in the corresponding group.
+     * @param taskOffsets      The latest per-task changelog offsets and end-offsets the member reported via its
+     *                         heartbeat. These are transient (never persisted) and read from live in-memory group
+     *                         state, so {@link MemberTaskOffsets#EMPTY} (or {@code null}) yields empty lists.
      *
      * @return The StreamsGroupMember mapped as StreamsGroupDescribeResponseData.Member.
      */
-    public StreamsGroupDescribeResponseData.Member asStreamsGroupDescribeMember(TasksTuple targetAssignment) {
+    public StreamsGroupDescribeResponseData.Member asStreamsGroupDescribeMember(
+        TasksTuple targetAssignment,
+        MemberTaskOffsets taskOffsets
+    ) {
+        final MemberTaskOffsets reportedOffsets = taskOffsets == null ? MemberTaskOffsets.EMPTY : taskOffsets;
         final StreamsGroupDescribeResponseData.Assignment describedTargetAssignment =
             new StreamsGroupDescribeResponseData.Assignment();
 
@@ -396,6 +404,8 @@ public StreamsGroupDescribeResponseData.Member asStreamsGroupDescribeMember(Task
                     .setKey(entry.getKey())
                     .setValue(entry.getValue())
             ).toList())
+            .setTaskOffsets(taskOffsetsFromMap(reportedOffsets.taskOffsets()))
+            .setTaskEndOffsets(taskOffsetsFromMap(reportedOffsets.taskEndOffsets()))
             .setProcessId(processId)
             .setTopologyEpoch(topologyEpoch)
             .setUserEndpoint(
@@ -407,6 +417,16 @@ public StreamsGroupDescribeResponseData.Member asStreamsGroupDescribeMember(Task
             );
     }
 
+    private static List<StreamsGroupDescribeResponseData.TaskOffset> taskOffsetsFromMap(Map<TaskId, Long> offsets) {
+        return offsets.entrySet().stream()
+            .sorted(Map.Entry.comparingByKey())
+            .map(entry -> new StreamsGroupDescribeResponseData.TaskOffset()
+                .setSubtopologyId(entry.getKey().subtopologyId())
+                .setPartition(entry.getKey().partition())
+                .setOffset(entry.getValue()))
+            .toList();
+    }
+
     private static List<StreamsGroupDescribeResponseData.TaskIds> taskIdsFromMap(Map<String, Set<Integer>> tasks) {
         List<StreamsGroupDescribeResponseData.TaskIds> taskIds = new ArrayList<>();
         tasks.keySet().stream().sorted().forEach(subtopologyId -> {

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java

@@ -11064,7 +11064,8 @@ public void testStreamsGroupDescribeNoErrors() {
                 .setGroupId(streamsGroupIds.get(1))
                 .setMembers(List.of(
                     memberBuilder.build().asStreamsGroupDescribeMember(
-                        TasksTuple.EMPTY
+                        TasksTuple.EMPTY,
+                        MemberTaskOffsets.EMPTY
                     )
                 ))
                 .setTopology(expectedTopology)
@@ -11076,6 +11077,70 @@ public void testStreamsGroupDescribeNoErrors() {
         assertEquals(expected, actual);
     }
 
+    @Test
+    public void testStreamsGroupDescribeReturnsPerMemberTaskOffsets() {
+        String groupId = "group-id";
+        int epoch = 10;
+        String memberId1 = "member-id-1";
+        String memberId2 = "member-id-2";
+        String subtopology1 = "subtopology1";
+        String fooTopicName = "foo";
+        StreamsTopology topology = new StreamsTopology(
+            0,
+            Map.of(subtopology1,
+                new StreamsGroupTopologyValue.Subtopology()
+                    .setSubtopologyId(subtopology1)
+                    .setSourceTopics(List.of(fooTopicName))
+            )
+        );
+
+        GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+            .withStreamsGroup(new StreamsGroupBuilder(groupId, epoch)
+                .withMember(streamsGroupMemberBuilderWithDefaults(memberId1)
+                    .setMemberEpoch(epoch)
+                    .setPreviousMemberEpoch(epoch - 1)
+                    .build())
+                .withMember(streamsGroupMemberBuilderWithDefaults(memberId2)
+                    .setMemberEpoch(epoch)
+                    .setPreviousMemberEpoch(epoch - 1)
+                    .build())
+                .withTopology(topology)
+            )
+            .build();
+
+        // Only member1 has reported changelog offsets; member2 has reported none. Both are described in the same
+        // response, so this verifies each member's row carries its own offsets (and that a member with none yields
+        // empty lists rather than borrowing another member's values).
+        StreamsGroup group = context.groupMetadataManager.streamsGroup(groupId);
+        group.updateTaskOffsets(memberId1, new MemberTaskOffsets(
+            Map.of(new org.apache.kafka.coordinator.group.streams.assignor.TaskId(subtopology1, 0), 100L),
+            Map.of(new org.apache.kafka.coordinator.group.streams.assignor.TaskId(subtopology1, 0), 500L)
+        ));
+
+        List<StreamsGroupDescribeResponseData.DescribedGroup> described = context.sendStreamsGroupDescribe(List.of(groupId));
+        assertEquals(1, described.size());
+        List<StreamsGroupDescribeResponseData.Member> members = described.get(0).members();
+        assertEquals(2, members.size());
+
+        StreamsGroupDescribeResponseData.Member describedMember1 = members.stream()
+            .filter(m -> m.memberId().equals(memberId1)).findFirst().orElseThrow();
+        StreamsGroupDescribeResponseData.Member describedMember2 = members.stream()
+            .filter(m -> m.memberId().equals(memberId2)).findFirst().orElseThrow();
+
+        assertEquals(
+            List.of(new StreamsGroupDescribeResponseData.TaskOffset()
+                .setSubtopologyId(subtopology1).setPartition(0).setOffset(100L)),
+            describedMember1.taskOffsets()
+        );
+        assertEquals(
+            List.of(new StreamsGroupDescribeResponseData.TaskOffset()
+                .setSubtopologyId(subtopology1).setPartition(0).setOffset(500L)),
+            describedMember1.taskEndOffsets()
+        );
+        assertEquals(List.of(), describedMember2.taskOffsets());
+        assertEquals(List.of(), describedMember2.taskEndOffsets());
+    }
+
     @Test
     public void testStreamsGroupDescribeWithErrors() {
         String groupId = "groupId";
@@ -11145,8 +11210,8 @@ public void testStreamsGroupDescribeBeforeAndAfterCommittingOffset() {
         describedGroup = new StreamsGroupDescribeResponseData.DescribedGroup()
             .setGroupId(streamsGroupId)
             .setMembers(Arrays.asList(
-                memberBuilder1.build().asStreamsGroupDescribeMember(TasksTuple.EMPTY),
-                memberBuilder2.build().asStreamsGroupDescribeMember(assignment)
+                memberBuilder1.build().asStreamsGroupDescribeMember(TasksTuple.EMPTY, MemberTaskOffsets.EMPTY),
+                memberBuilder2.build().asStreamsGroupDescribeMember(assignment, MemberTaskOffsets.EMPTY)
             ))
             .setTopology(
                 new StreamsGroupDescribeResponseData.Topology()
@@ -18777,6 +18842,23 @@ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
             ),
             group.taskOffsets(memberId)
         );
+
+        // Describing the group surfaces the retained offsets on the member, even though they were never persisted.
+        context.commit();
+        List<StreamsGroupDescribeResponseData.DescribedGroup> describedGroups =
+            context.groupMetadataManager.streamsGroupDescribe(List.of(groupId), context.lastCommittedOffset).describedGroups();
+        assertEquals(1, describedGroups.size());
+        StreamsGroupDescribeResponseData.Member describedMember = describedGroups.get(0).members().get(0);
+        assertEquals(
+            List.of(new StreamsGroupDescribeResponseData.TaskOffset()
+                .setSubtopologyId(subtopology1).setPartition(0).setOffset(12L)),
+            describedMember.taskOffsets()
+        );
+        assertEquals(
+            List.of(new StreamsGroupDescribeResponseData.TaskOffset()
+                .setSubtopologyId(subtopology1).setPartition(0).setOffset(20L)),
+            describedMember.taskEndOffsets()
+        );
     }
 
     @Test
@@ -19019,7 +19101,7 @@ public void testJoinEmptyStreamsGroupAndDescribe() {
             )
             .setMembers(Collections.singletonList(
                 expectedMember.asStreamsGroupDescribeMember(TaskAssignmentTestUtil.mkTasksTuple(TaskRole.ACTIVE,
-                    TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2, 3, 4, 5)))
+                    TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2, 3, 4, 5)), MemberTaskOffsets.EMPTY)
             ))
             .setGroupState(StreamsGroupState.STABLE.toString())
             .setGroupEpoch(2);

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMemberTest.java

@@ -21,6 +21,7 @@
 import org.apache.kafka.coordinator.group.generated.StreamsGroupCurrentMemberAssignmentValue.TaskIds;
 import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataValue;
 import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataValue.KeyValue;
+import org.apache.kafka.coordinator.group.streams.assignor.TaskId;
 
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
@@ -343,8 +344,20 @@ public void testAsStreamsGroupDescribeMember() {
             mkMap(mkEntry(SUBTOPOLOGY2, new HashSet<>(assignedTasks2))),
             mkMap(mkEntry(SUBTOPOLOGY3, new HashSet<>(assignedTasks1)))
         );
+        MemberTaskOffsets taskOffsets = new MemberTaskOffsets(
+            mkMap(
+                mkEntry(new TaskId(SUBTOPOLOGY1, 0), 100L),
+                mkEntry(new TaskId(SUBTOPOLOGY1, 2), 120L),
+                mkEntry(new TaskId(SUBTOPOLOGY2, 1), 200L)
+            ),
+            mkMap(
+                mkEntry(new TaskId(SUBTOPOLOGY1, 0), 150L),
+                mkEntry(new TaskId(SUBTOPOLOGY1, 2), 170L),
+                mkEntry(new TaskId(SUBTOPOLOGY2, 1), 250L)
+            )
+        );
 
-        StreamsGroupDescribeResponseData.Member actual = member.asStreamsGroupDescribeMember(targetAssignment);
+        StreamsGroupDescribeResponseData.Member actual = member.asStreamsGroupDescribeMember(targetAssignment, taskOffsets);
         StreamsGroupDescribeResponseData.Member expected = new StreamsGroupDescribeResponseData.Member()
             .setMemberId(MEMBER_ID)
             .setMemberEpoch(MEMBER_EPOCH)
@@ -357,6 +370,16 @@ public void testAsStreamsGroupDescribeMember() {
             .setClientTags(List.of(
                 new StreamsGroupDescribeResponseData.KeyValue().setKey(CLIENT_TAG_KEY).setValue(CLIENT_TAG_VALUE))
             )
+            .setTaskOffsets(List.of(
+                new StreamsGroupDescribeResponseData.TaskOffset().setSubtopologyId(SUBTOPOLOGY1).setPartition(0).setOffset(100L),
+                new StreamsGroupDescribeResponseData.TaskOffset().setSubtopologyId(SUBTOPOLOGY1).setPartition(2).setOffset(120L),
+                new StreamsGroupDescribeResponseData.TaskOffset().setSubtopologyId(SUBTOPOLOGY2).setPartition(1).setOffset(200L)
+            ))
+            .setTaskEndOffsets(List.of(
+                new StreamsGroupDescribeResponseData.TaskOffset().setSubtopologyId(SUBTOPOLOGY1).setPartition(0).setOffset(150L),
+                new StreamsGroupDescribeResponseData.TaskOffset().setSubtopologyId(SUBTOPOLOGY1).setPartition(2).setOffset(170L),
+                new StreamsGroupDescribeResponseData.TaskOffset().setSubtopologyId(SUBTOPOLOGY2).setPartition(1).setOffset(250L)
+            ))
             .setAssignment(
                 new StreamsGroupDescribeResponseData.Assignment()
                     .setActiveTasks(List.of(
@@ -404,11 +427,29 @@ public void testAsStreamsGroupDescribeMember() {
     @Test
     public void testAsStreamsGroupDescribeWithTargetAssignmentNull() {
         final StreamsGroupMember member = createStreamsGroupMember();
-        StreamsGroupDescribeResponseData.Member streamsGroupDescribeMember = member.asStreamsGroupDescribeMember(null);
+        StreamsGroupDescribeResponseData.Member streamsGroupDescribeMember =
+            member.asStreamsGroupDescribeMember(null, MemberTaskOffsets.EMPTY);
 
         assertEquals(new StreamsGroupDescribeResponseData.Assignment(), streamsGroupDescribeMember.targetAssignment());
     }
 
+    @Test
+    public void testAsStreamsGroupDescribeMemberWithEmptyOrNullTaskOffsets() {
+        final StreamsGroupMember member = createStreamsGroupMember();
+
+        // Task (end-)offsets are transient telemetry: a member that has reported none (EMPTY), or a null passed
+        // defensively, must describe as empty lists rather than throwing.
+        StreamsGroupDescribeResponseData.Member fromEmpty =
+            member.asStreamsGroupDescribeMember(TasksTuple.EMPTY, MemberTaskOffsets.EMPTY);
+        assertEquals(List.of(), fromEmpty.taskOffsets());
+        assertEquals(List.of(), fromEmpty.taskEndOffsets());
+
+        StreamsGroupDescribeResponseData.Member fromNull =
+            member.asStreamsGroupDescribeMember(TasksTuple.EMPTY, null);
+        assertEquals(List.of(), fromNull.taskOffsets());
+        assertEquals(List.of(), fromNull.taskEndOffsets());
+    }
+
     @Test
     public void testHasAssignedTasksChanged() {
         TasksTupleWithEpochs assignedTasks1 = new TasksTupleWithEpochs(

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java

@@ -1231,6 +1231,11 @@ public void testAsDescribedGroupWithStreamsTopologyHavingSubtopologies() {
             .setAssignedTasks(TasksTupleWithEpochs.EMPTY)
             .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
             .build());
+        // Transient, unpersisted per-task offsets reported by the member; the describe path must surface them.
+        group.updateTaskOffsets("member1", new MemberTaskOffsets(
+            Map.of(new TaskId("sub-1", 0), 5L),
+            Map.of(new TaskId("sub-1", 0), 9L)
+        ));
         snapshotRegistry.idempotentCreateSnapshot(1);
 
         StreamsGroupDescribeResponseData.DescribedGroup describedGroup = group.asDescribedGroup(1);
@@ -1255,6 +1260,14 @@ public void testAsDescribedGroupWithStreamsTopologyHavingSubtopologies() {
 
         assertEquals(1, describedGroup.members().size());
         assertEquals("member1", describedGroup.members().get(0).memberId());
+        assertEquals(
+            List.of(new StreamsGroupDescribeResponseData.TaskOffset().setSubtopologyId("sub-1").setPartition(0).setOffset(5L)),
+            describedGroup.members().get(0).taskOffsets()
+        );
+        assertEquals(
+            List.of(new StreamsGroupDescribeResponseData.TaskOffset().setSubtopologyId("sub-1").setPartition(0).setOffset(9L)),
+            describedGroup.members().get(0).taskEndOffsets()
+        );
     }
 
     @Test