[GitHub] [kafka] divijvaidya commented on a diff in pull request #13572: KAFKA-14908: Set setReuseAddress on the kafka server socket

["divijvaidya (via GitHub)" <gi...@apache.org>]

divijvaidya commented on code in PR #13572:
URL: https://github.com/apache/kafka/pull/13572#discussion_r1170213078


##########
core/src/test/scala/unit/kafka/network/SocketServerTest.scala:
##########
@@ -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")

Review Comment:
   please add a comment here.
   
   We are using 0 so that OS can choose to associated any available port here.



##########
core/src/test/scala/unit/kafka/network/SocketServerTest.scala:
##########
@@ -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)

Review Comment:
   perhaps do this in try/finally so that the resources are cleaned even if the test fails with an exception.



##########
core/src/test/scala/unit/kafka/network/SocketServerTest.scala:
##########
@@ -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

Review Comment:
   since we are starting up two listeners, could we verify reuse address for both please.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org