diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala index cdc8ece103c03..881f7d3f9ec1d 100644 --- a/core/src/main/scala/kafka/network/SocketServer.scala +++ b/core/src/main/scala/kafka/network/SocketServer.scala @@ -723,6 +723,10 @@ 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..3f804b343ac4e 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,42 @@ class SocketServerTest { }, false) } + @Test + def testDataPlaneAcceptingSocketUsesReuseAddress(): Unit = { + val acceptor = server.dataPlaneAcceptor(listener) + val channel = acceptor.get.serverChannel + verifySocketUsesReuseAddress(channel) + } + + @Test + 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) + 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 = { + 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.