KAFKA-18275: Restarting broker in testing should use the same port #18381

peterxcli · 2025-01-03T08:45:42Z

Enable brokers to restart with the same port in ClusterTest.

Tests

In the newly added test testBrokerRestart, if we don't have:

for (TestKitNode node : nodes.brokerNodes().values()) {
    socketFactoryManager.getOrCreatePortForListener(node.id(), brokerListenerName);
}

to preallocate the ports for brokers, it would failed with following logs and error:

...

[2025-01-03 16:38:19,741] INFO [Producer clientId=producer-1] Node -1 disconnected. (org.apache.kafka.clients.NetworkClient:1073)
[2025-01-03 16:38:19,742] WARN [Producer clientId=producer-1] Connection to node -1 (localhost/127.0.0.1:55465) could not be established. Node may not be available. (org.apache.kafka.clients.NetworkClient:899)
[2025-01-03 16:38:19,742] WARN [Producer clientId=producer-1] Bootstrap broker localhost:55465 (id: -1 rack: null isFenced: false) disconnected (org.apache.kafka.clients.NetworkClient:1256)
[2025-01-03 16:38:20,745] INFO [Producer clientId=producer-1] Node -1 disconnected. (org.apache.kafka.clients.NetworkClient:1073)
[2025-01-03 16:38:20,745] WARN [Producer clientId=producer-1] Connection to node -1 (localhost/127.0.0.1:55465) could not be established. Node may not be available. (org.apache.kafka.clients.NetworkClient:899)
[2025-01-03 16:38:20,745] WARN [Producer clientId=producer-1] Bootstrap broker localhost:55465 (id: -1 rack: null isFenced: false) disconnected (org.apache.kafka.clients.NetworkClient:1256)
[2025-01-03 16:38:21,667] INFO [Producer clientId=producer-1] Node -1 disconnected. (org.apache.kafka.clients.NetworkClient:1073)
[2025-01-03 16:38:21,667] WARN [Producer clientId=producer-1] Connection to node -1 (localhost/127.0.0.1:55465) could not be established. Node may not be available. (org.apache.kafka.clients.NetworkClient:899)
[2025-01-03 16:38:21,667] WARN [Producer clientId=producer-1] Bootstrap broker localhost:55465 (id: -1 rack: null isFenced: false) disconnected (org.apache.kafka.clients.NetworkClient:1256)
[2025-01-03 16:38:21,667] INFO [Producer clientId=producer-1] Rebootstrapping with [localhost/127.0.0.1:55465] (org.apache.kafka.clients.Metadata:314)

...


testBrokerRestart(org.apache.kafka.common.test.api.ClusterInstance) timed out after 60 seconds

peterxcli · 2025-01-03T08:46:23Z

test-common/src/main/java/org/apache/kafka/common/test/PreboundSocketFactoryManager.java

