Skip to content
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

KAFKA-18521: Cleanup NodeApiVersions zkMigrationEnabled field #18535

Merged
merged 4 commits into from
Jan 16, 2025
Merged

KAFKA-18521: Cleanup NodeApiVersions zkMigrationEnabled field #18535

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
809d0db
remove all field and refactor constructor
m1a2st Jan 14, 2025
a13d0f9
fix the test
m1a2st Jan 14, 2025
f0d31d7
fix the test
m1a2st Jan 14, 2025
75ade66
fix the test
m1a2st Jan 14, 2025
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
1 change: 0 additions & 1 deletion clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1047,7 +1047,6 @@ private void handleApiVersionsResponse(List<ClientResponse> responses,
NodeApiVersions nodeVersionInfo = new NodeApiVersions(
apiVersionsResponse.data().apiKeys(),
apiVersionsResponse.data().supportedFeatures(),
apiVersionsResponse.data().zkMigrationReady(),
apiVersionsResponse.data().finalizedFeatures(),
apiVersionsResponse.data().finalizedFeaturesEpoch());
apiVersions.update(node, nodeVersionInfo);
Expand Down
16 changes: 3 additions & 13 deletions clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,8 +49,6 @@ public class NodeApiVersions {

private final Map<String, SupportedVersionRange> supportedFeatures;

private final boolean zkMigrationEnabled;

private final Map<String, Short> finalizedFeatures;

private final long finalizedFeaturesEpoch;
Expand Down Expand Up @@ -83,7 +81,7 @@ public static NodeApiVersions create(Collection<ApiVersion> overrides) {
}
if (!exists) apiVersions.add(ApiVersionsResponse.toApiVersion(apiKey));
}
return new NodeApiVersions(apiVersions, Collections.emptyList(), false, Collections.emptyList(), -1);
return new NodeApiVersions(apiVersions, Collections.emptyList(), Collections.emptyList(), -1);
}


