(kafka) branch trunk updated: MINOR: Write the streams topology tombstone before the group epoch tombstone (#23552)

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dajac pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 433ec656ce1 MINOR: Write the streams topology tombstone before the group epoch tombstone (#23552)
433ec656ce1 is described below

commit 433ec656ce151d789867efa62b63c20e9aa1d0af
Author: David Jacot <da...@apache.org>
AuthorDate: Wed Sep 23 10:40:56 2026 +0200

    MINOR: Write the streams topology tombstone before the group epoch tombstone (#23552)
    
    Deleting a streams group leaks its topics in the group coordinator's
    topic-to-groups index (`groupsByTopics`), which `onMetadataUpdate` uses
    to tell groups that a topic they depend on changed.
    
    A streams group's entry in that index is derived from its topology:
    replaying a `StreamsGroupTopologyValue` record subscribes the group to
    the topology's required topics and replaying its tombstone unsubscribes
    it. `StreamsGroup.createGroupTombstoneRecords` wrote the
    `StreamsGroupMetadataValue` tombstone before the
    `StreamsGroupTopologyValue` tombstone. On replay, the first one removes
    the group, so the second one finds no group and returns before
    unsubscribing anything. The entry survives for the lifetime of the
    coordinator, and every deleted streams group adds to it. The consumer of
    the index checks that the group still exists, so the effect is bounded
    growth of the index and a spurious metadata refresh request on a group
    that later reuses the id, not a failure.
    
    This change writes the topology tombstone before the group epoch
    tombstone. This restores the invariant the other group types already
    have: every sibling record is torn down before the epoch tombstone, so
    the epoch tombstone is the last record of a deletion. Deletions already
    in the log with the previous order keep leaking on replay until
    compaction removes their tombstones.
    
    The new test builds a streams group with a topology, replays exactly the
    records that `createGroupTombstoneRecords` produces, and verifies that
    the group is removed and that the index no longer references it. It
    fails with the previous record order.
    
    Reviewers: Lucas Brutschy <lb...@confluent.io>
---
 .../coordinator/group/streams/StreamsGroup.java    |  4 ++-
 .../group/GroupMetadataManagerTest.java            | 31 +++++++++++++++++++---
 2 files changed, 30 insertions(+), 5 deletions(-)

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 213e49a2eb1..11eb0875cba 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
@@ -1138,8 +1138,10 @@ public class StreamsGroup implements Group {
             records.add(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord(groupId(), memberId))
         );
 
-        records.add(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord(groupId()));
+        // The topology tombstone must precede the group epoch tombstone: replaying it unsubscribes
+        // the group from its topics, which is only possible while the group still exists.
         records.add(StreamsCoordinatorRecordHelpers.newStreamsGroupTopologyRecordTombstone(groupId()));
+        records.add(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord(groupId()));
     }
 
     /**
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 a56652cbe7a..7cba2b92310 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
@@ -13196,8 +13196,8 @@ public class GroupMetadataManagerTest {
             StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentTombstoneRecord(groupId, memberId),
             StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentMetadataTombstoneRecord(groupId),
             StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord(groupId, memberId),
-            StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord(groupId),
-            StreamsCoordinatorRecordHelpers.newStreamsGroupTopologyRecordTombstone(groupId)
+            StreamsCoordinatorRecordHelpers.newStreamsGroupTopologyRecordTombstone(groupId),
+            StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord(groupId)
         );
         assertEquals(expectedRecords, records);
     }
@@ -24169,8 +24169,8 @@ public class GroupMetadataManagerTest {
 
         List<CoordinatorRecord> expectedRecords = List.of(
             StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentMetadataTombstoneRecord(streamsGroupId),
-            StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord(streamsGroupId),
-            StreamsCoordinatorRecordHelpers.newStreamsGroupTopologyRecordTombstone(streamsGroupId)
+            StreamsCoordinatorRecordHelpers.newStreamsGroupTopologyRecordTombstone(streamsGroupId),
+            StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord(streamsGroupId)
         );
 
         assertEquals(Errors.MEMBER_ID_REQUIRED.code(), joinResult.joinFuture.get().errorCode());
@@ -25409,6 +25409,29 @@ public class GroupMetadataManagerTest {
         assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
     }
 
+    @Test
+    public void testReplayStreamsGroupEpochTombstoneUnsubscribesTopologyTopics() {
+        String subtopologyId = "subtopology-1";
+        StreamsTopology topology = new StreamsTopology(1, Map.of(subtopologyId,
+            new StreamsGroupTopologyValue.Subtopology()
+                .setSubtopologyId(subtopologyId)
+                .setSourceTopics(List.of("bar"))));
+        GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+            .withStreamsGroup(new StreamsGroupBuilder("foo", 10).withTopology(topology))
+            .build();
+
+        assertEquals(Set.of("foo"), context.groupMetadataManager.groupsSubscribedToTopic("bar"));
+
+        // The topology tombstone must be written before the group epoch tombstone: once the group
+        // is gone, replaying the topology tombstone can no longer unsubscribe the group.
+        List<CoordinatorRecord> records = new ArrayList<>();
+        context.groupMetadataManager.streamsGroup("foo").createGroupTombstoneRecords(records);
+        records.forEach(context::replay);
+
+        assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+        assertEquals(Set.of(), context.groupMetadataManager.groupsSubscribedToTopic("bar"));
+    }
+
     @Test
     public void testReplayStreamsGroupTargetAssignmentMember() {
         GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()