KAFKA-14908: Set setReuseAddress on the kafka server socket

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)
 

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.