MINOR: Fix response for consumer group describe with empty group id (#20030) (#20036)

rajinisivaram · web-flow · commit 6351bc05aafc · 2025-06-25T17:56:58.000+01:00
ConsumerGroupDescribe with an empty group id returns a response containing `null` groupId in a non-nullable field. Since the response cannot be serialized, this results in UNKNOWN_SERVER_ERROR being returned to the client. This PR sets the group id in the response to an empty string instead and adds request tests for empty group id.

Reviewers: David Jacot &lt;djacot@confluent.io&gt;, Chia-Ping Tsai &lt;chia7712@gmail.com&gt;
diff --git a/core/src/test/scala/unit/kafka/server/ConsumerGroupDescribeRequestTest.scala b/core/src/test/scala/unit/kafka/server/ConsumerGroupDescribeRequestTest.scala
@@ -211,6 +211,20 @@ class ConsumerGroupDescribeRequestTest(cluster: ClusterInstance) extends GroupCo
         )
 
         assertEquals(expected, actual)
+
+        val unknownGroupResponse = consumerGroupDescribe(
+          groupIds = List("grp-unknown"),
+          includeAuthorizedOperations = true,
+          version = version.toShort,
+        )
+        assertEquals(Errors.GROUP_ID_NOT_FOUND.code, unknownGroupResponse.head.errorCode())
+
+        val emptyGroupResponse = consumerGroupDescribe(
+          groupIds = List(""),
+          includeAuthorizedOperations = true,
+          version = version.toShort,
+        )
+        assertEquals(Errors.INVALID_GROUP_ID.code, emptyGroupResponse.head.errorCode())
       }
     } finally {
       admin.close()
diff --git a/core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala b/core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala
@@ -301,6 +301,48 @@ class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
     }
   }
 
+  @ClusterTest
+  def testEmptyConsumerGroupId(): Unit = {
+    val admin = cluster.admin()
+
+    // Creates the __consumer_offsets topics because it won't be created automatically
+    // in this test because it does not use FindCoordinator API.
+    try {
+      TestUtils.createOffsetsTopicWithAdmin(
+        admin = admin,
+        brokers = cluster.brokers.values().asScala.toSeq,
+        controllers = cluster.controllers().values().asScala.toSeq
+      )
+
+      // Heartbeat request to join the group. Note that the member subscribes
+      // to an nonexistent topic.
+      val consumerGroupHeartbeatRequest = new ConsumerGroupHeartbeatRequest.Builder(
+        new ConsumerGroupHeartbeatRequestData()
+          .setGroupId("")
+          .setMemberId(Uuid.randomUuid().toString)
+          .setMemberEpoch(0)
+          .setRebalanceTimeoutMs(5 * 60 * 1000)
+          .setSubscribedTopicNames(List("foo").asJava)
+          .setTopicPartitions(List.empty.asJava),
+        true
+      ).build()
+
+      // Send the request until receiving a successful response. There is a delay
+      // here because the group coordinator is loaded in the background.
+      var consumerGroupHeartbeatResponse: ConsumerGroupHeartbeatResponse = null
+      TestUtils.waitUntilTrue(() => {
+        consumerGroupHeartbeatResponse = connectAndReceive(consumerGroupHeartbeatRequest)
+        consumerGroupHeartbeatResponse.data.errorCode == Errors.INVALID_REQUEST.code
+      }, msg = s"Did not receive the expected error. Last response $consumerGroupHeartbeatResponse.")
+
+      // Verify the response.
+      assertEquals(Errors.INVALID_REQUEST.code, consumerGroupHeartbeatResponse.data.errorCode)
+      assertEquals("GroupId can't be empty.", consumerGroupHeartbeatResponse.data.errorMessage)
+    } finally {
+      admin.close()
+    }
+  }
+
   @ClusterTest
   def testConsumerGroupHeartbeatWithEmptySubscription(): Unit = {
     val admin = cluster.admin()
diff --git a/core/src/test/scala/unit/kafka/server/DeleteGroupsRequestTest.scala b/core/src/test/scala/unit/kafka/server/DeleteGroupsRequestTest.scala
@@ -99,8 +99,8 @@ class DeleteGroupsRequestTest(cluster: ClusterInstance) extends GroupCoordinator
       )
 
       deleteGroups(
-        groupIds = List("grp-non-empty", "grp"),
-        expectedErrors = List(Errors.NON_EMPTY_GROUP, Errors.NONE),
+        groupIds = List("grp-non-empty", "grp", ""),
+        expectedErrors = List(Errors.NON_EMPTY_GROUP, Errors.NONE, Errors.GROUP_ID_NOT_FOUND),
         version = version.toShort
       )
 
diff --git a/core/src/test/scala/unit/kafka/server/DescribeGroupsRequestTest.scala b/core/src/test/scala/unit/kafka/server/DescribeGroupsRequestTest.scala
@@ -104,10 +104,15 @@ class DescribeGroupsRequestTest(cluster: ClusterInstance) extends GroupCoordinat
             .setGroupId("grp-unknown")
             .setGroupState(ClassicGroupState.DEAD.toString) // Return DEAD group when the group does not exist.
             .setErrorCode(if (version >= 6) Errors.GROUP_ID_NOT_FOUND.code() else Errors.NONE.code())
