Skip to content

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

Closed
j1wonpark wants to merge 2 commits into
apache:masterfrom
j1wonpark:tws-state-server-socket-leak
Closed

[SPARK-58932][SS] Close the accepted socket in TransformWithStateInPySparkStateServer#58205
j1wonpark wants to merge 2 commits into
apache:masterfrom
j1wonpark:tws-state-server-socket-leak

Conversation

@j1wonpark

@j1wonpark j1wonpark commented Aug 21, 2026

Copy link
Copy Markdown
Contributor

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

@j1wonpark
[SPARK-58932][SS] Close the accepted socket in TransformWithStateInPy…
9f7e7ba 
…SparkStateServer

### 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

Signed-off-by: Jiwon Park <jpark92@outlook.kr>
@j1wonpark
j1wonpark force-pushed the tws-state-server-socket-leak branch from 2334021 to 9f7e7ba Compare August 21, 2026 14:05
@uros-b

uros-b commented Aug 21, 2026

Copy link
Copy Markdown
Member

Thank you @j1wonpark!

@HeartSaVioR

HeartSaVioR commented Aug 26, 2026

Copy link
Copy Markdown
Contributor

@j1wonpark
Thanks for the contribution!
I see we missed to close the socket, but we still need to take care of some exception cases, hence the merge conflict. Could you please rebase and retain the exception handling while ensuring the closure of the socket?
Thanks in advance!

@j1wonpark
Merge remote-tracking branch 'upstream/master' into tws-state-server-…
119cf85 
…socket-leak

Signed-off-by: Jiwon Park <jpark92@outlook.kr>

# Conflicts:
#	sql/core/src/main/scala/org/apache/spark/sql/execution/python/streaming/TransformWithStateInPySparkStateServer.scala
#	sql/core/src/test/scala/org/apache/spark/sql/execution/python/streaming/TransformWithStateInPySparkStateServerSuite.scala
@j1wonpark
j1wonpark force-pushed the tws-state-server-socket-leak branch from 9e01469 to 119cf85 Compare August 26, 2026 04:51
@j1wonpark

j1wonpark commented Aug 26, 2026

Copy link
Copy Markdown
Contributor Author

@HeartSaVioR Thanks! Merged the latest master, keeping the exception handling from SPARK-58977 while closing the accepted socket via Utils.tryWithResource. All tests in the suite pass locally.

HeartSaVioR
HeartSaVioR approved these changes Aug 26, 2026
View reviewed changes

@HeartSaVioR HeartSaVioR left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1

Comment thread ...org/apache/spark/sql/execution/python/streaming/TransformWithStateInPySparkStateServer.scala
Utils.tryWithResource(listeningSocket)(serveRequests)
}

private def serveRequests(listeningSocket: SocketChannel): Unit = {

@HeartSaVioR HeartSaVioR Aug 26, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Neat approach to avoid indentation :) Thanks!

HeartSaVioR pushed a commit that referenced this pull request Aug 26, 2026
@j1wonpark @HeartSaVioR
[SPARK-58932][SS] Close the accepted socket in TransformWithStateInPy…
516942f 
…SparkStateServer

### 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

Closes #58205 from j1wonpark/tws-state-server-socket-leak.

Authored-by: Jiwon Park <jpark92@outlook.kr>
Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>
(cherry picked from commit c5fbdf7)
Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>
HeartSaVioR pushed a commit that referenced this pull request Aug 26, 2026
@j1wonpark @HeartSaVioR
[SPARK-58932][SS] Close the accepted socket in TransformWithStateInPy…
818c2c8 
…SparkStateServer

### 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

Closes #58205 from j1wonpark/tws-state-server-socket-leak.

Authored-by: Jiwon Park <jpark92@outlook.kr>
Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>
(cherry picked from commit c5fbdf7)
Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>
@HeartSaVioR

HeartSaVioR commented Aug 26, 2026

Copy link
Copy Markdown
Contributor

Merge Summary:

  • merged into master c5fbdf7
  • merged into branch-4.x 516942f
  • merged into branch-4.3 818c2c8
    Posted by merge_spark_pr.py

@HeartSaVioR

HeartSaVioR commented Aug 26, 2026

Copy link
Copy Markdown
Contributor

@j1wonpark
I've merged the fix till 4.3 which we hope to release in a month. Please submit a PR for older version branch if you were aiming to have the fix for older version lines. Thanks!

This was referenced Aug 26, 2026
Closed
Closed
@j1wonpark

j1wonpark commented Aug 26, 2026

Copy link
Copy Markdown
Contributor Author

@HeartSaVioR Opened backport PRs for the older version lines: #58308 (branch-4.2) and #58309 (branch-4.1). Both branches lack SPARK-58977, so the cherry-pick keeps accept() unchanged and only wraps the request loop. Thanks!

HeartSaVioR pushed a commit that referenced this pull request Aug 27, 2026
@j1wonpark @HeartSaVioR
[SPARK-58932][4.2][SS] Close the accepted socket in TransformWithStat…
6cb5216 
…eInPySparkStateServer

### 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>
HeartSaVioR pushed a commit that referenced this pull request Aug 27, 2026
@j1wonpark @HeartSaVioR
[SPARK-58932][4.2][SS] Close the accepted socket in TransformWithStat…
993357d 
…eInPySparkStateServer

### 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>
(cherry picked from commit 6cb5216)
Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@HeartSaVioR HeartSaVioR HeartSaVioR approved these changes
@uros-b uros-b uros-b approved these changes

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

3 participants

@j1wonpark @uros-b @HeartSaVioR