Skip to content

KAFKA-17011: SupportedFeatures.MinVersion incorrectly blocks v0 (3.8) #16420

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
jolshan wants to merge 5 commits into from
Closed

KAFKA-17011: SupportedFeatures.MinVersion incorrectly blocks v0 (3.8) #16420

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
1cd3967
Do not return 0 in apiversons response
jolshan Jun 21, 2024
38a8b07
Add unit test
jolshan Jun 21, 2024
6b897b4
Suppress 0 versions for older versions
jolshan Jun 22, 2024
58a9cfc
test fixes
jolshan Jun 24, 2024
1faf2ab
Add comment
jolshan Jun 25, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 16 additions & 1 deletion clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;

/**
* Possible error codes:
Expand Down Expand Up @@ -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<SupportedVersionRange> backwardsCompatibleFeatures = Features.supportedFeatures(latestSupportedFeatures.features().entrySet()
.stream().filter(entry -> {
SupportedVersionRange supportedVersionRange = entry.getValue();
return supportedVersionRange.min() != 0;

@chia7712 chia7712 Jun 24, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

not sure whether I fail to catch your description.

so for now we will change 0 to 1 in the response in order to be backwards compatible.

the code looks like it gets rid of "0" instead of changing to from 0 to 1.

@chia7712 chia7712 Jun 24, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For another, does this PR mean Admin#describeFeatures can't see the feature "group.version" from the broker running in 3.8.0?

@jolshan jolshan Jun 24, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hey sorry, the comment was from the old implementation, the new way is to omit as you mentioned.
I believe describeFeatures will not be able to show the version for this release.

@junrao junrao Jun 25, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Perhaps it's useful to add a comment on why we are filtering version 0 features.

})
.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);
Expand Down
20 changes: 20 additions & 0 deletions clients/src/test/java/org/apache/kafka/common/requests/ApiVersionsResponseTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
8 changes: 3 additions & 5 deletions core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand Down
10 changes: 2 additions & 8 deletions tools/src/test/java/org/apache/kafka/tools/FeatureCommandTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,12 +64,9 @@ public void testDescribeWithKRaft(ClusterInstance cluster) {

List<String> 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)
Expand All @@ -80,12 +77,9 @@ public void testDescribeWithKRaftAndBootstrapControllers(ClusterInstance cluster

List<String> 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)
Expand Down