diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java index c9d57cb9e53c1..9a41e2676ef10 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java +++ b/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()) ) ) ); diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java index 1c042778bbe89..313f3e2c1f499 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java +++ b/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 taskOffsetsFromMap(Map 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 taskIdsFromMap(Map> tasks) { List taskIds = new ArrayList<>(); tasks.keySet().stream().sorted().forEach(subtopologyId -> { diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java index cb9f7bd5fc25a..79f687a0bfb57 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java +++ b/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 described = context.sendStreamsGroupDescribe(List.of(groupId)); + assertEquals(1, described.size()); + List 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 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); diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMemberTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMemberTest.java index d332ffb632a8e..d5da36215b350 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMemberTest.java +++ b/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( diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java index a75c108314998..342b01aadad15 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java +++ b/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