KAFKA-20620: Address review feedback on KIP-1331 RPC scaffolding

- Bump StreamsGroupHeartbeat request and response to v1 so the new
  TopologyDescriptionRequired field is actually negotiated.
- Mark StreamsGroupTopologyDescriptionUpdate as latestVersionUnstable
  until the broker handler lands; exclude from RequestQuotaTest via a
  new UnimplementedApis set.
- Fix StreamsGroupDescribeRequest.getErrorResponse to set
  TopologyDescriptionStatus=ERROR when the client requested topology;
  previously defaulted to NOT_REQUESTED, contradicting KIP-1331.
- Switch the Update RPC getErrorResponse to ApiError.fromThrowable so
  cause messages survive while UNKNOWN_SERVER_ERROR is sanitized.
- Reword "plugin" out of client-visible strings (Errors.java default
  message, JSON about fields).
- Add lifecycle Javadoc to the new Request class (drops duplicated
  error code list) and error-code Javadoc to the new Response class;
  document the TopologyDescription/Status biconditional invariant and
  the errorCounts() vs Status=ERROR caveat on DescribeResponse.
- Add "Supported errors" comment block to the new UpdateResponse JSON.
- Strengthen tests: nullable SinkTopic round-trip, TopologyDescription
  null/NOT_STORED coverage, explicit assertion on DeleteGroups v3
  per-group ErrorMessage.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: aliehsaeedii

clients/src/main/java/org/apache/kafka/common/protocol/Errors.java

