KAFKA-19661: Streams groups sometimes describe as NOT_READY when STABLE #20457

lucasbru · 2025-09-02T12:16:50Z

Streams groups sometimes describe as NOT_READY when STABLE. That is, the
group is configured and all topics exist, but when you use LIST_GROUP
and STREAMS_GROUP_DESCRIBE, the group will show up as not ready.

The root cause seems to be that #19802 moved the creation of the soft
state configured topology from the replay path to the heartbeat. This
way, LIST_GROUP and STREAMS_GROUP_DESCRIBE, which show the snapshot of
the last committed offset, may not show the configured topology, because
the configured topology that is created in the heartbeat is "thrown
away", and the new group is recreated on the replay-path.

We can work around this issue by storing a reference to the streams
group objected created in the heartbeat immediately in the
GroupMetadataManager. This is a snapshottable collection, so if we fail
to write the corresponding records to the consumer offset topic, this
addition of the streams group will be reverted.

Streams groups sometimes describe as NOT_READY when STABLE. That is, the group is configured and all topics exist, but when you use LIST_GROUP and STREAMS_GROUP_DESCRIBE, the group will show up as not ready. The root cause seems to be that apache#19802 moved the creation of the soft state configured topology from the replay path to the heartbeat. This way, LIST_GROUP and STREAMS_GROUP_DESCRIBE, which show the snapshot of the last committed offset, may not show the configured topology, because the configured topology that is created in the heartbeat is "thrown away", and the new group is recreated on the replay-path. We can work around this issue by storing a reference to the streams group objected created in the heartbeat immediately in the GroupMetadataManager. This is a snapshottable collection, so if we fail to write the corresponding records to the consumer offset topic, this addition of the streams group will be reverted.

mjsax

I guess I need some more background information, as I am not familiar with this part of the code.

From the PR description:

this addition of the streams group will be reverted.

How does this revert happen?

mjsax · 2025-09-05T20:11:22Z

core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala

@@ -239,7 +239,8 @@ abstract class IntegrationTestHarness extends KafkaServerTestHarness {

  def createStreamsGroup[K, V](configOverrides: Properties = new Properties,
                               configsToRemove: List[String] = List(),
-                               inputTopic: String,
+                               inputTopics: Set[String],
+                               changelogTopics: Set[String] = Set(),


While this is only test code, I am not sure if I understand why we add this parameter now? Why is this required?

mjsax · 2025-09-05T20:13:31Z

core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala

          util.Set.of(),
          util.Map.of(),
-          util.Map.of(inputTopic + "-store-changelog", new StreamsRebalanceData.TopicInfo(Optional.of(1), Optional.empty(), util.Map.of())),
+          changelogTopics.map(c => (c, new StreamsRebalanceData.TopicInfo(Optional.empty(), Optional.empty(), util.Map.of()))).toMap.asJava,


Why do we change number of partition from 1 to "empty"?

mjsax · 2025-09-05T20:22:25Z

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

@@ -17194,8 +17288,7 @@ public void testStreamsGroupHeartbeatPartialResponseWhenNothingChanges() {
            new StreamsGroupHeartbeatResponseData()
                .setMemberId(memberId)
                .setMemberEpoch(1)
-                .setHeartbeatIntervalMs(5000)
-                .setEndpointInformationEpoch(-1),


Not sure why we need this change and the changes below?

mjsax · 2025-09-05T20:27:34Z

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

+                TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2, 3, 4, 5)))
+            .build();
+
+        // Commit the offset and test again


Not sure if I understand this comment? What "offset" do we commit? -- Also "test again"? Above we use context.groupMetadataManager.streamsGroup(groupId) to see if the group exists, but below we us ontext.groupMetadataManager.streamsGroupDescribe(...) -- so we do different things.

mjsax · 2025-09-05T20:28:30Z

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

+        context.commit();
+
+        List<StreamsGroupDescribeResponseData.DescribedGroup> actual = context.groupMetadataManager.streamsGroupDescribe(List.of(groupId), context.lastCommittedOffset);
+        StreamsGroupDescribeResponseData.DescribedGroup describedGroup = new StreamsGroupDescribeResponseData.DescribedGroup()


