diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java index 04ee2014d1d44..0df15382a96d9 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java @@ -37,6 +37,7 @@ import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.stream.Collectors; /** * Possible error codes: @@ -258,11 +259,25 @@ private static ApiVersionsResponseData createApiVersionsResponseData( final long finalizedFeaturesEpoch, final boolean zkMigrationEnabled ) { + // Older versions do not support 0 version in SupportedVersionRange. If an older broker receives a response with a zero version, it will fail with + // IllegalArgumentException. In order to not block upgrades, filter out features with 0 versions from the response. + // Future versions will bump the ApiVersionsRequest to handle this case and return features with version 0, but that version bump is not in 3.8. + Features backwardsCompatibleFeatures = Features.supportedFeatures(latestSupportedFeatures.features().entrySet() + .stream().filter(entry -> { + SupportedVersionRange supportedVersionRange = entry.getValue(); + return supportedVersionRange.min() != 0; + }) + .collect(Collectors.toMap( + entry -> entry.getKey(), + entry -> entry.getValue() + )) + ); + final ApiVersionsResponseData data = new ApiVersionsResponseData(); data.setThrottleTimeMs(throttleTimeMs); data.setErrorCode(error.code()); data.setApiKeys(apiKeys); - data.setSupportedFeatures(createSupportedFeatureKeys(latestSupportedFeatures)); + data.setSupportedFeatures(createSupportedFeatureKeys(backwardsCompatibleFeatures)); data.setFinalizedFeatures(createFinalizedFeatureKeys(finalizedFeatures)); data.setFinalizedFeaturesEpoch(finalizedFeaturesEpoch); data.setZkMigrationReady(zkMigrationEnabled); diff --git a/clients/src/test/java/org/apache/kafka/common/requests/ApiVersionsResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/ApiVersionsResponseTest.java index cf832050f2a5d..7776aa34945a5 100644 --- a/clients/src/test/java/org/apache/kafka/common/requests/ApiVersionsResponseTest.java +++ b/clients/src/test/java/org/apache/kafka/common/requests/ApiVersionsResponseTest.java @@ -275,6 +275,26 @@ public void testIntersect() { assertEquals(expected, ApiVersionsResponse.intersect(other, thisVersion).get()); } + @Test + public void testZeroVersionNotReturned() { + String featureName = "test.feature.version"; + ApiVersionsResponse response = ApiVersionsResponse.createApiVersionsResponse( + 10, + RecordVersion.V1, + Features.supportedFeatures(Collections.singletonMap(featureName, new SupportedVersionRange((short) 0, (short) 1))), + Collections.emptyMap(), + ApiVersionsResponse.UNKNOWN_FINALIZED_FEATURES_EPOCH, + null, + ListenerType.BROKER, + true, + false, + true + ); + + // Feature should not be in the supported features due to the 0 version. + assertEquals(null, response.data().supportedFeatures().find(featureName)); + } + private void verifyVersions(short forwardableAPIKey, short minVersion, short maxVersion, diff --git a/core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala b/core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala index b79cf12aa4b33..482d76e5f937b 100644 --- a/core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala @@ -26,7 +26,7 @@ import org.apache.kafka.common.protocol.ApiKeys import org.apache.kafka.common.record.RecordVersion import org.apache.kafka.common.requests.{ApiVersionsRequest, ApiVersionsResponse, RequestUtils} import org.apache.kafka.common.utils.Utils -import org.apache.kafka.server.common.{GroupVersion, MetadataVersion} +import org.apache.kafka.server.common.MetadataVersion import org.apache.kafka.test.TestUtils import org.junit.jupiter.api.Assertions._ import org.junit.jupiter.api.Tag @@ -69,12 +69,10 @@ abstract class AbstractApiVersionsRequestTest(cluster: ClusterInstance) { assertEquals(MetadataVersion.latestTesting().featureLevel(), apiVersionsResponse.data().finalizedFeatures().find(MetadataVersion.FEATURE_NAME).minVersionLevel()) assertEquals(MetadataVersion.latestTesting().featureLevel(), apiVersionsResponse.data().finalizedFeatures().find(MetadataVersion.FEATURE_NAME).maxVersionLevel()) - assertEquals(2, apiVersionsResponse.data().supportedFeatures().size()) + // Since the min version of group version is 0, it is not included in the response. + assertEquals(1, apiVersionsResponse.data().supportedFeatures().size()) assertEquals(MetadataVersion.MINIMUM_KRAFT_VERSION.featureLevel(), apiVersionsResponse.data().supportedFeatures().find(MetadataVersion.FEATURE_NAME).minVersion()) assertEquals(MetadataVersion.latestTesting().featureLevel(), apiVersionsResponse.data().supportedFeatures().find(MetadataVersion.FEATURE_NAME).maxVersion()) - - assertEquals(0, apiVersionsResponse.data().supportedFeatures().find(GroupVersion.FEATURE_NAME).minVersion()) - assertEquals(GroupVersion.GV_1.featureLevel(), apiVersionsResponse.data().supportedFeatures().find(GroupVersion.FEATURE_NAME).maxVersion()) } val expectedApis = if (!cluster.isKRaftTest) { ApiVersionsResponse.collectApis( diff --git a/tools/src/test/java/org/apache/kafka/tools/FeatureCommandTest.java b/tools/src/test/java/org/apache/kafka/tools/FeatureCommandTest.java index b6d4cea0edbc9..f90903b615047 100644 --- a/tools/src/test/java/org/apache/kafka/tools/FeatureCommandTest.java +++ b/tools/src/test/java/org/apache/kafka/tools/FeatureCommandTest.java @@ -64,12 +64,9 @@ public void testDescribeWithKRaft(ClusterInstance cluster) { List features = Arrays.stream(commandOutput.split("\n")).sorted().collect(Collectors.toList()); - assertEquals("Feature: group.version\tSupportedMinVersion: 0\t" + - "SupportedMaxVersion: 1\tFinalizedVersionLevel: 0\t", outputWithoutEpoch(features.get(0))); - // Change expected message to reflect latest MetadataVersion (SupportedMaxVersion increases when adding a new version) assertEquals("Feature: metadata.version\tSupportedMinVersion: 3.0-IV1\t" + - "SupportedMaxVersion: 4.0-IV0\tFinalizedVersionLevel: 3.3-IV1\t", outputWithoutEpoch(features.get(1))); + "SupportedMaxVersion: 4.0-IV0\tFinalizedVersionLevel: 3.3-IV1\t", outputWithoutEpoch(features.get(0))); } @ClusterTest(types = {Type.KRAFT}, metadataVersion = MetadataVersion.IBP_3_7_IV4) @@ -80,12 +77,9 @@ public void testDescribeWithKRaftAndBootstrapControllers(ClusterInstance cluster List features = Arrays.stream(commandOutput.split("\n")).sorted().collect(Collectors.toList()); - assertEquals("Feature: group.version\tSupportedMinVersion: 0\t" + - "SupportedMaxVersion: 1\tFinalizedVersionLevel: 0\t", outputWithoutEpoch(features.get(0))); - // Change expected message to reflect latest MetadataVersion (SupportedMaxVersion increases when adding a new version) assertEquals("Feature: metadata.version\tSupportedMinVersion: 3.0-IV1\t" + - "SupportedMaxVersion: 4.0-IV0\tFinalizedVersionLevel: 3.7-IV4\t", outputWithoutEpoch(features.get(1))); + "SupportedMaxVersion: 4.0-IV0\tFinalizedVersionLevel: 3.7-IV4\t", outputWithoutEpoch(features.get(0))); } @ClusterTest(types = {Type.ZK}, metadataVersion = MetadataVersion.IBP_3_3_IV1)