j1wonpark opened a new pull request, #58205: URL: https://github.com/apache/spark/pull/58205
### What changes were proposed in this pull request? Run the request loop of `TransformWithStateInPySparkStateServer.run()` inside `Utils.tryWithResource` so the socket returned by `stateServerSocket.accept()` is always closed. ### Why are the changes needed? `run()` accepts a connection from the Python worker but never closes it. The task completion listener closes only the listening `ServerSocketChannel`, so the accepted connection's file descriptor is held until the executor exits. `initStateServer()` binds an ephemeral port per task, so a long-running streaming query leaks one socket per task. Once the ephemeral port range is exhausted, every new connection on that executor fails with `java.net.BindException: Cannot assign requested address`, which also takes down unrelated connections such as the Kafka source. The loop has several early returns, so the close has to be tied to the scope. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Two tests added to `TransformWithStateInPySparkStateServerSuite`, covering the path where the request loop is never entered and the EOF early-return path. Both fail without the fix. The suite previously had no test calling `run()`. ### Was this patch authored or co-authored using generative AI tooling? Yes. Generated-by: Claude Opus 5 -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
