KAFKA-18157: Consider UnsupportedVersionException child class to represent the case of unsupported fields

alanlau28 · alanlau28 · commit 7c52efb0246e · 2026-05-27T14:35:17.000-04:00
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java
@@ -21,6 +21,7 @@
 import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler;
 import org.apache.kafka.clients.consumer.internals.metrics.HeartbeatMetricsManager;
 import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.UnsupportedProtocolFieldException;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
 import org.apache.kafka.common.metrics.Metrics;
@@ -100,12 +101,14 @@ public boolean handleSpecificFailure(Throwable exception) {
         String errorMessage = exception.getMessage();
         if (exception instanceof UnsupportedVersionException) {
             String message = CONSUMER_PROTOCOL_NOT_SUPPORTED_MSG;
-            if (errorMessage.equals(REGEX_RESOLUTION_NOT_SUPPORTED_MSG)) {
+            if (exception instanceof UnsupportedProtocolFieldException) {
                 message = REGEX_RESOLUTION_NOT_SUPPORTED_MSG;
                 logger.error("{} regex resolution not supported: {}", heartbeatRequestName(), message);
             } else {
                 logger.error("{} failed due to unsupported version while sending request: {}", heartbeatRequestName(), errorMessage);
             }
+            // Surface the parent type here: handleFatalFailure propagates this via BackgroundEvent
+            // to the user-facing API. Propagating the subclass would be a user-visible behavior change.
             handleFatalFailure(new UnsupportedVersionException(message, exception));
             errorHandled = true;
         }
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/UnsupportedProtocolFieldException.java b/clients/src/main/java/org/apache/kafka/common/errors/UnsupportedProtocolFieldException.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.errors;
+
+public class UnsupportedProtocolFieldException extends UnsupportedVersionException {
+    private static final long serialVersionUID = 1L;
+
+    public UnsupportedProtocolFieldException(String fieldOrValue, String apiKeyName,
+                                             int apiVersion, int lowestSupportedVersion) {
+        super("The cluster does not support [" + fieldOrValue + "] in " + apiKeyName
+            + " API version " + apiVersion + ". Upgrade the cluster to " + apiKeyName
+            + " API version >= " + lowestSupportedVersion + " to enable [" + fieldOrValue + "].");
+    }
+
+    public UnsupportedProtocolFieldException(String message) {
+        super(message);
+    }
+}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ConsumerGroupHeartbeatRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ConsumerGroupHeartbeatRequest.java
@@ -16,7 +16,7 @@
  */
 package org.apache.kafka.common.requests;
 
-import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.errors.UnsupportedProtocolFieldException;
 import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
 import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -63,7 +63,7 @@ public Builder(ConsumerGroupHeartbeatRequestData data, boolean enableUnstableLas
         @Override
         public ConsumerGroupHeartbeatRequest build(short version) {
             if (version == 0 && data.subscribedTopicRegex() != null) {
-                throw new UnsupportedVersionException(REGEX_RESOLUTION_NOT_SUPPORTED_MSG);
+                throw new UnsupportedProtocolFieldException(REGEX_RESOLUTION_NOT_SUPPORTED_MSG);
             }
             return new ConsumerGroupHeartbeatRequest(data, version);
         }
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreateAclsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/CreateAclsRequest.java
@@ -21,7 +21,7 @@
 import org.apache.kafka.common.acl.AclBinding;
 import org.apache.kafka.common.acl.AclOperation;
 import org.apache.kafka.common.acl.AclPermissionType;
-import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.errors.UnsupportedProtocolFieldException;
 import org.apache.kafka.common.message.CreateAclsRequestData;
 import org.apache.kafka.common.message.CreateAclsRequestData.AclCreation;
 import org.apache.kafka.common.message.CreateAclsResponseData;
@@ -32,8 +32,10 @@
 import org.apache.kafka.common.resource.ResourcePattern;
 import org.apache.kafka.common.resource.ResourceType;
 
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.stream.Collectors;
 
 public class CreateAclsRequest extends AbstractRequest {
 
@@ -90,8 +92,13 @@ private void validate(CreateAclsRequestData data) {
         if (version() == 0) {
             final boolean unsupported = data.creations().stream().anyMatch(creation ->
                 creation.resourcePatternType() != PatternType.LITERAL.code());
-            if (unsupported)
-                throw new UnsupportedVersionException("Version 0 only supports literal resource pattern types");
+            if (unsupported) {
+                String unsupportedType = Arrays.stream(PatternType.values())
+                    .filter(type -> type != PatternType.LITERAL)
+                    .map(PatternType::name)
+                    .collect(Collectors.joining(","));
+                throw new UnsupportedProtocolFieldException(unsupportedType, apiKey().name(), version(), 1);
+            }
         }
 
         final boolean unknown = data.creations().stream().anyMatch(creation ->
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java
@@ -16,7 +16,7 @@
  */
 package org.apache.kafka.common.requests;
 
-import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.errors.UnsupportedProtocolFieldException;
 import org.apache.kafka.common.message.CreateTopicsRequestData;
 import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
 import org.apache.kafka.common.message.CreateTopicsResponseData;
@@ -39,8 +39,7 @@ public Builder(CreateTopicsRequestData data) {
         @Override
         public CreateTopicsRequest build(short version) {
             if (data.validateOnly() && version == 0)
-                throw new UnsupportedVersionException("validateOnly is not supported in version 0 of " +
-                        "CreateTopicsRequest");
+                throw new UnsupportedProtocolFieldException("validateOnly", apiKey().name(), version, 1);
 
             final List<String> topicsWithDefaults = data.topics()
                 .stream()
@@ -52,10 +51,8 @@ public CreateTopicsRequest build(short version) {
                 .collect(Collectors.toList());
 
             if (!topicsWithDefaults.isEmpty() && version < 4) {
-                throw new UnsupportedVersionException("Creating topics with default "
-                    + "partitions/replication factor are only supported in CreateTopicRequest "
-                    + "version 4+. The following topics need values for partitions and replicas: "
-                    + topicsWithDefaults);
+                throw new UnsupportedProtocolFieldException(String.join(",", topicsWithDefaults),
+                    apiKey().name(), version, 4);
             }
 
             return new CreateTopicsRequest(data, version);
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsRequest.java
@@ -20,7 +20,7 @@
 import org.apache.kafka.common.acl.AclBindingFilter;
 import org.apache.kafka.common.acl.AclOperation;
 import org.apache.kafka.common.acl.AclPermissionType;
-import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.errors.UnsupportedProtocolFieldException;
 import org.apache.kafka.common.message.DeleteAclsRequestData;
 import org.apache.kafka.common.message.DeleteAclsRequestData.DeleteAclsFilter;
 import org.apache.kafka.common.message.DeleteAclsResponseData;
@@ -31,6 +31,7 @@
 import org.apache.kafka.common.resource.ResourcePatternFilter;
 import org.apache.kafka.common.resource.ResourceType;
 
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
@@ -76,9 +77,13 @@ private void normalizeAndValidate() {
                 // to LITERAL. Note that the wildcard `*` is considered `LITERAL` for compatibility reasons.
                 if (patternType == PatternType.ANY)
                     filter.setPatternTypeFilter(PatternType.LITERAL.code());
-                else if (patternType != PatternType.LITERAL)
-                    throw new UnsupportedVersionException("Version 0 does not support pattern type " +
-                            patternType + " (only LITERAL and ANY are supported)");
+                else if (patternType != PatternType.LITERAL) {
+                    String unsupportedTypes = Arrays.stream(PatternType.values())
+                        .filter(type -> type != PatternType.ANY && type != PatternType.LITERAL)
+                        .map(PatternType::name)
+                        .collect(Collectors.joining(","));
+                    throw new UnsupportedProtocolFieldException(unsupportedTypes, apiKey().name(), version(), 1);
+                }
             }
         }
 
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsRequest.java
@@ -20,7 +20,7 @@
 import org.apache.kafka.common.acl.AclBindingFilter;
 import org.apache.kafka.common.acl.AclOperation;
 import org.apache.kafka.common.acl.AclPermissionType;
-import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.errors.UnsupportedProtocolFieldException;
 import org.apache.kafka.common.message.DescribeAclsRequestData;
 import org.apache.kafka.common.message.DescribeAclsResponseData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -29,6 +29,9 @@
 import org.apache.kafka.common.resource.ResourcePatternFilter;
 import org.apache.kafka.common.resource.ResourceType;
 
+import java.util.Arrays;
+import java.util.stream.Collectors;
+
 public class DescribeAclsRequest extends AbstractRequest {
 
     public static class Builder extends AbstractRequest.Builder<DescribeAclsRequest> {
@@ -75,8 +78,13 @@ private void normalizeAndValidate(short version) {
             // to LITERAL. Note that the wildcard `*` is considered `LITERAL` for compatibility reasons.
             if (patternType == PatternType.ANY)
                 data.setPatternTypeFilter(PatternType.LITERAL.code());
-            else if (patternType != PatternType.LITERAL)
-                throw new UnsupportedVersionException("Version 0 only supports literal resource pattern types");
+            else if (patternType != PatternType.LITERAL) {
+                String unsupportedTypes = Arrays.stream(PatternType.values())
+                    .filter(type -> type != PatternType.LITERAL)
+                    .map(PatternType::name)
+                    .collect(Collectors.joining(","));
+                throw new UnsupportedProtocolFieldException(unsupportedTypes, apiKey().name(), version, 1);
+            }
         }
 
         if (data.patternTypeFilter() == PatternType.UNKNOWN.code()
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ElectLeadersRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ElectLeadersRequest.java
@@ -19,7 +19,7 @@
 
 import org.apache.kafka.common.ElectionType;
 import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.errors.UnsupportedProtocolFieldException;
 import org.apache.kafka.common.message.ElectLeadersRequestData;
 import org.apache.kafka.common.message.ElectLeadersRequestData.TopicPartitions;
 import org.apache.kafka.common.message.ElectLeadersResponseData.PartitionResult;
@@ -63,7 +63,7 @@ public String toString() {
 
         private ElectLeadersRequestData toRequestData(short version) {
             if (electionType != ElectionType.PREFERRED && version == 0) {
-                throw new UnsupportedVersionException("API Version 0 only supports PREFERRED election type");
+                throw new UnsupportedProtocolFieldException("ElectionType", apiKey().name(), version, 1);
             }
 
             ElectLeadersRequestData data = new ElectLeadersRequestData()
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/FindCoordinatorRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/FindCoordinatorRequest.java
@@ -18,6 +18,7 @@
 
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.errors.InvalidRequestException;
+import org.apache.kafka.common.errors.UnsupportedProtocolFieldException;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.message.FindCoordinatorRequestData;
 import org.apache.kafka.common.message.FindCoordinatorResponseData;
@@ -43,8 +44,8 @@ public Builder(FindCoordinatorRequestData data) {
         @Override
         public FindCoordinatorRequest build(short version) {
             if (version < 1 && data.keyType() == CoordinatorType.TRANSACTION.id()) {
-                throw new UnsupportedVersionException("Cannot create a v" + version + " FindCoordinator request " +
-                        "because we require features supported only in 2 or later.");
+                throw new UnsupportedProtocolFieldException(CoordinatorType.TRANSACTION.name(),
+                    apiKey().name(), version, 2);
             }
             int batchedKeys = data.coordinatorKeys().size();
             if (version < MIN_BATCHED_VERSION) {
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java
@@ -16,7 +16,7 @@
  */
 package org.apache.kafka.common.requests;
 
-import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.errors.UnsupportedProtocolFieldException;
 import org.apache.kafka.common.message.HeartbeatRequestData;
 import org.apache.kafka.common.message.HeartbeatResponseData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -36,8 +36,7 @@ public Builder(HeartbeatRequestData data) {
         @Override
         public HeartbeatRequest build(short version) {
             if (data.groupInstanceId() != null && version < 3) {
-                throw new UnsupportedVersionException("The broker heartbeat protocol version " +
-                        version + " does not support usage of config group.instance.id.");
+                throw new UnsupportedProtocolFieldException("GroupInstanceId", apiKey().name(), version, 3);
             }
             return new HeartbeatRequest(data, version);
         }
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java
@@ -17,7 +17,7 @@
 package org.apache.kafka.common.requests;
 
 import org.apache.kafka.common.errors.InvalidConfigurationException;
-import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.errors.UnsupportedProtocolFieldException;
 import org.apache.kafka.common.internals.Topic;
 import org.apache.kafka.common.message.JoinGroupRequestData;
 import org.apache.kafka.common.message.JoinGroupResponseData;
@@ -42,8 +42,7 @@ public Builder(JoinGroupRequestData data) {
         @Override
         public JoinGroupRequest build(short version) {
             if (data.groupInstanceId() != null && version < 5) {
-                throw new UnsupportedVersionException("The broker join group protocol version " +
-                        version + " does not support usage of config group.instance.id.");
+                throw new UnsupportedProtocolFieldException("GroupInstanceId", apiKey().name(), version, 5);
             }
             return new JoinGroupRequest(data, version);
         }
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ListGroupsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ListGroupsRequest.java
@@ -17,7 +17,7 @@
 package org.apache.kafka.common.requests;
 
 import org.apache.kafka.common.GroupType;
-import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.errors.UnsupportedProtocolFieldException;
 import org.apache.kafka.common.message.ListGroupsRequestData;
 import org.apache.kafka.common.message.ListGroupsResponseData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -48,8 +48,7 @@ public Builder(ListGroupsRequestData data) {
         @Override
         public ListGroupsRequest build(short version) {
             if (!data.statesFilter().isEmpty() && version < 4) {
-                throw new UnsupportedVersionException("The broker only supports ListGroups " +
-                        "v" + version + ", but we need v4 or newer to request groups by states.");
+                throw new UnsupportedProtocolFieldException("StatesFilter", apiKey().name(), version, 4);
             }
             if (!data.typesFilter().isEmpty() && version < 5) {
                 // Types filter is supported by brokers with version 3.8.0 or later. Older brokers only support
@@ -60,9 +59,7 @@ public ListGroupsRequest build(short version) {
                 boolean containedClassic = typesCopy.remove(GroupType.CLASSIC.toString());
                 boolean containedConsumer = typesCopy.remove(GroupType.CONSUMER.toString());
                 if (!typesCopy.isEmpty() || (!containedClassic && containedConsumer)) {
-                    throw new UnsupportedVersionException("The broker only supports ListGroups " +
-                        "v" + version + ", but we need v5 or newer to request groups by type. " +
-                        "Requested group types: [" + String.join(", ", data.typesFilter()) + "].");
+                    throw new UnsupportedProtocolFieldException("TypesFilter", apiKey().name(), version, 5);
                 }
                 return new ListGroupsRequest(data.duplicate().setTypesFilter(List.of()), version);
             }
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ListTransactionsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ListTransactionsRequest.java
@@ -16,7 +16,7 @@
  */
 package org.apache.kafka.common.requests;
 
-import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.errors.UnsupportedProtocolFieldException;
 import org.apache.kafka.common.message.ListTransactionsRequestData;
 import org.apache.kafka.common.message.ListTransactionsResponseData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -35,12 +35,10 @@ public Builder(ListTransactionsRequestData data) {
         @Override
         public ListTransactionsRequest build(short version) {
             if (data.durationFilter() >= 0 && version < 1) {
-                throw new UnsupportedVersionException("Duration filter can be set only when using API version 1 or higher." +
-                        " If client is connected to an older broker, do not specify duration filter or set duration filter to -1.");
+                throw new UnsupportedProtocolFieldException("DurationFilter", apiKey().name(), version, 1);
             }
             if (data.transactionalIdPattern() != null && version < 2) {
-                throw new UnsupportedVersionException("Transactional ID pattern filter can be set only when using API version 2 or higher." +
-                    " If client is connected to an older broker, do not specify the pattern filter.");
+                throw new UnsupportedProtocolFieldException("TransactionalIdPattern", apiKey().name(), version, 2);
             }
             return new ListTransactionsRequest(data, version);
         }
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/SyncGroupRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/SyncGroupRequest.java