Expand All @@ -104,16 +102,14 @@ public static NodeApiVersions create(short apiKey, short minVersion, short maxVe

public NodeApiVersions(
Collection<ApiVersion> nodeApiVersions,
Collection<SupportedFeatureKey> nodeSupportedFeatures,
boolean zkMigrationEnabled
Collection<SupportedFeatureKey> nodeSupportedFeatures
) {
this(nodeApiVersions, nodeSupportedFeatures, zkMigrationEnabled, Collections.emptyList(), -1);
this(nodeApiVersions, nodeSupportedFeatures, Collections.emptyList(), -1);
}

public NodeApiVersions(
Collection<ApiVersion> nodeApiVersions,
Collection<SupportedFeatureKey> nodeSupportedFeatures,
boolean zkMigrationEnabled,
Collection<ApiVersionsResponseData.FinalizedFeatureKey> nodeFinalizedFeatures,
long finalizedFeaturesEpoch
) {
Expand All @@ -133,8 +129,6 @@ public NodeApiVersions(
new SupportedVersionRange(supportedFeature.minVersion(), supportedFeature.maxVersion()));
}
this.supportedFeatures = Collections.unmodifiableMap(supportedFeaturesBuilder);
this.zkMigrationEnabled = zkMigrationEnabled;

this.finalizedFeaturesEpoch = finalizedFeaturesEpoch;
this.finalizedFeatures = new HashMap<>();
for (ApiVersionsResponseData.FinalizedFeatureKey finalizedFeature : nodeFinalizedFeatures) {
Expand Down Expand Up @@ -264,10 +258,6 @@ public Map<String, SupportedVersionRange> supportedFeatures() {
return supportedFeatures;
}

public boolean zkMigrationEnabled() {
return zkMigrationEnabled;
}

public Map<String, Short> finalizedFeatures() {
return finalizedFeatures;
}
Expand Down
2 changes: 0 additions & 2 deletions clients/src/test/java/org/apache/kafka/clients/ApiVersionsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,6 @@ public void testFinalizedFeaturesUpdate() {
.setName("transaction.version")
.setMaxVersion((short) 2)
.setMinVersion((short) 0)),
false,
Arrays.asList(new ApiVersionsResponseData.FinalizedFeatureKey()
.setName("transaction.version")
.setMaxVersionLevel((short) 2)
Expand All @@ -53,7 +52,6 @@ public void testFinalizedFeaturesUpdate() {
.setName("transaction.version")
.setMaxVersion((short) 2)
.setMinVersion((short) 0)),
false,
Arrays.asList(new ApiVersionsResponseData.FinalizedFeatureKey()
.setName("transaction.version")
.setMaxVersionLevel((short) 1)
Expand Down
11 changes: 5 additions & 6 deletions clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ public class NodeApiVersionsTest {

@Test
public void testUnsupportedVersionsToString() {
NodeApiVersions versions = new NodeApiVersions(new ApiVersionCollection(), Collections.emptyList(), false);
NodeApiVersions versions = new NodeApiVersions(new ApiVersionCollection(), Collections.emptyList());
StringBuilder bld = new StringBuilder();
String prefix = "(";
for (ApiKeys apiKey : ApiKeys.clientApis()) {
Expand Down Expand Up @@ -73,7 +73,7 @@ public void testVersionsToString() {
.setMaxVersion((short) 10001));
} else versionList.add(ApiVersionsResponse.toApiVersion(apiKey));
}
NodeApiVersions versions = new NodeApiVersions(versionList, Collections.emptyList(), false);
NodeApiVersions versions = new NodeApiVersions(versionList, Collections.emptyList());
StringBuilder bld = new StringBuilder();
String prefix = "(";
for (ApiKeys apiKey : ApiKeys.values()) {
Expand Down Expand Up @@ -130,7 +130,7 @@ public void testLatestUsableVersionOutOfRangeHigh() {

@Test
public void testUsableVersionCalculationNoKnownVersions() {
NodeApiVersions versions = new NodeApiVersions(new ApiVersionCollection(), Collections.emptyList(), false);
NodeApiVersions versions = new NodeApiVersions(new ApiVersionCollection(), Collections.emptyList());
assertThrows(UnsupportedVersionException.class,
() -> versions.latestUsableVersion(ApiKeys.FETCH));
}
Expand All @@ -152,7 +152,7 @@ public void testUsableVersionLatestVersions(ApiMessageType.ListenerType scope) {
.setApiKey((short) 100)
.setMinVersion((short) 0)
.setMaxVersion((short) 1));
NodeApiVersions versions = new NodeApiVersions(versionList, Collections.emptyList(), false);
NodeApiVersions versions = new NodeApiVersions(versionList, Collections.emptyList());
for (ApiKeys apiKey: ApiKeys.apisForListener(scope)) {
assertEquals(apiKey.latestVersion(), versions.latestUsableVersion(apiKey));
}
Expand All @@ -162,7 +162,7 @@ public void testUsableVersionLatestVersions(ApiMessageType.ListenerType scope) {
@EnumSource(ApiMessageType.ListenerType.class)
public void testConstructionFromApiVersionsResponse(ApiMessageType.ListenerType scope) {
ApiVersionsResponse apiVersionsResponse = TestUtils.defaultApiVersionsResponse(scope);
NodeApiVersions versions = new NodeApiVersions(apiVersionsResponse.data().apiKeys(), Collections.emptyList(), false);
NodeApiVersions versions = new NodeApiVersions(apiVersionsResponse.data().apiKeys(), Collections.emptyList());

for (ApiVersion apiVersionKey : apiVersionsResponse.data().apiKeys()) {
ApiVersion apiVersion = versions.apiVersion(ApiKeys.forId(apiVersionKey.apiKey()));
Expand All @@ -180,7 +180,6 @@ public void testFeatures() {
.setName("transaction.version")
.setMaxVersion((short) 2)
.setMinVersion((short) 0)),
false,
Arrays.asList(new ApiVersionsResponseData.FinalizedFeatureKey()
.setName("transaction.version")
.setMaxVersionLevel((short) 2)
Expand Down
2 changes: 0 additions & 2 deletions clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1634,7 +1634,6 @@ public void testSendTxnOffsetsWithGroupIdTransactionV2() {
.setName("transaction.version")
.setMaxVersion((short) 2)
.setMinVersion((short) 0)),
false,
Arrays.asList(new ApiVersionsResponseData.FinalizedFeatureKey()
.setName("transaction.version")
.setMaxVersionLevel((short) 2)
Expand Down Expand Up @@ -1701,7 +1700,6 @@ public void testTransactionV2Produce() throws Exception {
.setName("transaction.version")
.setMaxVersion((short) 2)
.setMinVersion((short) 0)),
false,
Arrays.asList(new ApiVersionsResponseData.FinalizedFeatureKey()
.setName("transaction.version")
.setMaxVersionLevel((short) 2)
Expand Down
6 changes: 0 additions & 6 deletions ...nts/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -182,7 +182,6 @@ private void initializeTransactionManager(Optional<String> transactionalId, bool
.setName("transaction.version")
.setMaxVersion(transactionV2Enabled ? (short) 2 : (short) 1)
.setMinVersion((short) 0)),
false,
Arrays.asList(new ApiVersionsResponseData.FinalizedFeatureKey()
.setName("transaction.version")
.setMaxVersionLevel(transactionV2Enabled ? (short) 2 : (short) 1)
Expand Down Expand Up @@ -930,7 +929,6 @@ public void testTransactionManagerEnablesV2() {
.setName("transaction.version")
.setMaxVersion((short) 2)
.setMinVersion((short) 0)),
false,
Arrays.asList(new ApiVersionsResponseData.FinalizedFeatureKey()
.setName("transaction.version")
.setMaxVersionLevel((short) 2)
Expand Down Expand Up @@ -1035,7 +1033,6 @@ public void testTransactionManagerDisablesV2() {
.setName("transaction.version")
.setMaxVersion((short) 1)
.setMinVersion((short) 0)),
false,
Arrays.asList(new ApiVersionsResponseData.FinalizedFeatureKey()
.setName("transaction.version")
.setMaxVersionLevel((short) 1)
Expand Down Expand Up @@ -2972,7 +2969,6 @@ public void testTransitionToFatalErrorWhenRetriedBatchIsExpired() throws Interru
.setMinVersion((short) 0)
.setMaxVersion((short) 7)),
Collections.emptyList(),
false,
Collections.emptyList(),
0));

Expand Down Expand Up @@ -3267,7 +3263,6 @@ public void testAbortTransactionAndReuseSequenceNumberOnError() throws Interrupt
.setMinVersion((short) 0)
.setMaxVersion((short) 7)),
Collections.emptyList(),
false,
Collections.emptyList(),
0));

Expand Down Expand Up @@ -3328,7 +3323,6 @@ public void testAbortTransactionAndResetSequenceNumberOnUnknownProducerId() thro
.setMinVersion((short) 0)
.setMaxVersion((short) 4)),
Collections.emptyList(),
false,
Collections.emptyList(),
0));

Expand Down
2 changes: 1 addition & 1 deletion tools/src/main/java/org/apache/kafka/tools/BrokerApiVersionsCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -249,7 +249,7 @@ protected KafkaFuture<NodeApiVersions> getNodeApiVersions(Node node) {
if (error.exception() != null) {
future.completeExceptionally(error.exception());
} else {
future.complete(new NodeApiVersions(response.data().apiKeys(), response.data().supportedFeatures(), response.data().zkMigrationReady()));
future.complete(new NodeApiVersions(response.data().apiKeys(), response.data().supportedFeatures()));
}
} catch (Exception e) {
future.completeExceptionally(e);
Expand Down
3 changes: 1 addition & 2 deletions tools/src/test/java/org/apache/kafka/tools/BrokerApiVersionsCommandTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,8 +60,7 @@ public void testBrokerApiVersionsCommandOutput(ClusterInstance clusterInstance)

NodeApiVersions nodeApiVersions = new NodeApiVersions(
ApiVersionsResponse.collectApis(ApiKeys.clientApis(), true),
Collections.emptyList(),
false);
Collections.emptyList());
Iterator<ApiKeys> apiKeysIter = ApiKeys.clientApis().iterator();
while (apiKeysIter.hasNext()) {
ApiKeys apiKey = apiKeysIter.next();
Expand Down
Loading