@@ -422,7 +422,7 @@ public enum Errors {
     STREAMS_TOPOLOGY_FENCED(132, "The supplied topology epoch is outdated.", StreamsTopologyFencedException::new),
     SHARE_SESSION_LIMIT_REACHED(133, "The limit of share sessions has been reached.", ShareSessionLimitReachedException::new),
     GROUP_DELETION_FAILED(134, "DeleteGroups could not complete; see the error message on the per-group result for details.", GroupDeletionFailedException::new),
-    STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED(135, "The topology description plugin failed to process the request.", StreamsTopologyDescriptionUpdateFailedException::new);
+    STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED(135, "The broker could not process the topology description update; see the error message for details.", StreamsTopologyDescriptionUpdateFailedException::new);
 
     private static final Logger log = LoggerFactory.getLogger(Errors.class);
 

clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupDescribeRequest.java

@@ -58,13 +58,19 @@ public StreamsGroupDescribeRequest(StreamsGroupDescribeRequestData data, short v
     public StreamsGroupDescribeResponse getErrorResponse(int throttleTimeMs, Throwable e) {
         StreamsGroupDescribeResponseData data = new StreamsGroupDescribeResponseData()
             .setThrottleTimeMs(throttleTimeMs);
-        // Set error for each group
+        short errorCode = Errors.forException(e).code();
+        boolean topologyRequested = this.data.includeTopologyDescription();
         this.data.groupIds().forEach(
-            groupId -> data.groups().add(
-                new StreamsGroupDescribeResponseData.DescribedGroup()
-                    .setGroupId(groupId)
-                    .setErrorCode(Errors.forException(e).code())
-            )
+            groupId -> {
+                StreamsGroupDescribeResponseData.DescribedGroup group =
+                    new StreamsGroupDescribeResponseData.DescribedGroup()
+                        .setGroupId(groupId)
+                        .setErrorCode(errorCode);
+                if (topologyRequested) {
+                    group.setTopologyDescriptionStatus((byte) 2); // ERROR
+                }
+                data.groups().add(group);
+            }
         );
         return new StreamsGroupDescribeResponse(data);
     }

clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupDescribeResponse.java

@@ -35,6 +35,15 @@
  * - {@link Errors#INVALID_GROUP_ID}
  * - {@link Errors#GROUP_ID_NOT_FOUND}
  * - {@link Errors#TOPIC_AUTHORIZATION_FAILED}
+ *
+ * <p>TopologyDescription invariant (v1+): the {@code TopologyDescription} field is non-null
+ * if and only if {@code TopologyDescriptionStatus} is {@code AVAILABLE} (3). The broker MUST
+ * set the status to {@code AVAILABLE} whenever it attaches a {@code TopologyDescription},
+ * and leave {@code TopologyDescription} null for any other status value.
+ *
+ * <p>Note: {@code TopologyDescriptionStatus == ERROR} (2) is a data-level signal, not a
+ * protocol error, and is intentionally not reflected in {@link #errorCounts()}. Operators
+ * tracking topology-description failures should monitor the status field directly.
  */
 public class StreamsGroupDescribeResponse extends AbstractResponse {
 

clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupTopologyDescriptionUpdateRequest.java

@@ -19,9 +19,17 @@
 import org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
 import org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateResponseData;
 import org.apache.kafka.common.protocol.ApiKeys;
-import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.protocol.Readable;
 
+/**
+ * Sent by a Streams client to push its topology description to the broker, in response
+ * to {@code TopologyDescriptionRequired=true} on a {@code StreamsGroupHeartbeatResponse}.
+ * The broker validates that {@code MemberId} still belongs to the group, checks the
+ * {@code TopologyEpoch} against the group's current epoch, and persists the description.
+ * See KIP-1331.
+ *
+ * <p>Legal error codes are documented on {@link StreamsGroupTopologyDescriptionUpdateResponse}.
+ */
 public class StreamsGroupTopologyDescriptionUpdateRequest extends AbstractRequest {
 
     public static class Builder extends AbstractRequest.Builder<StreamsGroupTopologyDescriptionUpdateRequest> {
@@ -52,11 +60,12 @@ public StreamsGroupTopologyDescriptionUpdateRequest(StreamsGroupTopologyDescript
 
     @Override
     public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
+        ApiError apiError = ApiError.fromThrowable(e);
         return new StreamsGroupTopologyDescriptionUpdateResponse(
             new StreamsGroupTopologyDescriptionUpdateResponseData()
                 .setThrottleTimeMs(throttleTimeMs)
-                .setErrorCode(Errors.forException(e).code())
-                .setErrorMessage(Errors.forException(e).message())
+                .setErrorCode(apiError.error().code())
+                .setErrorMessage(apiError.message())
         );
     }
 

clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupTopologyDescriptionUpdateResponse.java

@@ -23,6 +23,19 @@
 
 import java.util.Map;
 
+/**
+ * Possible error codes.
+ *
+ * - {@link Errors#GROUP_AUTHORIZATION_FAILED}
+ * - {@link Errors#NOT_COORDINATOR}
+ * - {@link Errors#COORDINATOR_NOT_AVAILABLE}
+ * - {@link Errors#COORDINATOR_LOAD_IN_PROGRESS}
+ * - {@link Errors#INVALID_REQUEST}
+ * - {@link Errors#UNSUPPORTED_VERSION}
+ * - {@link Errors#UNKNOWN_MEMBER_ID}
+ * - {@link Errors#GROUP_ID_NOT_FOUND}
+ * - {@link Errors#STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED}
+ */
 public class StreamsGroupTopologyDescriptionUpdateResponse extends AbstractResponse {
 
     private final StreamsGroupTopologyDescriptionUpdateResponseData data;

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

@@ -110,9 +110,9 @@
           "about": "32-bit bitfield to represent authorized operations for this group." },
         { "name": "TopologyDescription", "type": "TopologyDescription", "versions": "1+",
           "nullableVersions": "1+", "default": "null",
-          "about": "Full topology description retrieved from the topology description plugin. Null if the client did not request it, no plugin is configured, no description is stored for the group, or retrieval failed." },
+          "about": "The full topology description for this group. Non-null if and only if TopologyDescriptionStatus is AVAILABLE (3); null otherwise." },
         { "name": "TopologyDescriptionStatus", "type": "int8", "versions": "1+", "default": "0",
-          "about": "The status of the topology description for this group: 0=NOT_REQUESTED (client did not set IncludeTopologyDescription), 1=NOT_STORED (no topology description has been recorded for this group), 2=ERROR (the broker failed to fetch the topology description; check broker logs), 3=AVAILABLE (a topology description is present in the TopologyDescription field). The broker MUST set this field to AVAILABLE whenever it attaches a TopologyDescription." }
+          "about": "The status of the topology description for this group, paired with TopologyDescription: 0=NOT_REQUESTED (client did not set IncludeTopologyDescription; TopologyDescription is null); 1=NOT_STORED (no description recorded for this group; TopologyDescription is null); 2=ERROR (broker failed to fetch the description, see broker logs; TopologyDescription is null); 3=AVAILABLE (TopologyDescription is non-null and carries the description)." }
       ]
     }
   ],

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

@@ -18,7 +18,9 @@
   "type": "request",
   "listeners": ["broker"],
   "name": "StreamsGroupHeartbeatRequest",
-  "validVersions": "0",
+  // Version 1 is the same as version 0; bumped together with StreamsGroupHeartbeatResponse v1,
+  // which adds TopologyDescriptionRequired (KIP-1331). Required so the response v1 is negotiated.
+  "validVersions": "0-1",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",

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

@@ -66,7 +66,7 @@
       "about": "Assigned warm-up tasks for this client. Null if unchanged since last heartbeat." },
 
     { "name": "TopologyDescriptionRequired", "type": "bool", "versions": "1+", "default": "false",
-      "about": "True if the broker's topology description plugin does not have an up-to-date topology description for this group. The client should send the topology description via StreamsGroupTopologyDescriptionUpdate." },
+      "about": "True if the broker does not have an up-to-date topology description for this group. The client should send the topology description via StreamsGroupTopologyDescriptionUpdate." },
 
     // IQ-related information
     { "name": "EndpointInformationEpoch", "type": "int32", "versions": "0+",

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

@@ -18,6 +18,9 @@
   "type": "request",
   "listeners": ["broker"],
   "name": "StreamsGroupTopologyDescriptionUpdateRequest",
+  // The broker handler is not yet implemented (deferred to a later sub-task of KIP-1331),
+  // so the latest version is marked unstable to suppress ApiVersions advertisement.
+  "latestVersionUnstable": true,
   "validVersions": "0",
   "flexibleVersions": "0+",
   "fields": [

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

@@ -19,6 +19,16 @@
   "name": "StreamsGroupTopologyDescriptionUpdateResponse",
   "validVersions": "0",
   "flexibleVersions": "0+",
+  // Supported errors:
+  // - GROUP_AUTHORIZATION_FAILED (version 0+)
+  // - NOT_COORDINATOR (version 0+)
+  // - COORDINATOR_NOT_AVAILABLE (version 0+)
+  // - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
+  // - INVALID_REQUEST (version 0+)
+  // - UNSUPPORTED_VERSION (version 0+)
+  // - UNKNOWN_MEMBER_ID (version 0+)
+  // - GROUP_ID_NOT_FOUND (version 0+)
+  // - STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED (version 0+)
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
       "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },