[SPARK-47146][CORE] Possible thread leak when doing sort merge join #45327

JacobZheng0927 · 2024-02-29T02:31:06Z

What changes were proposed in this pull request?

Add TaskCompletionListener to close inputStream to avoid thread leakage caused by unclosed ReadAheadInputStream.

Why are the changes needed?

SPARK-40849 modified the implementation of newDaemonSingleThreadExecutor to use newFixedThreadPool instead of newSingleThreadExecutor .The difference is that newSingleThreadExecutor uses the FinalizableDelegatedExecutorService, which provides a finalize method that automatically closes the thread pool. In some cases, sort merge join execution uses ReadAheadSteam and does not close it, so this change caused a thread leak. Since Finalization is deprecated and subject to removal in a future release, we should close the associated streams instead of relying on the finalize method.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Unit test

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

No

zml1206 · 2024-02-29T02:38:41Z

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

 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.util.ThreadUtils

+import java.util.concurrent.{Executors, ThreadPoolExecutor}


Scala style error.

zml1206 · 2024-02-29T02:54:58Z

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

-    = ThreadUtils.newDaemonSingleThreadExecutor("async-log-purge")
+  private val asyncPurgeExecutorService: ThreadPoolExecutor = {
+    val threadFactory =
+      new ThreadFactoryBuilder().setDaemon(true).setNameFormat("async-log-purge").build()


How about adding an additional function to ThreadUtils?

mridulm · 2024-02-29T05:11:09Z

Thanks for surfacing this issue !

To fix this, I would suggest to change ReadAheadInputStream to take TaskContext as a parameter - and add a task completion listener - which calls close.

That would fix the issue, more proactively release resources (instead of waiting for finalization to kick in) and also future proof this code (when support for finalize is removed from JDK)

mridulm · 2024-02-29T05:12:29Z

Also, +CC @HeartSaVioR who reviewed the previous PR

JacobZheng0927 · 2024-02-29T05:28:18Z

Thanks for surfacing this issue !

To fix this, I would suggest to change ReadAheadInputStream to take TaskContext as a parameter - and add a task completion listener - which calls close.

That would fix the issue, more proactively release resources (instead of waiting for finalization to kick in) and also future proof this code (when support for finalize is removed from JDK)

To fix this, I would suggest to change ReadAheadInputStream to take TaskContext as a parameter - and add a task completion listener - which calls close.

Thanks, that's a very good point. I will try this modification.

…ge caused by unclosed ReadAheadInputStream.

mridulm · 2024-02-29T07:22:54Z

core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeSorterSpillReader.java

+        close();
+      } catch (IOException e) {
+        logger.error("error while closing UnsafeSorterSpillReader", e);
+        throw new RuntimeException(e);


We can log in info and drop the exception - for 'normal' flow, that is what ends up happening when close throws exception.

sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala

mridulm · 2024-02-29T07:25:53Z

sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala

+        .find {
+          _.getName.startsWith("read-ahead")
+        }
+      assert(readAheadThread.isEmpty)


I am not sure if this test can end up becoming flakey - in case some other tests result in read-ahead thread hanging around (for example, if some previous tests task is still running).

+CC @dongjoon-hyun for thoughts on how to test this more robustly (I am wondering if this will end up in a similar flakeyness with the other memory manager PR recently, where tests were interacting).

I'm not aware of any flakiness about JoinSuite in these days, @mridulm .

I was referring to this PR @dongjoon-hyun - the RC for that was some other previous test task interfering the the current (due to delays in killing tasks).

@mridulm I guess this case should be fine. In the memory manager PR, the JobCancellationSuite specifically set some tasks to sleep for certain amount of time which causes them to not finish even if the test itself is finished. I don't see similar pattern in the JoinSuite.

Thanks for confirming, that was my assessment as well - but wanted to double check given we faced it very recently :-)

…nListener.

dongjoon-hyun · 2024-02-29T16:49:12Z

Hi, @viirya . Did you see any thread leaks while you are working on Join (or shuffle)?

dongjoon-hyun · 2024-02-29T16:51:07Z

core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeSorterSpillReader.java