-            .setErrorMessage(if (version >= 6) "Group grp-unknown not found." else null)
+            .setErrorMessage(if (version >= 6) "Group grp-unknown not found." else null),
+          new DescribedGroup()
+            .setGroupId("")
+            .setGroupState(ClassicGroupState.DEAD.toString) // Return DEAD group when the group does not exist.
+            .setErrorCode(if (version >= 6) Errors.GROUP_ID_NOT_FOUND.code() else Errors.NONE.code())
+            .setErrorMessage(if (version >= 6) "Group  not found." else null)
         ),
         describeGroups(
-          groupIds = List("grp-1", "grp-2", "grp-unknown"),
+          groupIds = List("grp-1", "grp-2", "grp-unknown", ""),
           version = version.toShort
         )
       )
diff --git a/core/src/test/scala/unit/kafka/server/HeartbeatRequestTest.scala b/core/src/test/scala/unit/kafka/server/HeartbeatRequestTest.scala
@@ -190,6 +190,15 @@ class HeartbeatRequestTest(cluster: ClusterInstance) extends GroupCoordinatorBas
         expectedError = Errors.UNKNOWN_MEMBER_ID,
         version = version.toShort
       )
+
+      // Heartbeat with empty group id.
+      heartbeat(
+        groupId = "",
+        memberId = leaderMemberId,
+        generationId = -1,
+        expectedError = Errors.INVALID_GROUP_ID,
+        version = version.toShort
+      )
     }
   }
 }
diff --git a/core/src/test/scala/unit/kafka/server/JoinGroupRequestTest.scala b/core/src/test/scala/unit/kafka/server/JoinGroupRequestTest.scala
@@ -149,6 +149,17 @@ class JoinGroupRequestTest(cluster: ClusterInstance) extends GroupCoordinatorBas
         )
       )
 
+      // Join with an empty group id.
+      verifyJoinGroupResponseDataEquals(
+        new JoinGroupResponseData()
+          .setErrorCode(Errors.INVALID_GROUP_ID.code)
+          .setProtocolName(if (version >= 7) null else ""),
+        sendJoinRequest(
+          groupId = "",
+          version = version.toShort
+        )
+      )
+
       // Join with an inconsistent protocolType.
       verifyJoinGroupResponseDataEquals(
         new JoinGroupResponseData()
diff --git a/core/src/test/scala/unit/kafka/server/OffsetFetchRequestTest.scala b/core/src/test/scala/unit/kafka/server/OffsetFetchRequestTest.scala
@@ -269,6 +269,39 @@ class OffsetFetchRequestTest(cluster: ClusterInstance) extends GroupCoordinatorB
           )
         )
 
+        // Fetch with empty group id.
+        assertEquals(
+          new OffsetFetchResponseData.OffsetFetchResponseGroup()
+            .setGroupId("")
+            .setTopics(List(
+              new OffsetFetchResponseData.OffsetFetchResponseTopics()
+                .setName("foo")
+                .setPartitions(List(
+                  new OffsetFetchResponseData.OffsetFetchResponsePartitions()
+                    .setPartitionIndex(0)
+                    .setCommittedOffset(-1L),
+                  new OffsetFetchResponseData.OffsetFetchResponsePartitions()
+                    .setPartitionIndex(1)
+                    .setCommittedOffset(-1L),
+                  new OffsetFetchResponseData.OffsetFetchResponsePartitions()
+                    .setPartitionIndex(5)
+                    .setCommittedOffset(-1L)
+                ).asJava)
+            ).asJava),
+          fetchOffsets(
+            groupId = "",
+            memberId = memberId,
+            memberEpoch = memberEpoch,
+            partitions = List(
+              new TopicPartition("foo", 0),
+              new TopicPartition("foo", 1),
+              new TopicPartition("foo", 5) // This one does not exist.
+            ),
+            requireStable = requireStable,
+            version = version.toShort
+          )
+        )
+
         // Fetch with stale member epoch.
         assertEquals(
           new OffsetFetchResponseData.OffsetFetchResponseGroup()
diff --git a/core/src/test/scala/unit/kafka/server/SyncGroupRequestTest.scala b/core/src/test/scala/unit/kafka/server/SyncGroupRequestTest.scala
@@ -76,6 +76,17 @@ class SyncGroupRequestTest(cluster: ClusterInstance) extends GroupCoordinatorBas
         version = version.toShort
       )
 
+      // Sync with empty group id.
+      verifySyncGroupWithOldProtocol(
+        groupId = "",
+        memberId = "member-id",
+        generationId = -1,
+        expectedProtocolType = null,
+        expectedProtocolName = null,
+        expectedError = Errors.INVALID_GROUP_ID,
+        version = version.toShort
+      )
+
       val metadata = ConsumerProtocol.serializeSubscription(
         new ConsumerPartitionAssignor.Subscription(Collections.singletonList("foo"))
       ).array
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
@@ -636,7 +636,7 @@ public CompletableFuture<List<ConsumerGroupDescribeResponseData.DescribedGroup>>
             } else {
                 futures.add(CompletableFuture.completedFuture(Collections.singletonList(
                     new ConsumerGroupDescribeResponseData.DescribedGroup()
-                        .setGroupId(null)
+                        .setGroupId("")
                         .setErrorCode(Errors.INVALID_GROUP_ID.code())
                 )));
             }
