[SPARK-58932][4.2][SS] Close the accepted socket in TransformWithStateInPySparkStateServer

### What changes were proposed in this pull request?

Backport of #58205 to branch-4.2: run the request loop of `TransformWithStateInPySparkStateServer.run()` inside `Utils.tryWithResource` so the accepted socket is always closed. This branch does not have SPARK-58977, so unlike master the `accept()` call is unchanged.

### Why are the changes needed?

The accepted socket is never closed, leaking one file descriptor per task until the executor exhausts the ephemeral port range (`java.net.BindException`). See #58205 for details.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

The two tests added in #58205 are included; `TransformWithStateInPySparkStateServerSuite` passes locally on this branch.

### Was this patch authored or co-authored using generative AI tooling?

Yes. Generated-by: Claude Opus 5

Closes #58308 from j1wonpark/SPARK-58932-4.2.

Authored-by: Jiwon Park <jpark92@outlook.kr>
Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

Author: j1wonpark

sql/core/src/main/scala/org/apache/spark/sql/execution/python/streaming/TransformWithStateInPySparkStateServer.scala

@@ -18,7 +18,7 @@
 package org.apache.spark.sql.execution.python.streaming
 
 import java.io.{BufferedInputStream, BufferedOutputStream, DataInputStream, DataOutputStream, EOFException, InterruptedIOException}
-import java.nio.channels.{Channels, ClosedByInterruptException, ServerSocketChannel}
+import java.nio.channels.{Channels, ClosedByInterruptException, ServerSocketChannel, SocketChannel}
 import java.time.Duration
 
 import scala.collection.mutable
@@ -40,6 +40,7 @@ import org.apache.spark.sql.execution.streaming.state.StateMessage.KeyAndValuePa
 import org.apache.spark.sql.execution.streaming.state.StateMessage.StateResponseWithListGet
 import org.apache.spark.sql.streaming.{ListState, MapState, TTLConfig, ValueState}
 import org.apache.spark.sql.types.StructType
+import org.apache.spark.util.Utils
 
 /**
  * This class is used to handle the state requests from the Python side. It runs on a separate
@@ -140,6 +141,13 @@ class TransformWithStateInPySparkStateServer(
   def run(): Unit = {
     val listeningSocket = stateServerSocket.accept()
 
+    // The task completion listener closes only the listening server socket, and the
+    // request loop has several early returns, so the accepted connection is closed
+    // through tryWithResource.
+    Utils.tryWithResource(listeningSocket)(serveRequests)
+  }
+
+  private def serveRequests(listeningSocket: SocketChannel): Unit = {
     // SPARK-51667: We have a pattern of sending messages continuously from one side
     // (Python -> JVM, and vice versa) before getting response from other side. Since most
     // messages we are sending are small, this triggers the bad combination of Nagle's algorithm

sql/core/src/test/scala/org/apache/spark/sql/execution/python/streaming/TransformWithStateInPySparkStateServerSuite.scala

@@ -17,7 +17,9 @@
 package org.apache.spark.sql.execution.python.streaming
 
 import java.io.DataOutputStream
-import java.nio.channels.ServerSocketChannel
+import java.net.Socket
+import java.nio.ByteBuffer
+import java.nio.channels.{ServerSocketChannel, SocketChannel}
 
 import scala.collection.mutable
 
@@ -110,6 +112,34 @@ class TransformWithStateInPySparkStateServerSuite extends SparkFunSuite with Bef
       .thenReturn(Seq(getIntegerRow(1)))
   }
 
+  test("run closes the accepted socket once the request loop ends") {
+    val acceptedSocket = mock(classOf[SocketChannel])
+    when(serverSocket.accept()).thenReturn(acceptedSocket)
+    when(acceptedSocket.socket()).thenReturn(mock(classOf[Socket]))
+    // Ends the request loop right away: this test is about the socket, not the requests.
+    when(acceptedSocket.isConnected).thenReturn(false)
+
+    stateServer.run()
+
+    verify(acceptedSocket).close()
+  }
+
+  test("run closes the accepted socket when the client disconnects") {
+    val acceptedSocket = mock(classOf[SocketChannel])
+    when(serverSocket.accept()).thenReturn(acceptedSocket)
+    when(acceptedSocket.socket()).thenReturn(mock(classOf[Socket]))
+    when(acceptedSocket.isConnected).thenReturn(true)
+    // Channels.newInputStream synchronizes on this before reading.
+    when(acceptedSocket.blockingLock()).thenReturn(new Object)
+    when(acceptedSocket.isBlocking).thenReturn(true)
+    // No bytes ever arrive, so the read hits EOF and the loop returns early.
+    when(acceptedSocket.read(any(classOf[ByteBuffer]))).thenReturn(-1)
+
+    stateServer.run()
+
+    verify(acceptedSocket).close()
+  }
+
   test("set handle state") {
     val message = StatefulProcessorCall.newBuilder().setSetHandleState(
       SetHandleState.newBuilder().setState(HandleState.CREATED).build()).build()