From 01bb47a1cac3449bafb39dfa22158f7a4efacc83 Mon Sep 17 00:00:00 2001 From: kwall Date: Fri, 14 Apr 2023 12:58:24 +0100 Subject: [PATCH 1/3] KAFKA-14908: Set setReuseAddress on the kafka server socket Aids use-cases where kafka is started/stopped on the same port, such as embedded testing. Signed-off-by: kwall --- core/src/main/scala/kafka/network/SocketServer.scala | 1 + 1 file changed, 1 insertion(+) diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala index cdc8ece103c03..4ae77f4915a6d 100644 --- a/core/src/main/scala/kafka/network/SocketServer.scala +++ b/core/src/main/scala/kafka/network/SocketServer.scala @@ -723,6 +723,7 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer, new InetSocketAddress(host, port) val serverChannel = ServerSocketChannel.open() serverChannel.configureBlocking(false) + serverChannel.socket().setReuseAddress(true); if (recvBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE) serverChannel.socket().setReceiveBufferSize(recvBufferSize) From 44c5bd5ced08ef970d29bcb899d5c9a3f95a0bbe Mon Sep 17 00:00:00 2001 From: kwall Date: Tue, 18 Apr 2023 16:13:56 +0100 Subject: [PATCH 2/3] add unit tests --- .../scala/kafka/network/SocketServer.scala | 3 ++ .../unit/kafka/network/SocketServerTest.scala | 29 ++++++++++++++++++- 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala index 4ae77f4915a6d..881f7d3f9ec1d 100644 --- a/core/src/main/scala/kafka/network/SocketServer.scala +++ b/core/src/main/scala/kafka/network/SocketServer.scala @@ -723,6 +723,9 @@ private[kafka] abstract class Acceptor(val socketServer: SocketServer, new InetSocketAddress(host, port) val serverChannel = ServerSocketChannel.open() serverChannel.configureBlocking(false) + // Configure the socket with setReuseAddress(true). This is done to aid use-cases where the kafka + // server is rapidly shut down and started up on the same port (e.g. application integration test suites + // that embed a kafka cluster). serverChannel.socket().setReuseAddress(true); if (recvBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE) serverChannel.socket().setReceiveBufferSize(recvBufferSize) diff --git a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala index 28b797f7c1679..69b5632d8b19b 100644 --- a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala +++ b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala @@ -20,7 +20,7 @@ package kafka.network import java.io._ import java.net._ import java.nio.ByteBuffer -import java.nio.channels.{SelectionKey, SocketChannel} +import java.nio.channels.{SelectionKey, ServerSocketChannel, SocketChannel} import java.nio.charset.StandardCharsets import java.util import java.util.concurrent.{CompletableFuture, ConcurrentLinkedQueue, ExecutionException, Executors, TimeUnit} @@ -1893,6 +1893,33 @@ class SocketServerTest { }, false) } + @Test + def testDataPlaneAcceptingSocketUsesReuseAddress(): Unit = { + val acceptor = server.dataPlaneAcceptor(listener) + val channel = acceptor.get.serverChannel + verifySocketUsesReuseAddress(channel) + } + + @Test + def testControlPlaneAcceptingSocketUsesReuseAddress(): Unit = { + shutdownServerAndMetrics(server) + val testProps = new Properties + testProps ++= props + testProps.put("listeners", "PLAINTEXT://localhost:0,CONTROL_PLANE://localhost:0") + testProps.put("listener.security.protocol.map", "PLAINTEXT:PLAINTEXT,CONTROL_PLANE:PLAINTEXT") + testProps.put("control.plane.listener.name", "CONTROL_PLANE") + val config = KafkaConfig.fromProps(testProps) + val testServer = new SocketServer(config, metrics, Time.SYSTEM, credentialProvider, apiVersionManager) + val channel = testServer.controlPlaneAcceptorOpt.get.serverChannel + verifySocketUsesReuseAddress(channel) + shutdownServerAndMetrics(testServer) + } + + private def verifySocketUsesReuseAddress(channel: ServerSocketChannel): Unit = { + assertTrue(channel.socket().isBound, "Listening channel not bound") + assertTrue(channel.socket().getReuseAddress, "Listening socket reuseAddress in unexpected state") + } + /** * Test to ensure "Selector.poll()" does not block at "select(timeout)" when there is no data in the socket but there * is data in the buffer. This only happens when SSL protocol is used. From 2a0a40b61265572f679c5ccd85febbee88e09e8b Mon Sep 17 00:00:00 2001 From: kwall Date: Tue, 18 Apr 2023 16:54:05 +0100 Subject: [PATCH 3/3] use try/finally in the new tests, assess re-use state of both acceptors --- .../unit/kafka/network/SocketServerTest.scala | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala index 69b5632d8b19b..3f804b343ac4e 100644 --- a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala +++ b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala @@ -1901,18 +1901,27 @@ class SocketServerTest { } @Test - def testControlPlaneAcceptingSocketUsesReuseAddress(): Unit = { + def testControlAndDataPlaneAcceptingSocketsUseReuseAddress(): Unit = { shutdownServerAndMetrics(server) val testProps = new Properties testProps ++= props + // Use port 0 so that the system will assign an available port from the ephemeral range testProps.put("listeners", "PLAINTEXT://localhost:0,CONTROL_PLANE://localhost:0") testProps.put("listener.security.protocol.map", "PLAINTEXT:PLAINTEXT,CONTROL_PLANE:PLAINTEXT") testProps.put("control.plane.listener.name", "CONTROL_PLANE") val config = KafkaConfig.fromProps(testProps) val testServer = new SocketServer(config, metrics, Time.SYSTEM, credentialProvider, apiVersionManager) - val channel = testServer.controlPlaneAcceptorOpt.get.serverChannel - verifySocketUsesReuseAddress(channel) - shutdownServerAndMetrics(testServer) + try { + val dataPlaneAcceptor = testServer.dataPlaneAcceptor(listener) + val dataPlaneChannel = dataPlaneAcceptor.get.serverChannel + verifySocketUsesReuseAddress(dataPlaneChannel) + + val controlPlaneAcceptor = testServer.controlPlaneAcceptorOpt.get + val acceptingChannel = controlPlaneAcceptor.serverChannel + verifySocketUsesReuseAddress(acceptingChannel) + } finally { + shutdownServerAndMetrics(testServer) + } } private def verifySocketUsesReuseAddress(channel: ServerSocketChannel): Unit = {