@@ -687,7 +687,7 @@ public CompletableFuture<List<DescribedGroup>> shareGroupDescribe(
             } else {
                 futures.add(CompletableFuture.completedFuture(Collections.singletonList(
                     new ShareGroupDescribeResponseData.DescribedGroup()
-                        .setGroupId(null)
+                        .setGroupId("")
                         .setErrorCode(Errors.INVALID_GROUP_ID.code())
                 )));
             }
@@ -736,7 +736,7 @@ public CompletableFuture<List<DescribeGroupsResponseData.DescribedGroup>> descri
             if (groupId == null) {
                 futures.add(CompletableFuture.completedFuture(Collections.singletonList(
                     new DescribeGroupsResponseData.DescribedGroup()
-                        .setGroupId(null)
+                        .setGroupId("")
                         .setErrorCode(Errors.INVALID_GROUP_ID.code())
                 )));
             } else {
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
@@ -1038,7 +1038,7 @@ public void testDescribeGroupsInvalidGroupId() throws Exception {
             .setGroupId("");
         List<DescribeGroupsResponseData.DescribedGroup> expectedDescribedGroups = Arrays.asList(
             new DescribeGroupsResponseData.DescribedGroup()
-                .setGroupId(null)
+                .setGroupId("")
                 .setErrorCode(Errors.INVALID_GROUP_ID.code()),
             describedGroup
         );
@@ -1470,11 +1470,11 @@ public void testConsumerGroupDescribeInvalidGroupId() throws ExecutionException,
         service.startup(() -> partitionCount);
 
         ConsumerGroupDescribeResponseData.DescribedGroup describedGroup = new ConsumerGroupDescribeResponseData.DescribedGroup()
-            .setGroupId(null)
+            .setGroupId("")
             .setErrorCode(Errors.INVALID_GROUP_ID.code());
         List<ConsumerGroupDescribeResponseData.DescribedGroup> expectedDescribedGroups = Arrays.asList(
             new ConsumerGroupDescribeResponseData.DescribedGroup()
-                .setGroupId(null)
+                .setGroupId("")
                 .setErrorCode(Errors.INVALID_GROUP_ID.code()),
             describedGroup
         );
@@ -2347,11 +2347,11 @@ public void testShareGroupDescribeInvalidGroupId() throws ExecutionException, In
         service.startup(() -> partitionCount);
 
         ShareGroupDescribeResponseData.DescribedGroup describedGroup = new ShareGroupDescribeResponseData.DescribedGroup()
-            .setGroupId(null)
+            .setGroupId("")
             .setErrorCode(Errors.INVALID_GROUP_ID.code());
         List<ShareGroupDescribeResponseData.DescribedGroup> expectedDescribedGroups = Arrays.asList(
             new ShareGroupDescribeResponseData.DescribedGroup()
-                .setGroupId(null)
+                .setGroupId("")
                 .setErrorCode(Errors.INVALID_GROUP_ID.code()),
             describedGroup
         );

Original file line number	Diff line number	Diff line change
`@@ -99,8 +99,8 @@ class DeleteGroupsRequestTest(cluster: ClusterInstance) extends GroupCoordinator`
`99`	`99`	`)`
`100`	`100`
`101`	`101`	`deleteGroups(`
`102`		`- groupIds = List("grp-non-empty", "grp"),`
`103`		`- expectedErrors = List(Errors.NON_EMPTY_GROUP, Errors.NONE),`
	`102`	`+ groupIds = List("grp-non-empty", "grp", ""),`
	`103`	`+ expectedErrors = List(Errors.NON_EMPTY_GROUP, Errors.NONE, Errors.GROUP_ID_NOT_FOUND),`
`104`	`104`	`version = version.toShort`
`105`	`105`	`)`
`106`	`106`
Original file line number	Diff line number	Diff line change
`@@ -190,6 +190,15 @@ class HeartbeatRequestTest(cluster: ClusterInstance) extends GroupCoordinatorBas`
`190`	`190`	`expectedError = Errors.UNKNOWN_MEMBER_ID,`
`191`	`191`	`version = version.toShort`
`192`	`192`	`)`
	`193`	`+`
	`194`	`+ // Heartbeat with empty group id.`
	`195`	`+ heartbeat(`
	`196`	`+ groupId = "",`
	`197`	`+ memberId = leaderMemberId,`
	`198`	`+ generationId = -1,`
	`199`	`+ expectedError = Errors.INVALID_GROUP_ID,`
	`200`	`+ version = version.toShort`
	`201`	`+ )`
`193`	`202`	`}`
`194`	`203`	`}`
`195`	`204`	`}`