Suggested change

StreamsGroupDescribeResponseData.DescribedGroup describedGroup = new StreamsGroupDescribeResponseData.DescribedGroup()

StreamsGroupDescribeResponseData.DescribedGroup excpetedGroupDescription = new StreamsGroupDescribeResponseData.DescribedGroup()

mjsax · 2025-09-05T20:30:26Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

        if (group == null) {
-            return new StreamsGroup(logContext, snapshotRegistry, groupId, metrics);
+            StreamsGroup newGroup = new StreamsGroup(logContext, snapshotRegistry, groupId, metrics);
+            groups.put(groupId, newGroup);


This is the fix, right? We add the new group directly into the soft state? (same below)

mjsax · 2025-09-05T20:30:53Z

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

+                expectedMember.asStreamsGroupDescribeMember(TaskAssignmentTestUtil.mkTasksTuple(TaskRole.ACTIVE,
+                    TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2, 3, 4, 5)))
+            ))
+            .setGroupState(StreamsGroupState.STABLE.toString())


So this status would be NOT_READY w/o this fix? Because the HB itself did not update the soft state, which "describe" code path queries?

mjsax · 2025-09-05T20:35:52Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

-            removeGroup(groupId);
+            // In case of streams groups, which get inserted into memory immediately to store soft state,
+            // It may happen that the groups map contains the new streams groups already, and the classic group
+            // was removed already. In this case, we can ignore the tombstone.


Not sure if I fully understand? Also for my own education: when is replay actually executed? Only during GC load? During this phase, do we accept HB already (if not, softstate could not have been updated while we are loading).

It's not fully clear to me, what this means (from the PR description):

because the configured topology that is created in the heartbeat is "thrown
away", and the new group is recreated on the replay-path.

mjsax · 2025-09-05T20:37:19Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

@@ -4441,6 +4442,55 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
    }
  }

+  @Test
+  def testDescribeStreamsGroupsForStatelessTopology(): Unit = {


Why is it relevant if the group is stateless (vs stateful)?

github-actions bot added core Kafka Broker group-coordinator labels Sep 2, 2025

lucasbru force-pushed the KAFKA-19661 branch from 959a6d5 to 641d90c Compare September 5, 2025 11:53

lucasbru requested a review from mjsax September 5, 2025 11:54

lucasbru assigned mjsax Sep 5, 2025

add unit test

e84ba2f

lucasbru force-pushed the KAFKA-19661 branch from 641d90c to e84ba2f Compare September 5, 2025 12:38

mjsax added the streams label Sep 5, 2025

mjsax reviewed Sep 5, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

KAFKA-19661: Streams groups sometimes describe as NOT_READY when STABLE #20457

KAFKA-19661: Streams groups sometimes describe as NOT_READY when STABLE #20457

lucasbru commented Sep 2, 2025 •

edited by github-actions bot

Loading

Uh oh!

mjsax left a comment

Uh oh!

mjsax Sep 5, 2025

Uh oh!

mjsax Sep 5, 2025

Uh oh!

mjsax Sep 5, 2025

Uh oh!

mjsax Sep 5, 2025

Uh oh!

mjsax Sep 5, 2025

Uh oh!

mjsax Sep 5, 2025

Uh oh!

mjsax Sep 5, 2025

Uh oh!

mjsax Sep 5, 2025

Uh oh!

mjsax Sep 5, 2025

Uh oh!

Uh oh!

	StreamsGroupDescribeResponseData.DescribedGroup describedGroup = new StreamsGroupDescribeResponseData.DescribedGroup()
	StreamsGroupDescribeResponseData.DescribedGroup excpetedGroupDescription = new StreamsGroupDescribeResponseData.DescribedGroup()

KAFKA-19661: Streams groups sometimes describe as NOT_READY when STABLE #20457

Are you sure you want to change the base?

KAFKA-19661: Streams groups sometimes describe as NOT_READY when STABLE #20457

Conversation

lucasbru commented Sep 2, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

lucasbru commented Sep 2, 2025 •

edited by github-actions bot

Loading