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 all 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
15 changes: 15 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,22 @@ public ServerSocketChannel openServerSocket(
ServerSocketChannel socketChannel = getSocketForListenerAndMarkAsUsed(
nodeId,
listenerName);

if (socketChannel != null) {
if (socketChannel.isOpen()) {
return socketChannel;
}
// When restarting components(e.g. controllers, brokers) in tests, we want to reuse the same
// port that was previously allocated to maintain consistent addressing
// so the client can reconnect to the same port.
// Since those components would close the socket when they are restarted,
// we need to rebind the socket to the same port.
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