KAFKA-19945: Always set status field in StreamsGroupHeartbeat response (#21031)

The status field should always set in the response, even when empty, to
ensure that the status gets reset on the client.

We are not doing this due to an oversight - it is defined as nullable,
and it is null when not set. So status does not clear correctly on the
client, which ignores the field if it's null.

This can cause the streams application to incorrectly timeout, if the
source topic does not exist when the application is first started, but
it is created after the application started. Otherwise, there is no
noticable difference.

Reviewers: Lucas Brutschy <[email protected]>

Author: lucasbru

clients/src/main/resources/common/message/StreamsGroupHeartbeatResponse.json

@@ -53,8 +53,8 @@
     { "name": "TaskOffsetIntervalMs", "type": "int32", "versions": "0+",
       "about": "The interval in which the task changelog offsets on a client are updated on the broker. The offsets are sent with the next heartbeat after this time has passed." },
 
-    { "name": "Status", "type":  "[]Status", "versions":  "0+",  "nullableVersions": "0+", "default": "null",
-      "about": "Indicate zero or more status for the group.  Null if unchanged since last heartbeat." },
+    { "name": "Status", "type":  "[]Status", "versions":  "0+",  "nullableVersions": "0+",
+      "about": "Indicate zero or more status for the group." },
 
     // The streams app knows which partitions to fetch from given this information
     { "name": "ActiveTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",

core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala

@@ -209,7 +209,7 @@ class StreamsGroupHeartbeatRequestTest(cluster: ClusterInstance) extends GroupCo
       assertNotNull(streamsGroupHeartbeatResponse, "StreamsGroupHeartbeatResponse should not be null")
       assertEquals(memberId, streamsGroupHeartbeatResponse.memberId())
       assertEquals(3, streamsGroupHeartbeatResponse.memberEpoch())
-      assertEquals(null, streamsGroupHeartbeatResponse.status())
+      assertEquals(List.empty.asJava, streamsGroupHeartbeatResponse.status())
       val expectedActiveTasks = List(
         new StreamsGroupHeartbeatResponseData.TaskIds()
           .setSubtopologyId("subtopology-1")

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

@@ -2153,9 +2153,7 @@ private CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> stream
                 )
         ));
 
-        if (!returnedStatus.isEmpty()) {
-            response.setStatus(returnedStatus);
-        }
+        response.setStatus(returnedStatus);
         return new CoordinatorResult<>(records, new StreamsGroupHeartbeatResult(response, internalTopicsToBeCreated));
     }
 
@@ -4215,7 +4213,8 @@ private CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> stream
         }
         StreamsGroupHeartbeatResponseData response = new StreamsGroupHeartbeatResponseData()
             .setMemberId(memberId)
-            .setMemberEpoch(memberEpoch);
+            .setMemberEpoch(memberEpoch)
+            .setStatus(List.of());
 
         if (instanceId == null) {
             StreamsGroupMember member = group.getMemberOrThrow(memberId);