+    taskContext.addTaskCompletionListener(context -> {
+      try {
+        close();
+      } catch (IOException e) {


For my understanding, what happens if we fail to close here? Thread Leaks still happen?

I don't think there should be a leak, executionService.shutdownNow() will always be called.

viirya · 2024-02-29T17:16:06Z

Hi, @viirya . Did you see any thread leaks while you are working on Join (or shuffle)?

Hi, @dongjoon-hyun. No, I've not seen it. Maybe it is because our tests are not long running ones that have no chance to show any symptom of noticeable leak.

sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala

mridulm

Looks good to me.
Any thoughts on this @dongjoon-hyun , @sunchao ?

mridulm · 2024-03-05T05:18:12Z

Merged to master.
Thanks for fixing this @JacobZheng0927 !
Thanks for reviewing @zml1206, @dongjoon-hyun :-)

mridulm · 2024-03-05T05:19:34Z

@JacobZheng0927, might be a good idea to backport this to 3.5 as well - will you be able to create a backport PR ?
(I ran into some issue locally when trying to merge to branch-3.5 and bailed)

### What changes were proposed in this pull request? Add TaskCompletionListener to close inputStream to avoid thread leakage caused by unclosed ReadAheadInputStream. ### Why are the changes needed? SPARK-40849 modified the implementation of `newDaemonSingleThreadExecutor` to use `newFixedThreadPool` instead of `newSingleThreadExecutor` .The difference is that `newSingleThreadExecutor` uses the `FinalizableDelegatedExecutorService`, which provides a `finalize` method that automatically closes the thread pool. In some cases, sort merge join execution uses `ReadAheadSteam` and does not close it, so this change caused a thread leak. Since Finalization is deprecated and subject to removal in a future release, we should close the associated streams instead of relying on the finalize method. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Unit test ### Was this patch authored or co-authored using generative AI tooling? No Closes apache#45327 from JacobZheng0927/SPARK-47146. Authored-by: JacobZheng0927 <zsh517559523@163.com> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

JacobZheng0927 · 2024-03-05T12:12:22Z

@JacobZheng0927, might be a good idea to backport this to 3.5 as well - will you be able to create a backport PR ? (I ran into some issue locally when trying to merge to branch-3.5 and bailed)

I created this pr #45390

sunchao · 2024-03-05T15:58:44Z

Any thoughts #45327 (comment) @dongjoon-hyun , @sunchao ?

Sorry for the delay @mridulm . Posted my reply. This PR LGTM too.

This pr backport #45327 to branch-3.5 ### What changes were proposed in this pull request? Add TaskCompletionListener to close inputStream to avoid thread leakage caused by unclosed ReadAheadInputStream. ### Why are the changes needed? To fix the issue SPARK-47146 ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Unit test ### Was this patch authored or co-authored using generative AI tooling? No Closes #45390 from JacobZheng0927/SPARK-47146-3.5. Authored-by: JacobZheng0927 <zsh517559523@163.com> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

This pr backport #45327 to branch-3.5 ### What changes were proposed in this pull request? Add TaskCompletionListener to close inputStream to avoid thread leakage caused by unclosed ReadAheadInputStream. ### Why are the changes needed? To fix the issue SPARK-47146 ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Unit test ### Was this patch authored or co-authored using generative AI tooling? No Closes #45390 from JacobZheng0927/SPARK-47146-3.5. Authored-by: JacobZheng0927 <zsh517559523@163.com> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com> (cherry picked from commit e9f7d36) Signed-off-by: Mridul Muralidharan <mridulatgmail.com>

pan3793 · 2024-03-06T05:37:05Z

core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeSorterSpillReader.java

@@ -36,6 +38,7 @@
 * of the file format).
 */
 public final class UnsafeSorterSpillReader extends UnsafeSorterIterator implements Closeable {
+  private static final Logger logger = LoggerFactory.getLogger(ReadAheadInputStream.class);


is the logger's name correct?

It was careless of me, thank you very much! I'll fix it right away.

### What changes were proposed in this pull request? Add TaskCompletionListener to close inputStream to avoid thread leakage caused by unclosed ReadAheadInputStream. ### Why are the changes needed? SPARK-40849 modified the implementation of `newDaemonSingleThreadExecutor` to use `newFixedThreadPool` instead of `newSingleThreadExecutor` .The difference is that `newSingleThreadExecutor` uses the `FinalizableDelegatedExecutorService`, which provides a `finalize` method that automatically closes the thread pool. In some cases, sort merge join execution uses `ReadAheadSteam` and does not close it, so this change caused a thread leak. Since Finalization is deprecated and subject to removal in a future release, we should close the associated streams instead of relying on the finalize method. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Unit test ### Was this patch authored or co-authored using generative AI tooling? No Closes apache#45327 from JacobZheng0927/SPARK-47146. Authored-by: JacobZheng0927 <zsh517559523@163.com> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

This pr backport apache#45327 to branch-3.5 To fix the issue SPARK-47146 No Unit test No Closes apache#45390 from JacobZheng0927/SPARK-47146-3.5. Authored-by: JacobZheng0927 <zsh517559523@163.com> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com> (cherry picked from commit e9f7d36) Signed-off-by: Mridul Muralidharan <mridulatgmail.com>

github-actions bot added SQL STRUCTURED STREAMING CORE labels Feb 29, 2024

zml1206 reviewed Feb 29, 2024

View reviewed changes

Add TaskCompletionListener to close inputStream to avoid thread leaka…

4af7af8

…ge caused by unclosed ReadAheadInputStream.

JacobZheng0927 force-pushed the SPARK-47146 branch from f1d1be2 to 4af7af8 Compare February 29, 2024 07:05

github-actions bot removed the STRUCTURED STREAMING label Feb 29, 2024

mridulm reviewed Feb 29, 2024

View reviewed changes

Ignore exceptions that close UnsafeSorterSpillReader in TaskCompletio…

39d4158

…nListener.

dongjoon-hyun reviewed Feb 29, 2024

View reviewed changes

mridulm reviewed Feb 29, 2024

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala Show resolved Hide resolved

check taskContext is null before add listener

84022f1

mridulm approved these changes Mar 3, 2024

View reviewed changes

mridulm closed this in dfb35be Mar 5, 2024

JacobZheng0927 mentioned this pull request Mar 5, 2024

[SPARK-47146][CORE][3.5] Possible thread leak when doing sort merge join #45390

Closed

pan3793 reviewed Mar 6, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-47146][CORE] Possible thread leak when doing sort merge join #45327

[SPARK-47146][CORE] Possible thread leak when doing sort merge join #45327

JacobZheng0927 commented Feb 29, 2024 •

edited

zml1206 Feb 29, 2024

zml1206 Feb 29, 2024

mridulm commented Feb 29, 2024

mridulm commented Feb 29, 2024

JacobZheng0927 commented Feb 29, 2024

mridulm Feb 29, 2024

mridulm Feb 29, 2024

dongjoon-hyun Feb 29, 2024 •

edited

mridulm Feb 29, 2024

sunchao Mar 5, 2024

mridulm Mar 5, 2024

dongjoon-hyun commented Feb 29, 2024

dongjoon-hyun Feb 29, 2024

JacobZheng0927 Mar 1, 2024

viirya commented Feb 29, 2024

mridulm left a comment

mridulm commented Mar 5, 2024 •

edited

mridulm commented Mar 5, 2024

JacobZheng0927 commented Mar 5, 2024

sunchao commented Mar 5, 2024

pan3793 Mar 6, 2024

JacobZheng0927 Mar 6, 2024 •

edited

[SPARK-47146][CORE] Possible thread leak when doing sort merge join #45327

[SPARK-47146][CORE] Possible thread leak when doing sort merge join #45327

Conversation

JacobZheng0927 commented Feb 29, 2024 • edited

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mridulm commented Feb 29, 2024

mridulm commented Feb 29, 2024

JacobZheng0927 commented Feb 29, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dongjoon-hyun Feb 29, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dongjoon-hyun commented Feb 29, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viirya commented Feb 29, 2024

mridulm left a comment

Choose a reason for hiding this comment

mridulm commented Mar 5, 2024 • edited

mridulm commented Mar 5, 2024

JacobZheng0927 commented Mar 5, 2024

sunchao commented Mar 5, 2024

Choose a reason for hiding this comment

JacobZheng0927 Mar 6, 2024 • edited

Choose a reason for hiding this comment

JacobZheng0927 commented Feb 29, 2024 •

edited

dongjoon-hyun Feb 29, 2024 •

edited

mridulm commented Mar 5, 2024 •

edited

JacobZheng0927 Mar 6, 2024 •

edited