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-18274: Failed to restart controller in testing due to closed socket channel [2/2] #18337

Open
wants to merge 4 commits into
base: trunk
Choose a base branch
from
+31 −0
Open

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

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
3bce960
restart controller with same port in ClusterTest
peterxcli Dec 28, 2024
f7bb644
combine the null check for socketChannel in PreboundSocketFactoryManager
peterxcli Jan 2, 2025
687af35
rename testKRaftIsolatedControllerRestart to testControllerRestart
peterxcli Jan 8, 2025
ab0c0ad
Add comment to explain the scenrio of same port bind after components…
peterxcli Jan 9, 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
11 changes: 11 additions & 0 deletions test-common/src/main/java/org/apache/kafka/common/test/PreboundSocketFactoryManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,18 @@ public ServerSocketChannel openServerSocket(
ServerSocketChannel socketChannel = getSocketForListenerAndMarkAsUsed(
nodeId,
listenerName);

if (socketChannel != null) {
if (socketChannel.isOpen()) {
return socketChannel;
}
// bind the server socket with same port
Copy link
Member

Choose a reason for hiding this comment

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

Could you please add comments to explain the use cases of binding the same port?

Copy link
Member

Choose a reason for hiding this comment

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

any updates?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Hi, sorry for the late response, added some comment on it, PTAL. Thanks!

socketAddress = new InetSocketAddress(socketAddress.getHostString(), socketChannel.socket().getLocalPort());
socketChannel = ServerSocketFactory.INSTANCE.openServerSocket(
listenerName,
socketAddress,
listenBacklogSize,
recvBufferSize);
return socketChannel;
}
return ServerSocketFactory.INSTANCE.openServerSocket(
Expand Down
16 changes: 16 additions & 0 deletions ...-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterTestExtensionsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@

package org.apache.kafka.common.test.api;

import kafka.server.ControllerServer;

import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.Config;
Expand Down Expand Up @@ -347,6 +349,20 @@ public void testControllerListenerName(ClusterInstance cluster) throws Execution
}
}

@ClusterTest(types = {Type.KRAFT})
public void testControllerRestart(ClusterInstance cluster) throws ExecutionException, InterruptedException {
try (Admin admin = cluster.admin()) {

ControllerServer controller = cluster.controllers().values().iterator().next();
controller.shutdown();
controller.awaitShutdown();
Copy link
Member

Choose a reason for hiding this comment

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

Could you please add test for broker as well? the fix should works for broker, right?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Thanks for asking,
The broker won't be fixed in this change, I have also filed another PR to address it, and this is the core change to fix the broker.


controller.startup();

assertEquals(1, admin.describeMetadataQuorum().quorumInfo().get().nodes().size());
}
}

@ClusterTest(
types = {Type.KRAFT, Type.CO_KRAFT},
brokerSecurityProtocol = SecurityProtocol.SASL_PLAINTEXT,
Expand Down
Loading