KAFKA-16765: Close leaked accepted sockets in EchoServer, NioEchoServ…

…er, ServerShutdownTest (apache#16576) Reviewers: Greg Harris <[email protected]>
smileby · Jul 24, 2024 · a38829e · a38829e
1 parent ee68f32
commit a38829e
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 1 deletion.
diff --git a/clients/src/test/java/org/apache/kafka/common/network/EchoServer.java b/clients/src/test/java/org/apache/kafka/common/network/EchoServer.java
@@ -89,6 +89,7 @@ public void run() {
                 }
                 synchronized (sockets) {
                     if (closing) {
+                        socket.close();
                         break;
                     }
                     sockets.add(socket);

diff --git a/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java b/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java
@@ -362,12 +362,20 @@ public void closeSocketChannels() throws IOException {
         socketChannels.clear();
     }
 
+    public void closeNewChannels() throws IOException {
+        for (SocketChannel channel : newChannels) {
+            channel.close();
+        }
+        newChannels.clear();
+    }
+
     public void close() throws IOException, InterruptedException {
         this.serverSocketChannel.close();
         closeSocketChannels();
         Utils.closeQuietly(selector, "selector");
         acceptorThread.interrupt();
         acceptorThread.join();
+        closeNewChannels();
         interrupt();
         join();
     }

diff --git a/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala b/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala
@@ -272,7 +272,9 @@ class ServerShutdownTest extends KafkaServerTestHarness {
       val receiveFuture = executor.submit(new Runnable {
         override def run(): Unit = {
           val socket = serverSocket.accept()
-          new DataInputStream(socket.getInputStream).readByte()
+          val inputStream = new DataInputStream(socket.getInputStream)
+          inputStream.readByte()
+          inputStream.close()
         }
       })