You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "apalan60 (via GitHub)" <gi...@apache.org> on 2025/08/11 17:57:51 UTC
[PR] KAFKA-19384: The passing of BrokerRegistrationRequestTest is a false positive [kafka]
apalan60 opened a new pull request, #20338:
URL: https://github.com/apache/kafka/pull/20338
Fixes a false positive in `BrokerRegistrationRequestTest` caused by `isMigratingZkBroker`, and migrates the test from Scala to Java.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19384: The passing of BrokerRegistrationRequestTest is a false positive [kafka]
Posted by "apalan60 (via GitHub)" <gi...@apache.org>.
apalan60 commented on PR #20338:
URL: https://github.com/apache/kafka/pull/20338#issuecomment-3193390481
@chia7712
Updated again based on the latest feedback. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19384: The passing of BrokerRegistrationRequestTest is a false positive [kafka]
Posted by "apalan60 (via GitHub)" <gi...@apache.org>.
apalan60 commented on PR #20338:
URL: https://github.com/apache/kafka/pull/20338#issuecomment-3193415533
Seems the refactored test is causing a thread leak. I’m looking into a fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19384: The passing of BrokerRegistrationRequestTest is a false positive [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20338:
URL: https://github.com/apache/kafka/pull/20338#discussion_r2279742785
##########
server/src/test/java/org/apache/kafka/server/BrokerRegistrationRequestTest.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * 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.server;
+
+import kafka.network.SocketServer;
+import kafka.server.ControllerInformation;
+import kafka.server.ControllerNodeProvider;
+import kafka.server.ControllerServer;
+import kafka.server.NodeToControllerChannelManagerImpl;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.message.BrokerRegistrationRequestData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractRequest;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.BrokerRegistrationRequest;
+import org.apache.kafka.common.requests.BrokerRegistrationResponse;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.Feature;
+import org.apache.kafka.server.common.MetadataVersion;
+import org.apache.kafka.server.common.MetadataVersionTestUtils;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+
+import org.junit.jupiter.api.Assertions;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import scala.Option;
+
+/**
+ * This test simulates a broker registering with the KRaft quorum under different configurations.
+ */
+public class BrokerRegistrationRequestTest {
+
+ public NodeToControllerChannelManager brokerToControllerChannelManager(ClusterInstance clusterInstance) {
+ var controllerSocketServer = clusterInstance.controllers().values().stream()
+ .map(ControllerServer::socketServer)
+ .findFirst()
+ .orElseThrow();
+
+ return new NodeToControllerChannelManagerImpl(
+ new TestControllerNodeProvider(controllerSocketServer, clusterInstance),
+ Time.SYSTEM,
+ new Metrics(),
+ controllerSocketServer.config(),
+ "heartbeat",
+ "test-heartbeat-",
+ 10000L
+ );
+ }
+
+ @SuppressWarnings("unchecked")
+ public <T extends AbstractRequest, R extends AbstractResponse> R sendAndReceive(
+ NodeToControllerChannelManager channelManager,
+ AbstractRequest.Builder<T> reqBuilder,
+ int timeoutMs
+ ) throws Exception {
+ var responseFuture = new CompletableFuture<R>();
+ channelManager.sendRequest(reqBuilder, new ControllerRequestCompletionHandler() {
+ @Override
+ public void onTimeout() {
+ responseFuture.completeExceptionally(new TimeoutException());
+ }
+
+ @Override
+ public void onComplete(ClientResponse response) {
+ responseFuture.complete((R) response.responseBody());
+ }
+ });
+ return responseFuture.get(timeoutMs, TimeUnit.MILLISECONDS);
+ }
+
+ public Errors registerBroker(
+ NodeToControllerChannelManager channelManager,
+ String clusterId,
+ int brokerId,
+ Long zkEpoch,
+ FeatureLevel featureLevelToSend
+ ) throws Exception {
+ var features = new BrokerRegistrationRequestData.FeatureCollection();
+
+ if (featureLevelToSend != null) {
+ features.add(new BrokerRegistrationRequestData.Feature()
+ .setName(MetadataVersion.FEATURE_NAME)
+ .setMinSupportedVersion(featureLevelToSend.min())
+ .setMaxSupportedVersion(featureLevelToSend.max())
+ );
+ }
+
+ Feature.PRODUCTION_FEATURES.stream()
+ .filter(feature -> !feature.featureName().equals(MetadataVersion.FEATURE_NAME))
+ .forEach(feature -> features.add(new BrokerRegistrationRequestData.Feature()
+ .setName(feature.featureName())
+ .setMinSupportedVersion(feature.minimumProduction())
+ .setMaxSupportedVersion(feature.latestTesting())));
+
+ var listener = new BrokerRegistrationRequestData.Listener()
+ .setName("EXTERNAL")
+ .setHost("example.com")
+ .setPort(8082)
+ .setSecurityProtocol(SecurityProtocol.PLAINTEXT.id);
+
+ var req = new BrokerRegistrationRequestData()
+ .setBrokerId(brokerId)
+ .setLogDirs(List.of(Uuid.randomUuid()))
+ .setClusterId(clusterId)
+ .setIncarnationId(Uuid.randomUuid())
+ .setIsMigratingZkBroker(zkEpoch != null)
+ .setFeatures(features)
+ .setListeners(new BrokerRegistrationRequestData.ListenerCollection(List.of(listener).iterator()));
+
+ BrokerRegistrationResponse resp = this.sendAndReceive(
+ channelManager,
+ new BrokerRegistrationRequest.Builder(req),
+ 30000
+ );
+ return Errors.forCode(resp.data().errorCode());
+ }
+
+ @ClusterTest(types = {Type.KRAFT}, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3)
+ public void shouldRejectZkMigratingBrokerWhenFeatureLevelDoesNotSupportMigration(ClusterInstance clusterInstance) throws Exception {
+ var clusterId = clusterInstance.clusterId();
+ var channelManager = brokerToControllerChannelManager(clusterInstance);
+ try {
+ channelManager.start();
+ Assertions.assertEquals(
+ Errors.BROKER_ID_NOT_REGISTERED,
+ registerBroker(channelManager, clusterId, 100, 1L,
+ new FeatureLevel(MetadataVersionTestUtils.IBP_3_3_IV0_FEATURE_LEVEL, MetadataVersion.IBP_3_3_IV3.featureLevel()))
+ );
+ } finally {
+ channelManager.shutdown();
+ }
+ }
+
+ @ClusterTest(types = {Type.KRAFT}, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3)
+ public void shouldRejectRegistrationWithoutFeatureLevels(ClusterInstance clusterInstance) throws Exception {
+ var clusterId = clusterInstance.clusterId();
+ var channelManager = brokerToControllerChannelManager(clusterInstance);
+ try {
+ channelManager.start();
+ Assertions.assertEquals(
+ Errors.INVALID_REGISTRATION,
+ registerBroker(channelManager, clusterId, 100, null, null)
+ );
+ } finally {
+ channelManager.shutdown();
+ }
+ }
+
+ @ClusterTest(types = {Type.KRAFT}, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3)
+ public void shouldRejectRegistrationWhenFeatureLevelTooHigh(ClusterInstance clusterInstance) throws Exception {
+ var clusterId = clusterInstance.clusterId();
+ var channelManager = brokerToControllerChannelManager(clusterInstance);
+ try {
+ channelManager.start();
+ Assertions.assertEquals(
+ Errors.UNSUPPORTED_VERSION,
+ registerBroker(channelManager, clusterId, 100, null,
+ new FeatureLevel(MetadataVersion.IBP_3_4_IV0.featureLevel(), MetadataVersion.IBP_3_4_IV0.featureLevel()))
+ );
+ } finally {
+ channelManager.shutdown();
+ }
+ }
+
+ @ClusterTest(types = {Type.KRAFT}, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3)
+ public void shouldRegisterWhenSupportedRangeAndNotMigrating(ClusterInstance clusterInstance) throws Exception {
+ var clusterId = clusterInstance.clusterId();
+ var channelManager = brokerToControllerChannelManager(clusterInstance);
+ try {
+ channelManager.start();
+ Assertions.assertEquals(
+ Errors.NONE,
+ registerBroker(channelManager, clusterId, 100, null,
+ new FeatureLevel(MetadataVersion.IBP_3_3_IV3.featureLevel(), MetadataVersion.IBP_3_4_IV0.featureLevel()))
+ );
+ } finally {
+ channelManager.shutdown();
+ }
+ }
+
+ public record FeatureLevel(short min, short max) { }
+
+ private record TestControllerNodeProvider(SocketServer controllerSocketServer, ClusterInstance clusterInstance)
+ implements ControllerNodeProvider {
+
+ public Optional<Node> node() {
+ return Optional.of(new Node(
Review Comment:
```java
public Optional<Node> node() {
return Optional.of(new Node(
clusterInstance.controllers().keySet().iterator().next(),
"127.0.0.1",
clusterInstance.controllerBoundPorts().get(0)
));
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19384: The passing of BrokerRegistrationRequestTest is a false positive [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20338:
URL: https://github.com/apache/kafka/pull/20338#discussion_r2279764302
##########
server/src/test/java/org/apache/kafka/server/BrokerRegistrationRequestTest.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * 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.server;
+
+import kafka.network.SocketServer;
+import kafka.server.ControllerInformation;
+import kafka.server.ControllerNodeProvider;
+import kafka.server.ControllerServer;
+import kafka.server.NodeToControllerChannelManagerImpl;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.message.BrokerRegistrationRequestData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractRequest;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.BrokerRegistrationRequest;
+import org.apache.kafka.common.requests.BrokerRegistrationResponse;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.Feature;
+import org.apache.kafka.server.common.MetadataVersion;
+import org.apache.kafka.server.common.MetadataVersionTestUtils;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+
+import org.junit.jupiter.api.Assertions;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import scala.Option;
+
+/**
+ * This test simulates a broker registering with the KRaft quorum under different configurations.
+ */
+public class BrokerRegistrationRequestTest {
+
+ public NodeToControllerChannelManager brokerToControllerChannelManager(ClusterInstance clusterInstance) {
+ var controllerSocketServer = clusterInstance.controllers().values().stream()
+ .map(ControllerServer::socketServer)
+ .findFirst()
+ .orElseThrow();
+
+ return new NodeToControllerChannelManagerImpl(
+ new TestControllerNodeProvider(controllerSocketServer, clusterInstance),
+ Time.SYSTEM,
+ new Metrics(),
Review Comment:
Could you please close the metrics manually? `NodeToControllerChannelManagerImpl` does not handle it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19384: The passing of BrokerRegistrationRequestTest is a false positive [kafka]
Posted by "apalan60 (via GitHub)" <gi...@apache.org>.
apalan60 commented on code in PR #20338:
URL: https://github.com/apache/kafka/pull/20338#discussion_r2267600515
##########
core/src/test/scala/unit/kafka/server/BrokerRegistrationRequestTest.scala:
##########
@@ -154,13 +154,13 @@ class BrokerRegistrationRequestTest {
// No features (MV) sent with registration, controller can't verify
assertEquals(
- Errors.BROKER_ID_NOT_REGISTERED,
- registerBroker(channelManager, clusterId, 100, Some(1), None))
Review Comment:
Previously, passing `Some(1)` as the `zkEpoch` parameter caused `zkEpoch.isDefined` to always be `true`, which led to a [BrokerIdNotRegisteredException](https://github.com/apache/kafka/blob/634e99e9ab16db61990f4922f0a30b6a93f27785/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java#L374C8-L376C10) being thrown.
As a result, the intended test scenario was never actually tested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19384: The passing of BrokerRegistrationRequestTest is a false positive [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #20338:
URL: https://github.com/apache/kafka/pull/20338
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19384: The passing of BrokerRegistrationRequestTest is a false positive [kafka]
Posted by "apalan60 (via GitHub)" <gi...@apache.org>.
apalan60 commented on code in PR #20338:
URL: https://github.com/apache/kafka/pull/20338#discussion_r2267616966
##########
core/src/test/scala/unit/kafka/server/BrokerRegistrationRequestTest.scala:
##########
@@ -154,13 +154,13 @@ class BrokerRegistrationRequestTest {
// No features (MV) sent with registration, controller can't verify
assertEquals(
- Errors.BROKER_ID_NOT_REGISTERED,
- registerBroker(channelManager, clusterId, 100, Some(1), None))
+ Errors.INVALID_REGISTRATION,
+ registerBroker(channelManager, clusterId, 100, None, None))
// Given MV is too high for controller to support
assertEquals(
- Errors.BROKER_ID_NOT_REGISTERED,
- registerBroker(channelManager, clusterId, 100, Some(1), Some((MetadataVersion.IBP_3_4_IV0.featureLevel, MetadataVersion.IBP_3_4_IV0.featureLevel))))
+ Errors.UNSUPPORTED_VERSION,
+ registerBroker(channelManager, clusterId, 100, None, Some((MetadataVersion.IBP_3_4_IV0.featureLevel, MetadataVersion.IBP_3_4_IV0.featureLevel))))
Review Comment:
Previously, passing `Some(1)` as the `zkEpoch` parameter caused `zkEpoch.isDefined` to always be `true`, which led to a [BrokerIdNotRegisteredException](https://github.com/apache/kafka/blob/634e99e9ab16db61990f4922f0a30b6a93f27785/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java#L374C8-L376C10) being thrown.
As a result, the intended test scenario was never actually tested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19384: The passing of BrokerRegistrationRequestTest is a false positive [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20338:
URL: https://github.com/apache/kafka/pull/20338#discussion_r2274997289
##########
core/src/test/java/kafka/server/BrokerRegistrationRequestTest.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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 kafka.server;
+
+import kafka.network.SocketServer;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.message.BrokerRegistrationRequestData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractRequest;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.BrokerRegistrationRequest;
+import org.apache.kafka.common.requests.BrokerRegistrationResponse;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.Feature;
+import org.apache.kafka.server.common.MetadataVersion;
+import org.apache.kafka.server.common.MetadataVersionTestUtils;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+
+import org.junit.jupiter.api.Assertions;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import scala.Option;
+
+/**
+ * This test simulates a broker registering with the KRaft quorum under different configurations.
+ */
+public class BrokerRegistrationRequestTest {
+
+ public NodeToControllerChannelManager brokerToControllerChannelManager(ClusterInstance clusterInstance) {
+ var controllerSocketServer = clusterInstance.controllers().values().stream()
+ .map(ControllerServer::socketServer)
+ .findFirst()
+ .orElseThrow();
+
+ return new NodeToControllerChannelManagerImpl(
+ new TestControllerNodeProvider(controllerSocketServer, clusterInstance),
+ Time.SYSTEM,
+ new Metrics(),
+ controllerSocketServer.config(),
+ "heartbeat",
+ "test-heartbeat-",
+ 10000L
+ );
+ }
+
+ @SuppressWarnings("unchecked")
+ public <T extends AbstractRequest, R extends AbstractResponse> R sendAndReceive(
+ NodeToControllerChannelManager channelManager,
+ AbstractRequest.Builder<T> reqBuilder,
+ int timeoutMs
+ ) throws Exception {
+ var responseFuture = new CompletableFuture<R>();
+ channelManager.sendRequest(reqBuilder, new ControllerRequestCompletionHandler() {
+ @Override
+ public void onTimeout() {
+ responseFuture.completeExceptionally(new TimeoutException());
+ }
+
+ @Override
+ public void onComplete(ClientResponse response) {
+ responseFuture.complete((R) response.responseBody());
+ }
+ });
+ return responseFuture.get(timeoutMs, TimeUnit.MILLISECONDS);
+ }
+
+ public Errors registerBroker(
+ NodeToControllerChannelManager channelManager,
+ String clusterId,
+ int brokerId,
+ Long zkEpoch,
+ FeatureLevel featureLevelToSend
+ ) throws Exception {
+ var features = new BrokerRegistrationRequestData.FeatureCollection();
+
+ if (featureLevelToSend != null) {
+ features.add(new BrokerRegistrationRequestData.Feature()
+ .setName(MetadataVersion.FEATURE_NAME)
+ .setMinSupportedVersion(featureLevelToSend.min())
+ .setMaxSupportedVersion(featureLevelToSend.max())
+ );
+ }
+
+ Feature.PRODUCTION_FEATURES.stream()
+ .filter(feature -> !feature.featureName().equals(MetadataVersion.FEATURE_NAME))
+ .forEach(feature -> features.add(new BrokerRegistrationRequestData.Feature()
+ .setName(feature.featureName())
+ .setMinSupportedVersion(feature.minimumProduction())
+ .setMaxSupportedVersion(feature.latestTesting())));
+
+ var listener = new BrokerRegistrationRequestData.Listener()
+ .setName("EXTERNAL")
+ .setHost("example.com")
+ .setPort(8082)
+ .setSecurityProtocol(SecurityProtocol.PLAINTEXT.id);
+
+ var req = new BrokerRegistrationRequestData()
+ .setBrokerId(brokerId)
+ .setLogDirs(List.of(Uuid.randomUuid()))
+ .setClusterId(clusterId)
+ .setIncarnationId(Uuid.randomUuid())
+ .setIsMigratingZkBroker(zkEpoch != null)
+ .setFeatures(features)
+ .setListeners(new BrokerRegistrationRequestData.ListenerCollection(List.of(listener).iterator()));
+
+ BrokerRegistrationResponse resp = this.sendAndReceive(
+ channelManager,
+ new BrokerRegistrationRequest.Builder(req),
+ 30000
+ );
+ return Errors.forCode(resp.data().errorCode());
+ }
+
+ @ClusterTest(types = {Type.KRAFT}, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3)
+ public void testRegisterZkWith33Controller(ClusterInstance clusterInstance) throws Exception {
Review Comment:
This test name is no longer accurate, since the RPCs are not all related to ZK. Perhaps you could split the scenarios into separate test cases?
##########
core/src/test/java/kafka/server/BrokerRegistrationRequestTest.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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 kafka.server;
Review Comment:
Could you please move it to server module?
##########
core/src/test/java/kafka/server/BrokerRegistrationRequestTest.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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 kafka.server;
+
+import kafka.network.SocketServer;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.message.BrokerRegistrationRequestData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractRequest;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.BrokerRegistrationRequest;
+import org.apache.kafka.common.requests.BrokerRegistrationResponse;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.Feature;
+import org.apache.kafka.server.common.MetadataVersion;
+import org.apache.kafka.server.common.MetadataVersionTestUtils;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+
+import org.junit.jupiter.api.Assertions;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import scala.Option;
+
+/**
+ * This test simulates a broker registering with the KRaft quorum under different configurations.
+ */
+public class BrokerRegistrationRequestTest {
+
+ public NodeToControllerChannelManager brokerToControllerChannelManager(ClusterInstance clusterInstance) {
+ var controllerSocketServer = clusterInstance.controllers().values().stream()
+ .map(ControllerServer::socketServer)
+ .findFirst()
+ .orElseThrow();
+
+ return new NodeToControllerChannelManagerImpl(
+ new TestControllerNodeProvider(controllerSocketServer, clusterInstance),
+ Time.SYSTEM,
+ new Metrics(),
+ controllerSocketServer.config(),
+ "heartbeat",
+ "test-heartbeat-",
+ 10000L
+ );
+ }
+
+ @SuppressWarnings("unchecked")
+ public <T extends AbstractRequest, R extends AbstractResponse> R sendAndReceive(
+ NodeToControllerChannelManager channelManager,
+ AbstractRequest.Builder<T> reqBuilder,
+ int timeoutMs
+ ) throws Exception {
+ var responseFuture = new CompletableFuture<R>();
+ channelManager.sendRequest(reqBuilder, new ControllerRequestCompletionHandler() {
+ @Override
+ public void onTimeout() {
+ responseFuture.completeExceptionally(new TimeoutException());
+ }
+
+ @Override
+ public void onComplete(ClientResponse response) {
+ responseFuture.complete((R) response.responseBody());
+ }
+ });
+ return responseFuture.get(timeoutMs, TimeUnit.MILLISECONDS);
+ }
+
+ public Errors registerBroker(
+ NodeToControllerChannelManager channelManager,
+ String clusterId,
+ int brokerId,
+ Long zkEpoch,
+ FeatureLevel featureLevelToSend
+ ) throws Exception {
+ var features = new BrokerRegistrationRequestData.FeatureCollection();
+
+ if (featureLevelToSend != null) {
+ features.add(new BrokerRegistrationRequestData.Feature()
+ .setName(MetadataVersion.FEATURE_NAME)
+ .setMinSupportedVersion(featureLevelToSend.min())
+ .setMaxSupportedVersion(featureLevelToSend.max())
+ );
+ }
+
+ Feature.PRODUCTION_FEATURES.stream()
+ .filter(feature -> !feature.featureName().equals(MetadataVersion.FEATURE_NAME))
+ .forEach(feature -> features.add(new BrokerRegistrationRequestData.Feature()
+ .setName(feature.featureName())
+ .setMinSupportedVersion(feature.minimumProduction())
+ .setMaxSupportedVersion(feature.latestTesting())));
+
+ var listener = new BrokerRegistrationRequestData.Listener()
+ .setName("EXTERNAL")
+ .setHost("example.com")
+ .setPort(8082)
+ .setSecurityProtocol(SecurityProtocol.PLAINTEXT.id);
+
+ var req = new BrokerRegistrationRequestData()
+ .setBrokerId(brokerId)
+ .setLogDirs(List.of(Uuid.randomUuid()))
+ .setClusterId(clusterId)
+ .setIncarnationId(Uuid.randomUuid())
+ .setIsMigratingZkBroker(zkEpoch != null)
+ .setFeatures(features)
+ .setListeners(new BrokerRegistrationRequestData.ListenerCollection(List.of(listener).iterator()));
+
+ BrokerRegistrationResponse resp = this.sendAndReceive(
+ channelManager,
+ new BrokerRegistrationRequest.Builder(req),
+ 30000
+ );
+ return Errors.forCode(resp.data().errorCode());
+ }
+
+ @ClusterTest(types = {Type.KRAFT}, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3)
+ public void testRegisterZkWith33Controller(ClusterInstance clusterInstance) throws Exception {
+ // Verify that a controller running an old metadata.version cannot register a ZK broker
+ var clusterId = clusterInstance.clusterId();
+ var channelManager = brokerToControllerChannelManager(clusterInstance);
+ try {
+ channelManager.start();
+ // Invalid registration (isMigratingZkBroker, but MV does not support migrations)
+ Assertions.assertEquals(
+ Errors.BROKER_ID_NOT_REGISTERED,
+ registerBroker(channelManager, clusterId, 100, 1L,
+ new FeatureLevel(MetadataVersionTestUtils.IBP_3_3_IV0_FEATURE_LEVEL, MetadataVersionTestUtils.IBP_3_3_IV3_FEATURE_LEVEL))
+ );
+
+ // No features (MV) sent with registration, controller can't verify
+ Assertions.assertEquals(
+ Errors.INVALID_REGISTRATION,
+ registerBroker(channelManager, clusterId, 100, null, null)
+ );
+
+ // Given MV is too high for controller to support
+ Assertions.assertEquals(
+ Errors.UNSUPPORTED_VERSION,
+ registerBroker(channelManager, clusterId, 100, null,
+ new FeatureLevel(MetadataVersionTestUtils.IBP_3_4_IV0_FEATURE_LEVEL, MetadataVersionTestUtils.IBP_3_4_IV0_FEATURE_LEVEL))
+ );
+
+ // Controller supports this MV and isMigratingZkBroker is false, so this one works
+ Assertions.assertEquals(
+ Errors.NONE,
+ registerBroker(channelManager, clusterId, 100, null,
+ new FeatureLevel(MetadataVersionTestUtils.IBP_3_3_IV3_FEATURE_LEVEL, MetadataVersionTestUtils.IBP_3_4_IV0_FEATURE_LEVEL))
Review Comment:
ditto to `IBP_3_4_IV0_FEATURE_LEVEL`
##########
core/src/test/java/kafka/server/BrokerRegistrationRequestTest.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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 kafka.server;
+
+import kafka.network.SocketServer;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.message.BrokerRegistrationRequestData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractRequest;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.BrokerRegistrationRequest;
+import org.apache.kafka.common.requests.BrokerRegistrationResponse;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.Feature;
+import org.apache.kafka.server.common.MetadataVersion;
+import org.apache.kafka.server.common.MetadataVersionTestUtils;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+
+import org.junit.jupiter.api.Assertions;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import scala.Option;
+
+/**
+ * This test simulates a broker registering with the KRaft quorum under different configurations.
+ */
+public class BrokerRegistrationRequestTest {
+
+ public NodeToControllerChannelManager brokerToControllerChannelManager(ClusterInstance clusterInstance) {
+ var controllerSocketServer = clusterInstance.controllers().values().stream()
+ .map(ControllerServer::socketServer)
+ .findFirst()
+ .orElseThrow();
+
+ return new NodeToControllerChannelManagerImpl(
+ new TestControllerNodeProvider(controllerSocketServer, clusterInstance),
+ Time.SYSTEM,
+ new Metrics(),
+ controllerSocketServer.config(),
+ "heartbeat",
+ "test-heartbeat-",
+ 10000L
+ );
+ }
+
+ @SuppressWarnings("unchecked")
+ public <T extends AbstractRequest, R extends AbstractResponse> R sendAndReceive(
+ NodeToControllerChannelManager channelManager,
+ AbstractRequest.Builder<T> reqBuilder,
+ int timeoutMs
+ ) throws Exception {
+ var responseFuture = new CompletableFuture<R>();
+ channelManager.sendRequest(reqBuilder, new ControllerRequestCompletionHandler() {
+ @Override
+ public void onTimeout() {
+ responseFuture.completeExceptionally(new TimeoutException());
+ }
+
+ @Override
+ public void onComplete(ClientResponse response) {
+ responseFuture.complete((R) response.responseBody());
+ }
+ });
+ return responseFuture.get(timeoutMs, TimeUnit.MILLISECONDS);
+ }
+
+ public Errors registerBroker(
+ NodeToControllerChannelManager channelManager,
+ String clusterId,
+ int brokerId,
+ Long zkEpoch,
+ FeatureLevel featureLevelToSend
+ ) throws Exception {
+ var features = new BrokerRegistrationRequestData.FeatureCollection();
+
+ if (featureLevelToSend != null) {
+ features.add(new BrokerRegistrationRequestData.Feature()
+ .setName(MetadataVersion.FEATURE_NAME)
+ .setMinSupportedVersion(featureLevelToSend.min())
+ .setMaxSupportedVersion(featureLevelToSend.max())
+ );
+ }
+
+ Feature.PRODUCTION_FEATURES.stream()
+ .filter(feature -> !feature.featureName().equals(MetadataVersion.FEATURE_NAME))
+ .forEach(feature -> features.add(new BrokerRegistrationRequestData.Feature()
+ .setName(feature.featureName())
+ .setMinSupportedVersion(feature.minimumProduction())
+ .setMaxSupportedVersion(feature.latestTesting())));
+
+ var listener = new BrokerRegistrationRequestData.Listener()
+ .setName("EXTERNAL")
+ .setHost("example.com")
+ .setPort(8082)
+ .setSecurityProtocol(SecurityProtocol.PLAINTEXT.id);
+
+ var req = new BrokerRegistrationRequestData()
+ .setBrokerId(brokerId)
+ .setLogDirs(List.of(Uuid.randomUuid()))
+ .setClusterId(clusterId)
+ .setIncarnationId(Uuid.randomUuid())
+ .setIsMigratingZkBroker(zkEpoch != null)
+ .setFeatures(features)
+ .setListeners(new BrokerRegistrationRequestData.ListenerCollection(List.of(listener).iterator()));
+
+ BrokerRegistrationResponse resp = this.sendAndReceive(
+ channelManager,
+ new BrokerRegistrationRequest.Builder(req),
+ 30000
+ );
+ return Errors.forCode(resp.data().errorCode());
+ }
+
+ @ClusterTest(types = {Type.KRAFT}, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3)
+ public void testRegisterZkWith33Controller(ClusterInstance clusterInstance) throws Exception {
+ // Verify that a controller running an old metadata.version cannot register a ZK broker
+ var clusterId = clusterInstance.clusterId();
+ var channelManager = brokerToControllerChannelManager(clusterInstance);
+ try {
+ channelManager.start();
+ // Invalid registration (isMigratingZkBroker, but MV does not support migrations)
+ Assertions.assertEquals(
+ Errors.BROKER_ID_NOT_REGISTERED,
+ registerBroker(channelManager, clusterId, 100, 1L,
+ new FeatureLevel(MetadataVersionTestUtils.IBP_3_3_IV0_FEATURE_LEVEL, MetadataVersionTestUtils.IBP_3_3_IV3_FEATURE_LEVEL))
+ );
+
+ // No features (MV) sent with registration, controller can't verify
+ Assertions.assertEquals(
+ Errors.INVALID_REGISTRATION,
+ registerBroker(channelManager, clusterId, 100, null, null)
+ );
+
+ // Given MV is too high for controller to support
+ Assertions.assertEquals(
+ Errors.UNSUPPORTED_VERSION,
+ registerBroker(channelManager, clusterId, 100, null,
+ new FeatureLevel(MetadataVersionTestUtils.IBP_3_4_IV0_FEATURE_LEVEL, MetadataVersionTestUtils.IBP_3_4_IV0_FEATURE_LEVEL))
+ );
+
+ // Controller supports this MV and isMigratingZkBroker is false, so this one works
+ Assertions.assertEquals(
+ Errors.NONE,
+ registerBroker(channelManager, clusterId, 100, null,
+ new FeatureLevel(MetadataVersionTestUtils.IBP_3_3_IV3_FEATURE_LEVEL, MetadataVersionTestUtils.IBP_3_4_IV0_FEATURE_LEVEL))
Review Comment:
`MetadataVersionTestUtils.IBP_3_3_IV3_FEATURE_LEVEL` -> `MetadataVersion.IBP_3_3_IV3.featureLevel()`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19384: The passing of BrokerRegistrationRequestTest is a false positive [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20338:
URL: https://github.com/apache/kafka/pull/20338#discussion_r2280340833
##########
server/src/test/java/org/apache/kafka/server/BrokerRegistrationRequestTest.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * 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.server;
+
+import kafka.server.ControllerInformation;
+import kafka.server.ControllerNodeProvider;
+import kafka.server.ControllerServer;
+import kafka.server.NodeToControllerChannelManagerImpl;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.message.BrokerRegistrationRequestData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractRequest;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.BrokerRegistrationRequest;
+import org.apache.kafka.common.requests.BrokerRegistrationResponse;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.Feature;
+import org.apache.kafka.server.common.MetadataVersion;
+import org.apache.kafka.server.common.MetadataVersionTestUtils;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import scala.Option;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * This test simulates a broker registering with the KRaft quorum under different configurations.
+ */
+class BrokerRegistrationRequestTest {
+ private final ClusterInstance clusterInstance;
+ public BrokerRegistrationRequestTest(ClusterInstance clusterInstance) {
+ this.clusterInstance = clusterInstance;
+ }
+
+ @ClusterTest(types = {Type.KRAFT}, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3)
+ public void shouldRejectZkMigratingBrokerWhenFeatureLevelDoesNotSupportMigration() throws Exception {
+ try (ChannelEnv env = openChannel()) {
+ assertEquals(
+ Errors.BROKER_ID_NOT_REGISTERED,
+ registerBroker(env.channelManager, clusterInstance.clusterId(), 1L,
+ new FeatureLevel(MetadataVersionTestUtils.IBP_3_3_IV0_FEATURE_LEVEL, MetadataVersion.IBP_3_3_IV3.featureLevel()))
+ );
+ }
+ }
+
+ @ClusterTest(types = {Type.KRAFT}, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3)
+ public void shouldRejectRegistrationWithoutFeatureLevels() throws Exception {
+ try (ChannelEnv env = openChannel()) {
+ assertEquals(
+ Errors.INVALID_REGISTRATION,
+ registerBroker(env.channelManager, clusterInstance.clusterId(), null, null)
+ );
+ }
+ }
+
+ @ClusterTest(types = {Type.KRAFT}, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3)
+ public void shouldRejectRegistrationWhenFeatureLevelTooHigh() throws Exception {
+ try (ChannelEnv env = openChannel()) {
+ assertEquals(
+ Errors.UNSUPPORTED_VERSION,
+ registerBroker(env.channelManager, clusterInstance.clusterId(), null,
+ new FeatureLevel(MetadataVersion.IBP_3_4_IV0.featureLevel(), MetadataVersion.IBP_3_4_IV0.featureLevel()))
+ );
+ }
+ }
+
+ @ClusterTest(types = {Type.KRAFT}, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3)
+ public void shouldRegisterWhenSupportedRangeAndNotMigrating() throws Exception {
+ try (ChannelEnv env = openChannel()) {
+ assertEquals(
+ Errors.NONE,
+ registerBroker(env.channelManager, clusterInstance.clusterId(), null,
+ new FeatureLevel(MetadataVersion.IBP_3_3_IV3.featureLevel(), MetadataVersion.IBP_3_4_IV0.featureLevel()))
+ );
+ }
+ }
+
+ final class ChannelEnv implements AutoCloseable {
+ final Metrics metrics;
+ final NodeToControllerChannelManager channelManager;
+
+ ChannelEnv(ClusterInstance clusterInstance) {
+ this.metrics = new Metrics();
+ this.channelManager = brokerToControllerChannelManager(clusterInstance, metrics);
+ this.channelManager.start();
+ }
+
+ @Override
+ public void close() {
+ channelManager.shutdown();
+ metrics.close();
+ }
+ }
+
+ ChannelEnv openChannel() {
+ return new ChannelEnv(clusterInstance);
+ }
+
+ NodeToControllerChannelManager brokerToControllerChannelManager(ClusterInstance clusterInstance, Metrics metrics) {
+ var controllerSocketServer = clusterInstance.controllers().values().stream()
+ .map(ControllerServer::socketServer)
+ .findFirst()
+ .orElseThrow();
+
+ return new NodeToControllerChannelManagerImpl(
+ new TestControllerNodeProvider(clusterInstance),
+ Time.SYSTEM,
+ metrics,
+ controllerSocketServer.config(),
+ "heartbeat",
+ "test-heartbeat-",
+ 10000L
+ );
+ }
+
+ @SuppressWarnings("unchecked")
+ <T extends AbstractRequest, R extends AbstractResponse> R sendAndReceive(
+ NodeToControllerChannelManager channelManager,
+ AbstractRequest.Builder<T> reqBuilder
+ ) throws Exception {
+ var responseFuture = new CompletableFuture<R>();
+ channelManager.sendRequest(reqBuilder, new ControllerRequestCompletionHandler() {
+ @Override
+ public void onTimeout() {
+ responseFuture.completeExceptionally(new TimeoutException());
+ }
+
+ @Override
+ public void onComplete(ClientResponse response) {
+ responseFuture.complete((R) response.responseBody());
+ }
+ });
+ return responseFuture.get(30000, TimeUnit.MILLISECONDS);
+ }
+
+ Errors registerBroker(
+ NodeToControllerChannelManager channelManager,
+ String clusterId,
+ Long zkEpoch,
+ FeatureLevel featureLevelToSend
+ ) throws Exception {
+ var features = new BrokerRegistrationRequestData.FeatureCollection();
+
+ if (featureLevelToSend != null) {
+ features.add(new BrokerRegistrationRequestData.Feature()
+ .setName(MetadataVersion.FEATURE_NAME)
+ .setMinSupportedVersion(featureLevelToSend.min())
+ .setMaxSupportedVersion(featureLevelToSend.max())
+ );
+ }
+
+ Feature.PRODUCTION_FEATURES.stream()
+ .filter(feature -> !feature.featureName().equals(MetadataVersion.FEATURE_NAME))
+ .forEach(feature -> features.add(new BrokerRegistrationRequestData.Feature()
+ .setName(feature.featureName())
+ .setMinSupportedVersion(feature.minimumProduction())
+ .setMaxSupportedVersion(feature.latestTesting())));
+
+ var listener = new BrokerRegistrationRequestData.Listener()
+ .setName("EXTERNAL")
+ .setHost("example.com")
+ .setPort(8082)
+ .setSecurityProtocol(SecurityProtocol.PLAINTEXT.id);
+
+ var req = new BrokerRegistrationRequestData()
+ .setBrokerId(100)
+ .setLogDirs(List.of(Uuid.randomUuid()))
+ .setClusterId(clusterId)
+ .setIncarnationId(Uuid.randomUuid())
+ .setIsMigratingZkBroker(zkEpoch != null)
+ .setFeatures(features)
+ .setListeners(new BrokerRegistrationRequestData.ListenerCollection(List.of(listener).iterator()));
+
+ BrokerRegistrationResponse resp = this.sendAndReceive(
+ channelManager,
+ new BrokerRegistrationRequest.Builder(req)
+ );
+ return Errors.forCode(resp.data().errorCode());
+ }
+
+ record FeatureLevel(short min, short max) { }
Review Comment:
this could be replaced by `BrokerRegistrationRequestData.Feature`, shouldn't it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19384: The passing of BrokerRegistrationRequestTest is a false positive [kafka]
Posted by "apalan60 (via GitHub)" <gi...@apache.org>.
apalan60 commented on PR #20338:
URL: https://github.com/apache/kafka/pull/20338#issuecomment-3191764939
@chia7712
Thanks for the review! All comments addressed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19384: The passing of BrokerRegistrationRequestTest is a false positive [kafka]
Posted by "apalan60 (via GitHub)" <gi...@apache.org>.
apalan60 commented on code in PR #20338:
URL: https://github.com/apache/kafka/pull/20338#discussion_r2280382821
##########
server/src/test/java/org/apache/kafka/server/BrokerRegistrationRequestTest.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * 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.server;
+
+import kafka.server.ControllerInformation;
+import kafka.server.ControllerNodeProvider;
+import kafka.server.ControllerServer;
+import kafka.server.NodeToControllerChannelManagerImpl;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.message.BrokerRegistrationRequestData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractRequest;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.BrokerRegistrationRequest;
+import org.apache.kafka.common.requests.BrokerRegistrationResponse;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.Feature;
+import org.apache.kafka.server.common.MetadataVersion;
+import org.apache.kafka.server.common.MetadataVersionTestUtils;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import scala.Option;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * This test simulates a broker registering with the KRaft quorum under different configurations.
+ */
+class BrokerRegistrationRequestTest {
+ private final ClusterInstance clusterInstance;
+ public BrokerRegistrationRequestTest(ClusterInstance clusterInstance) {
+ this.clusterInstance = clusterInstance;
+ }
+
+ @ClusterTest(types = {Type.KRAFT}, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3)
+ public void shouldRejectZkMigratingBrokerWhenFeatureLevelDoesNotSupportMigration() throws Exception {
+ try (ChannelEnv env = openChannel()) {
+ assertEquals(
+ Errors.BROKER_ID_NOT_REGISTERED,
+ registerBroker(env.channelManager, clusterInstance.clusterId(), 1L,
+ new FeatureLevel(MetadataVersionTestUtils.IBP_3_3_IV0_FEATURE_LEVEL, MetadataVersion.IBP_3_3_IV3.featureLevel()))
+ );
+ }
+ }
+
+ @ClusterTest(types = {Type.KRAFT}, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3)
+ public void shouldRejectRegistrationWithoutFeatureLevels() throws Exception {
+ try (ChannelEnv env = openChannel()) {
+ assertEquals(
+ Errors.INVALID_REGISTRATION,
+ registerBroker(env.channelManager, clusterInstance.clusterId(), null, null)
+ );
+ }
+ }
+
+ @ClusterTest(types = {Type.KRAFT}, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3)
+ public void shouldRejectRegistrationWhenFeatureLevelTooHigh() throws Exception {
+ try (ChannelEnv env = openChannel()) {
+ assertEquals(
+ Errors.UNSUPPORTED_VERSION,
+ registerBroker(env.channelManager, clusterInstance.clusterId(), null,
+ new FeatureLevel(MetadataVersion.IBP_3_4_IV0.featureLevel(), MetadataVersion.IBP_3_4_IV0.featureLevel()))
+ );
+ }
+ }
+
+ @ClusterTest(types = {Type.KRAFT}, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3)
+ public void shouldRegisterWhenSupportedRangeAndNotMigrating() throws Exception {
+ try (ChannelEnv env = openChannel()) {
+ assertEquals(
+ Errors.NONE,
+ registerBroker(env.channelManager, clusterInstance.clusterId(), null,
+ new FeatureLevel(MetadataVersion.IBP_3_3_IV3.featureLevel(), MetadataVersion.IBP_3_4_IV0.featureLevel()))
+ );
+ }
+ }
+
+ final class ChannelEnv implements AutoCloseable {
+ final Metrics metrics;
+ final NodeToControllerChannelManager channelManager;
+
+ ChannelEnv(ClusterInstance clusterInstance) {
+ this.metrics = new Metrics();
+ this.channelManager = brokerToControllerChannelManager(clusterInstance, metrics);
+ this.channelManager.start();
+ }
+
+ @Override
+ public void close() {
+ channelManager.shutdown();
+ metrics.close();
+ }
+ }
+
+ ChannelEnv openChannel() {
+ return new ChannelEnv(clusterInstance);
+ }
+
+ NodeToControllerChannelManager brokerToControllerChannelManager(ClusterInstance clusterInstance, Metrics metrics) {
+ var controllerSocketServer = clusterInstance.controllers().values().stream()
+ .map(ControllerServer::socketServer)
+ .findFirst()
+ .orElseThrow();
+
+ return new NodeToControllerChannelManagerImpl(
+ new TestControllerNodeProvider(clusterInstance),
+ Time.SYSTEM,
+ metrics,
+ controllerSocketServer.config(),
+ "heartbeat",
+ "test-heartbeat-",
+ 10000L
+ );
+ }
+
+ @SuppressWarnings("unchecked")
+ <T extends AbstractRequest, R extends AbstractResponse> R sendAndReceive(
+ NodeToControllerChannelManager channelManager,
+ AbstractRequest.Builder<T> reqBuilder
+ ) throws Exception {
+ var responseFuture = new CompletableFuture<R>();
+ channelManager.sendRequest(reqBuilder, new ControllerRequestCompletionHandler() {
+ @Override
+ public void onTimeout() {
+ responseFuture.completeExceptionally(new TimeoutException());
+ }
+
+ @Override
+ public void onComplete(ClientResponse response) {
+ responseFuture.complete((R) response.responseBody());
+ }
+ });
+ return responseFuture.get(30000, TimeUnit.MILLISECONDS);
+ }
+
+ Errors registerBroker(
+ NodeToControllerChannelManager channelManager,
+ String clusterId,
+ Long zkEpoch,
+ FeatureLevel featureLevelToSend
+ ) throws Exception {
+ var features = new BrokerRegistrationRequestData.FeatureCollection();
+
+ if (featureLevelToSend != null) {
+ features.add(new BrokerRegistrationRequestData.Feature()
+ .setName(MetadataVersion.FEATURE_NAME)
+ .setMinSupportedVersion(featureLevelToSend.min())
+ .setMaxSupportedVersion(featureLevelToSend.max())
+ );
+ }
+
+ Feature.PRODUCTION_FEATURES.stream()
+ .filter(feature -> !feature.featureName().equals(MetadataVersion.FEATURE_NAME))
+ .forEach(feature -> features.add(new BrokerRegistrationRequestData.Feature()
+ .setName(feature.featureName())
+ .setMinSupportedVersion(feature.minimumProduction())
+ .setMaxSupportedVersion(feature.latestTesting())));
+
+ var listener = new BrokerRegistrationRequestData.Listener()
+ .setName("EXTERNAL")
+ .setHost("example.com")
+ .setPort(8082)
+ .setSecurityProtocol(SecurityProtocol.PLAINTEXT.id);
+
+ var req = new BrokerRegistrationRequestData()
+ .setBrokerId(100)
+ .setLogDirs(List.of(Uuid.randomUuid()))
+ .setClusterId(clusterId)
+ .setIncarnationId(Uuid.randomUuid())
+ .setIsMigratingZkBroker(zkEpoch != null)
+ .setFeatures(features)
+ .setListeners(new BrokerRegistrationRequestData.ListenerCollection(List.of(listener).iterator()));
+
+ BrokerRegistrationResponse resp = this.sendAndReceive(
+ channelManager,
+ new BrokerRegistrationRequest.Builder(req)
+ );
+ return Errors.forCode(resp.data().errorCode());
+ }
+
+ record FeatureLevel(short min, short max) { }
Review Comment:
Addressed this, thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19384: The passing of BrokerRegistrationRequestTest is a false positive [kafka]
Posted by "apalan60 (via GitHub)" <gi...@apache.org>.
apalan60 commented on code in PR #20338:
URL: https://github.com/apache/kafka/pull/20338#discussion_r2267609602
##########
core/src/test/scala/unit/kafka/server/BrokerRegistrationRequestTest.scala:
##########
@@ -154,13 +154,13 @@ class BrokerRegistrationRequestTest {
// No features (MV) sent with registration, controller can't verify
Review Comment:
Previously, passing `Some(1)` as the `zkEpoch` parameter caused `zkEpoch.isDefined` to always be `true`, which led to a [BrokerIdNotRegisteredException](https://github.com/apache/kafka/blob/634e99e9ab16db61990f4922f0a30b6a93f27785/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java#L374C8-L376C10) being thrown.
As a result, the intended test scenario was never actually tested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19384: The passing of BrokerRegistrationRequestTest is a false positive [kafka]
Posted by "apalan60 (via GitHub)" <gi...@apache.org>.
apalan60 commented on code in PR #20338:
URL: https://github.com/apache/kafka/pull/20338#discussion_r2267600515
##########
core/src/test/scala/unit/kafka/server/BrokerRegistrationRequestTest.scala:
##########
@@ -154,13 +154,13 @@ class BrokerRegistrationRequestTest {
// No features (MV) sent with registration, controller can't verify
assertEquals(
- Errors.BROKER_ID_NOT_REGISTERED,
- registerBroker(channelManager, clusterId, 100, Some(1), None))
Review Comment:
Previously, passing `Some(1)` as the `zkEpoch` parameter caused `zkEpoch.isDefined` to always be `true`, which led to a [BrokerIdNotRegisteredException](https://github.com/apache/kafka/blob/634e99e9ab16db61990f4922f0a30b6a93f27785/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java#L374C8-L376C10) being thrown.
As a result, the intended test scenario was never actually tested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19384: The passing of BrokerRegistrationRequestTest is a false positive [kafka]
Posted by "apalan60 (via GitHub)" <gi...@apache.org>.
apalan60 commented on code in PR #20338:
URL: https://github.com/apache/kafka/pull/20338#discussion_r2267609602
##########
core/src/test/scala/unit/kafka/server/BrokerRegistrationRequestTest.scala:
##########
@@ -154,13 +154,13 @@ class BrokerRegistrationRequestTest {
// No features (MV) sent with registration, controller can't verify
Review Comment:
Previously, passing `Some(1)` as the `zkEpoch` parameter caused `zkEpoch.isDefined` to always be `true`, which led to a [BrokerIdNotRegisteredException](https://github.com/apache/kafka/blob/634e99e9ab16db61990f4922f0a30b6a93f27785/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java#L374C8-L376C10) being thrown.
As a result, the intended test scenario was never actually tested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org