+
            if (socketChannel != null) {
+                if (socketChannel.isOpen()) {
+                    return socketChannel;
+                }
+                // bind the server socket with same port
+                socketAddress = new InetSocketAddress(socketAddress.getHostString(), socketChannel.socket().getLocalPort());
+                socketChannel = ServerSocketFactory.INSTANCE.openServerSocket(
+                        listenerName,
+                        socketAddress,
+                        listenBacklogSize,
+                        recvBufferSize);


This PR depends #18337

peterxcli · 2025-01-04T06:50:13Z

...runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java

-    private void useFixedBrokerPort() throws IOException {
-        // Find a free port and use it in the Kafka broker's listeners config. We can't use port 0 in the listeners
-        // config to get a random free port because in this test we want to stop the Kafka broker and then bring it
-        // back up and listening on the same port in order to verify that the Connect cluster can re-connect to Kafka
-        // and continue functioning normally. If we were to use port 0 here, the Kafka broker would most likely listen
-        // on a different random free port the second time it is started. Note that we can only use the static port
-        // because we have a single broker setup in this test.
-        int listenerPort;
-        try (ServerSocket s = new ServerSocket(0)) {
-            listenerPort = s.getLocalPort();
-        }
-        brokerProps.put(SocketServerConfigs.LISTENERS_CONFIG, String.format("EXTERNAL://localhost:%d,CONTROLLER://localhost:0", listenerPort));
-        connectBuilder
-                .numBrokers(1)
-                .brokerProps(brokerProps);
-    }
-


Remove this to pass the following two tests.

Module Test Message Time

ConnectWorkerIntegrationTest testBrokerCoordinator() org.opentest4j.AssertionFailedError: The Kafka cluster used in this test was not able to start successfully in time. If no recent changes have altered the behavior of Kafka brokers or clients, and this error is not occurring frequently, it is probably the result of the testing machine being temporarily overloaded and can be safely ignored. 62.84s

ConnectWorkerIntegrationTest testRequestTimeouts() org.opentest4j.AssertionFailedError: The Kafka cluster used in this test was not able to start successfully in time. If no recent changes have altered the behavior of Kafka brokers or clients, and this error is not occurring frequently, it is probably the result of the testing machine being temporarily overloaded and can be safely ignored. 60.51s

https://github.com/apache/kafka/actions/runs/12595111592?pr=18381

But I don't really know why, investigating...

oh, EmbeddedConnect uses EmbeddedKafkaCluster, and EmbeddedKafkaCluster uses KafkaClusterTestKit to build cluster, and since the brokers has assigned fixed ports in KafkaClusterTestKit, there is no need to modify the LISTENERS_CONFIG for them

peterxcli · 2025-01-04T06:50:52Z

...runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java

-        useFixedBrokerPort();
-


peterxcli · 2025-01-04T06:50:59Z

...runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java

-        useFixedBrokerPort();
-


…imeouts

peterxcli · 2025-01-08T18:59:10Z

Failed tests are tracked by:

…rt-after-broker-restart

peterxcli · 2025-01-17T13:44:27Z

test-common/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java

-            Map<Integer, ControllerServer> controllers = new HashMap<>();
-            Map<Integer, BrokerServer> brokers = new HashMap<>();
-            Map<Integer, SharedServer> jointServers = new HashMap<>();
-            File baseDirectory = null;
-            File jaasFile = null;
-
+        private Optional<File> maybeSetupJaasFile() throws Exception {
            if (brokerSecurityProtocol.equals(SecurityProtocol.SASL_PLAINTEXT.name)) {
-                jaasFile = JaasUtils.writeJaasContextsToFile(Set.of(
+                File file = JaasUtils.writeJaasContextsToFile(Set.of(
                    new JaasUtils.JaasSection(JaasUtils.KAFKA_SERVER_CONTEXT_NAME,
                        List.of(
                            JaasModule.plainLoginModule(
-                                JaasUtils.KAFKA_PLAIN_ADMIN, JaasUtils.KAFKA_PLAIN_ADMIN_PASSWORD,
+                                JaasUtils.KAFKA_PLAIN_ADMIN, 
+                                JaasUtils.KAFKA_PLAIN_ADMIN_PASSWORD,
                                true,
                                Map.of(
                                    JaasUtils.KAFKA_PLAIN_USER1, JaasUtils.KAFKA_PLAIN_USER1_PASSWORD,
                                    JaasUtils.KAFKA_PLAIN_ADMIN, JaasUtils.KAFKA_PLAIN_ADMIN_PASSWORD)
-                                )
+                            )
                        )
                    )
                ));
-                JaasUtils.refreshJavaLoginConfigParam(jaasFile);
+                JaasUtils.refreshJavaLoginConfigParam(file);
+                return Optional.of(file);
            }
+            return Optional.empty();
+        }

+        public KafkaClusterTestKit build() throws Exception {
+            Map<Integer, ControllerServer> controllers = new HashMap<>();
+            Map<Integer, BrokerServer> brokers = new HashMap<>();
+            Map<Integer, SharedServer> jointServers = new HashMap<>();
+            File baseDirectory = null;
+            Optional<File> jaasFile = maybeSetupJaasFile();


This is for resolving the NComplexity error.

chia7712

@peterxcli thanks for this patch!

chia7712 · 2025-01-18T03:07:00Z

...est-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterTestExtensionsTest.java

        }
    }

+    @ClusterTest(types = {Type.KRAFT}, brokers = 1)


Should we test CO_KRAFT as well?

chia7712 · 2025-01-18T03:07:26Z

...est-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterTestExtensionsTest.java

+        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+
+        try (Admin admin = cluster.admin();
+             Producer<String, String> producer = cluster.producer(Utils.propsToMap(producerProps))) {


Could you please use Map.of instead?

chia7712 · 2025-02-03T08:02:41Z

@peterxcli could you please fix the conflicts?

…rt-after-broker-restart

peterxcli · 2025-02-03T11:10:18Z

@chia7712 Resolved. PTAL, Thanks!

…ache#18381) Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

github-actions bot added triage PRs from the community tests Test fixes (including flaky tests) small Small PRs labels Jan 3, 2025

peterxcli commented Jan 3, 2025

View reviewed changes

FrankYang0529 added the ci-approved label Jan 3, 2025

github-actions bot added connect and removed triage PRs from the community labels Jan 4, 2025

peterxcli commented Jan 4, 2025

View reviewed changes

peterxcli added 6 commits January 8, 2025 13:05

restart controller with same port in ClusterTest

3bce960

combine the null check for socketChannel in PreboundSocketFactoryManager

f7bb644

rename testKRaftIsolatedControllerRestart to testControllerRestart

687af35

restart controller with same port in ClusterTest

6849805

Allow broker restart on same port in CLusterTest

3f9ad3d

fix ConnectWorkerIntegrationTest.testBrokerCoordinator & testRequestT…

df01bb6

…imeouts

peterxcli force-pushed the k18275-use-same-port-after-broker-restart branch from 84e8137 to df01bb6 Compare January 8, 2025 13:28

peterxcli added 2 commits January 8, 2025 13:30

remove duplicated testKRaftIsolatedControllerRestart

5fe5225

Resolve Ncomplexity checkstyle issue

a7c3b70

github-actions bot removed the small Small PRs label Jan 8, 2025

peterxcli mentioned this pull request Jan 8, 2025

KAFKA-18274: Failed to restart controller in testing due to closed socket channel [2/2] #18337

Merged

Merge remote-tracking branch 'upstream/trunk' into k18275-use-same-po…

53ddbc8

…rt-after-broker-restart

github-actions bot added the small Small PRs label Jan 17, 2025

peterxcli commented Jan 17, 2025

View reviewed changes

chia7712 reviewed Jan 18, 2025

View reviewed changes

Address chia's comment

75157b9

peterxcli requested a review from chia7712 January 18, 2025 09:56

Merge remote-tracking branch 'upstream/trunk' into k18275-use-same-po…

e407722

…rt-after-broker-restart

chia7712 approved these changes Feb 6, 2025

View reviewed changes

chia7712 merged commit 0621c0b into apache:trunk Feb 6, 2025
8 of 9 checks passed

pdruley pushed a commit to pdruley/kafka that referenced this pull request Feb 12, 2025

KAFKA-18275 Restarting broker in testing should use the same port (ap…

a2aff7c

…ache#18381) Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

manoj-mathivanan pushed a commit to manoj-mathivanan/kafka that referenced this pull request Feb 19, 2025

KAFKA-18275 Restarting broker in testing should use the same port (ap…

53da843

…ache#18381) Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

Module	Test	Message	Time
ConnectWorkerIntegrationTest	testBrokerCoordinator()	org.opentest4j.AssertionFailedError: The Kafka cluster used in this test was not able to start successfully in time. If no recent changes have altered the behavior of Kafka brokers or clients, and this error is not occurring frequently, it is probably the result of the testing machine being temporarily overloaded and can be safely ignored.	62.84s
ConnectWorkerIntegrationTest	testRequestTimeouts()	org.opentest4j.AssertionFailedError: The Kafka cluster used in this test was not able to start successfully in time. If no recent changes have altered the behavior of Kafka brokers or clients, and this error is not occurring frequently, it is probably the result of the testing machine being temporarily overloaded and can be safely ignored.	60.51s

KAFKA-18275: Restarting broker in testing should use the same port #18381

KAFKA-18275: Restarting broker in testing should use the same port #18381

Uh oh!

Conversation

peterxcli commented Jan 3, 2025

Tests

Uh oh!

peterxcli Jan 3, 2025

Choose a reason for hiding this comment

Uh oh!

peterxcli Jan 4, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

peterxcli Jan 4, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

peterxcli Jan 4, 2025

Choose a reason for hiding this comment

Uh oh!

peterxcli Jan 4, 2025

Choose a reason for hiding this comment

Uh oh!

peterxcli commented Jan 8, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

peterxcli Jan 17, 2025

Choose a reason for hiding this comment

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

chia7712 Jan 18, 2025

Choose a reason for hiding this comment

Uh oh!

chia7712 Jan 18, 2025

Choose a reason for hiding this comment

Uh oh!

chia7712 commented Feb 3, 2025

Uh oh!

peterxcli commented Feb 3, 2025

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

peterxcli Jan 4, 2025 •

edited

Loading

peterxcli Jan 4, 2025 •

edited

Loading

peterxcli commented Jan 8, 2025 •

edited

Loading