You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "Ngone51 (via GitHub)" <gi...@apache.org> on 2025/01/08 08:59:07 UTC
[PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Ngone51 opened a new pull request, #49413:
URL: https://github.com/apache/spark/pull/49413
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
This PR fixes the potential stream leak issue by introduing `TaskContext.createResourceUninterruptibly`.
When a task is using `TaskContext.createResourceUninterruptibly` to create the resource, the task would be marked as uninterruptible. Thus, any interruption request during the call to `TaskContext.createResourceUninterruptibly` would be delayed until the creation finishes.
This PR introduces an new lock contention between `Task.kill()` and `TaskContext.createResourceUninterruptibly`. But I think it is acceptable given that both are not on the hot-path.
(I will submmit a followup to apply `TaskContext.createResourceUninterruptibly` in the codebase if this PR is approved by the community.)
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
We had https://github.com/apache/spark/pull/48483 tried to fix the potential stream leak issue by task interruption. It mitigates the issue by using
```scala
def tryInitializeResource[R <: Closeable, T](createResource: => R)(initialize: R => T): T = {
val resource = createResource
try {
initialize(resource)
} catch {
case e: Throwable =>
resource.close()
throw e
}
}
```
But this utility function has an issue that `resource.close()` would leak open resouces if `initialize(resource)` also created some resources internally, especially when `initialize(resource)` is interrupted (See the example of `InterruptionSensitiveInputStream` in the test).
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Added a unit test.
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
No.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591575536
Sure, I'd submit a separate PR for branch-3.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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591589908
Hi @LuciferYang I found we didn't backport https://github.com/apache/spark/pull/48483 to branch-3.5 for the same issue. Generally speaking, this is not a critial bug fix but only for the corner case. Not sure if we want to backport it or not.
@HyukjinKwon Any idea about the backport policy for this kind of fix? is it a requirement or optional?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591702102
Seems unrelated, I tested the result of reverting the current one, but `pyspark.conf` still failed:
- https://github.com/apache/spark/pull/49498
- https://github.com/LuciferYang/spark/actions/runs/12781866942/job/35630526718
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1916036819
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -82,6 +83,13 @@ private[spark] class TaskContextImpl(
// If defined, the corresponding task has been killed and this option contains the reason.
@volatile private var reasonIfKilled: Option[String] = None
+ // The pending interruption request, which is blocked by uninterruptible resource creation.
+ // Should be protected by `TaskContext.synchronized`.
+ private var pendingInterruptRequest: Option[(Option[Thread], String)] = None
+
+ // Whether this task is able to be interrupted. Should be protected by `TaskContext.synchronized`.
Review Comment:
Are you referring to `@volatile`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1916031706
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -82,6 +83,13 @@ private[spark] class TaskContextImpl(
// If defined, the corresponding task has been killed and this option contains the reason.
@volatile private var reasonIfKilled: Option[String] = None
+ // The pending interruption request, which is blocked by uninterruptible resource creation.
+ // Should be protected by `TaskContext.synchronized`.
+ private var pendingInterruptRequest: Option[(Option[Thread], String)] = None
+
+ // Whether this task is able to be interrupted. Should be protected by `TaskContext.synchronized`.
Review Comment:
or maybe mark as `@transient`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1915944557
##########
core/src/main/scala/org/apache/spark/BarrierTaskContext.scala:
##########
@@ -273,6 +274,18 @@ class BarrierTaskContext private[spark] (
}
override private[spark] def getLocalProperties: Properties = taskContext.getLocalProperties
+
+ override private[spark] def interruptible(): Boolean = taskContext.interruptible()
+
+ override private[spark] def pendingInterrupt(threadToInterrupt: Option[Thread], reason: String)
+ : Unit = {
Review Comment:
nit. Indentation?
- https://github.com/databricks/scala-style-guide?tab=readme-ov-file#spacing-and-indentation
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1916499066
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -296,4 +304,39 @@ private[spark] class TaskContextImpl(
private[spark] override def fetchFailed: Option[FetchFailedException] = _fetchFailedException
private[spark] override def getLocalProperties: Properties = localProperties
+
+
+ override def interruptible(): Boolean = TaskContext.synchronized(_interruptible)
+
+ override def pendingInterrupt(threadToInterrupt: Option[Thread], reason: String): Unit = {
Review Comment:
When `threadToInterrupt=None`, it would still "interrupt" the task by invoking `TaskContext.markInterrupted()`. It just won't invoke `Thread.interrupt()` on the task thread.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1913378870
##########
core/src/test/scala/org/apache/spark/JobCancellationSuite.scala:
##########
@@ -712,6 +713,140 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
assert(executionOfInterruptibleCounter.get() < numElements)
}
+ Seq(true, false).foreach { interruptible =>
+
+ val (hint1, hint2) = if (interruptible) {
+ (" not", "")
+ } else {
+ ("", " not")
+ }
+
+ val testName = s"SPARK-50768:$hint1 use TaskContext.createResourceUninterruptibly " +
+ s"would$hint2 cause stream leak on task interruption"
+
+ test(testName) {
+ import org.apache.spark.JobCancellationSuite._
+ withTempDir { dir =>
+
+ // `InterruptionSensitiveInputStream` is designed to easily leak the underlying stream
+ // when task thread interruption happens during its initialization.
+ class InterruptionSensitiveInputStream(fileHint: String) extends InputStream {
+ private var underlying: InputStream = _
+
+ def initialize(): InputStream = {
+ val in: InputStream = new InputStream {
+
+ open()
+
+ private def dumpFile(typeName: String): Unit = {
+ var fileOut: FileOutputStream = null
+ var objOut: ObjectOutputStream = null
+ try {
+ val file = new File(dir, s"$typeName.$fileHint")
+ fileOut = new FileOutputStream(file)
+ objOut = new ObjectOutputStream(fileOut)
+ objOut.writeBoolean(true)
+ objOut.flush()
+ } finally {
+ if (fileOut != null) {
+ fileOut.close()
+ }
+ if (objOut != null) {
+ objOut.close()
+ }
+ }
+
+ }
+
+ private def open(): Unit = {
+ dumpFile("open")
+ }
+
+ override def close(): Unit = {
+ dumpFile("close")
+ }
+
+ override def read(): Int = -1
+ }
+
+ // Leave some time for the task to be interrupted during the
+ // creation of `InterruptionSensitiveInputStream`.
+ Thread.sleep(5000)
Review Comment:
It is necessary to ensure the task is interrupted during `InterruptionSensitiveInputStream#initialize()` so that we can test the leaked stream. Increase the sleep time should less likely be flaky.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591670245
Thank you so much! I looked at, but didn't find any clear clue.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1915969662
##########
core/src/main/scala/org/apache/spark/BarrierTaskContext.scala:
##########
@@ -273,6 +274,18 @@ class BarrierTaskContext private[spark] (
}
override private[spark] def getLocalProperties: Properties = taskContext.getLocalProperties
+
+ override private[spark] def interruptible(): Boolean = taskContext.interruptible()
+
+ override private[spark] def pendingInterrupt(threadToInterrupt: Option[Thread], reason: String)
+ : Unit = {
Review Comment:
I'll fix it with a followup. Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1913361719
##########
core/src/test/scala/org/apache/spark/JobCancellationSuite.scala:
##########
@@ -712,6 +713,140 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
assert(executionOfInterruptibleCounter.get() < numElements)
}
+ Seq(true, false).foreach { interruptible =>
+
+ val (hint1, hint2) = if (interruptible) {
+ (" not", "")
+ } else {
+ ("", " not")
+ }
+
+ val testName = s"SPARK-50768:$hint1 use TaskContext.createResourceUninterruptibly " +
+ s"would$hint2 cause stream leak on task interruption"
+
+ test(testName) {
+ import org.apache.spark.JobCancellationSuite._
+ withTempDir { dir =>
+
+ // `InterruptionSensitiveInputStream` is designed to easily leak the underlying stream
Review Comment:
Updated.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591645190
For the record, after this PR, surprisingly, 3 commits fail consecutively at the same PySpark pipeline.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591573749
@LuciferYang Thanks. Yes, I think so. Do I need send a separate PR?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591702455
no tihs pr is fine
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "xuanyuanking (via GitHub)" <gi...@apache.org>.
xuanyuanking commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1911781043
##########
core/src/test/scala/org/apache/spark/JobCancellationSuite.scala:
##########
@@ -712,6 +713,140 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
assert(executionOfInterruptibleCounter.get() < numElements)
}
+ Seq(true, false).foreach { interruptible =>
+
+ val (hint1, hint2) = if (interruptible) {
+ (" not", "")
+ } else {
+ ("", " not")
+ }
+
+ val testName = s"SPARK-50768:$hint1 use TaskContext.createResourceUninterruptibly " +
+ s"would$hint2 cause stream leak on task interruption"
+
+ test(testName) {
+ import org.apache.spark.JobCancellationSuite._
+ withTempDir { dir =>
+
+ // `InterruptionSensitiveInputStream` is designed to easily leak the underlying stream
Review Comment:
It would be great to mention why InterruptionSensitiveInputStream can easily lead to stream leaks and the key conditions, such as creating internal resources during initialize, as highlighted in the PR description.
##########
core/src/main/scala/org/apache/spark/TaskContext.scala:
##########
@@ -305,4 +305,20 @@ abstract class TaskContext extends Serializable {
/** Gets local properties set upstream in the driver. */
private[spark] def getLocalProperties: Properties
+
+
+ /** Whether the current task is allowed to interrupt. */
+ private[spark] def interruptible(): Boolean
+
+ /**
+ * Pending the interruption request until the task is able to
+ * interrupt after creating the resource uninterruptibly.
+ */
+ private[spark] def pendingInterrupt(threadToInterrupt: Option[Thread], reason: String): Unit
+
+ /**
+ * Creating a closeable resource uninterruptibly. A task is not allowed to interrupt in this
+ * state until the resource creation finishes.
+ */
+ private[spark] def createResourceUninterruptibly[T <: Closeable](resourceBuilder: => T): T
Review Comment:
I noticed that the description mentions a follow-up to apply this function. Could you briefly explain how this method is invoked or provide a simple example?
##########
core/src/main/scala/org/apache/spark/TaskContext.scala:
##########
@@ -305,4 +305,20 @@ abstract class TaskContext extends Serializable {
/** Gets local properties set upstream in the driver. */
private[spark] def getLocalProperties: Properties
+
+
+ /** Whether the current task is allowed to interrupt. */
+ private[spark] def interruptible(): Boolean
+
+ /**
+ * Pending the interruption request until the task is able to
Review Comment:
code style nit
##########
core/src/main/scala/org/apache/spark/TaskContext.scala:
##########
@@ -305,4 +305,20 @@ abstract class TaskContext extends Serializable {
/** Gets local properties set upstream in the driver. */
private[spark] def getLocalProperties: Properties
+
Review Comment:
nit: extra line?
##########
core/src/test/scala/org/apache/spark/JobCancellationSuite.scala:
##########
@@ -712,6 +713,140 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
assert(executionOfInterruptibleCounter.get() < numElements)
}
+ Seq(true, false).foreach { interruptible =>
+
+ val (hint1, hint2) = if (interruptible) {
+ (" not", "")
+ } else {
+ ("", " not")
+ }
+
+ val testName = s"SPARK-50768:$hint1 use TaskContext.createResourceUninterruptibly " +
+ s"would$hint2 cause stream leak on task interruption"
+
+ test(testName) {
+ import org.apache.spark.JobCancellationSuite._
+ withTempDir { dir =>
+
+ // `InterruptionSensitiveInputStream` is designed to easily leak the underlying stream
+ // when task thread interruption happens during its initialization.
+ class InterruptionSensitiveInputStream(fileHint: String) extends InputStream {
+ private var underlying: InputStream = _
+
+ def initialize(): InputStream = {
+ val in: InputStream = new InputStream {
+
+ open()
+
+ private def dumpFile(typeName: String): Unit = {
+ var fileOut: FileOutputStream = null
+ var objOut: ObjectOutputStream = null
+ try {
+ val file = new File(dir, s"$typeName.$fileHint")
+ fileOut = new FileOutputStream(file)
+ objOut = new ObjectOutputStream(fileOut)
+ objOut.writeBoolean(true)
+ objOut.flush()
+ } finally {
+ if (fileOut != null) {
+ fileOut.close()
+ }
+ if (objOut != null) {
+ objOut.close()
+ }
+ }
+
+ }
+
+ private def open(): Unit = {
+ dumpFile("open")
+ }
+
+ override def close(): Unit = {
+ dumpFile("close")
+ }
+
+ override def read(): Int = -1
+ }
+
+ // Leave some time for the task to be interrupted during the
+ // creation of `InterruptionSensitiveInputStream`.
+ Thread.sleep(5000)
Review Comment:
How important is this sleep within the task? Could it potentially make the test flaky?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591668900
im taking a look now for the failure
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1916531539
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -296,4 +304,39 @@ private[spark] class TaskContextImpl(
private[spark] override def fetchFailed: Option[FetchFailedException] = _fetchFailedException
private[spark] override def getLocalProperties: Properties = localProperties
+
+
+ override def interruptible(): Boolean = TaskContext.synchronized(_interruptible)
+
+ override def pendingInterrupt(threadToInterrupt: Option[Thread], reason: String): Unit = {
+ TaskContext.synchronized {
+ pendingInterruptRequest = Some((threadToInterrupt, reason))
+ }
+ }
+
+ def createResourceUninterruptibly[T <: Closeable](resourceBuilder: => T): T = {
+
+ @inline def interruptIfRequired(): Unit = {
+ pendingInterruptRequest.foreach { case (threadToInterrupt, reason) =>
+ markInterrupted(reason)
+ threadToInterrupt.foreach(_.interrupt())
+ }
+ killTaskIfInterrupted()
+ }
+
+ TaskContext.synchronized {
+ interruptIfRequired()
Review Comment:
No. This is for the case where resource creation happens after the task has been marked as `interrupted`. In that case, `pendingInterruptRequest` is `None` and `reasonIfKilled` is not `None`, and `killTaskIfInterrupted()` would throw `TaskKilledException()` to stop the task thread.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1915944672
##########
core/src/main/scala/org/apache/spark/BarrierTaskContext.scala:
##########
@@ -273,6 +274,18 @@ class BarrierTaskContext private[spark] (
}
override private[spark] def getLocalProperties: Properties = taskContext.getLocalProperties
+
+ override private[spark] def interruptible(): Boolean = taskContext.interruptible()
+
+ override private[spark] def pendingInterrupt(threadToInterrupt: Option[Thread], reason: String)
+ : Unit = {
+ taskContext.pendingInterrupt(threadToInterrupt, reason)
+ }
+
+ override private[spark] def createResourceUninterruptibly[T <: Closeable](resourceBuilder: => T)
+ : T = {
Review Comment:
ditto. Indentation.
- https://github.com/databricks/scala-style-guide?tab=readme-ov-file#spacing-and-indentation
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1916546391
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -82,6 +83,13 @@ private[spark] class TaskContextImpl(
// If defined, the corresponding task has been killed and this option contains the reason.
@volatile private var reasonIfKilled: Option[String] = None
+ // The pending interruption request, which is blocked by uninterruptible resource creation.
+ // Should be protected by `TaskContext.synchronized`.
+ private var pendingInterruptRequest: Option[(Option[Thread], String)] = None
+
+ // Whether this task is able to be interrupted. Should be protected by `TaskContext.synchronized`.
Review Comment:
Since we already have `TaskContext.synchronized` protected, `@volatile` could be redundant. And if we use `@volatile` only, it's not thread safety as we could have two threads modify `_interruptible` concurrently.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1916042799
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -296,4 +304,39 @@ private[spark] class TaskContextImpl(
private[spark] override def fetchFailed: Option[FetchFailedException] = _fetchFailedException
private[spark] override def getLocalProperties: Properties = localProperties
+
+
+ override def interruptible(): Boolean = TaskContext.synchronized(_interruptible)
+
+ override def pendingInterrupt(threadToInterrupt: Option[Thread], reason: String): Unit = {
+ TaskContext.synchronized {
+ pendingInterruptRequest = Some((threadToInterrupt, reason))
+ }
+ }
+
+ def createResourceUninterruptibly[T <: Closeable](resourceBuilder: => T): T = {
+
+ @inline def interruptIfRequired(): Unit = {
+ pendingInterruptRequest.foreach { case (threadToInterrupt, reason) =>
+ markInterrupted(reason)
+ threadToInterrupt.foreach(_.interrupt())
+ }
+ killTaskIfInterrupted()
+ }
+
+ TaskContext.synchronized {
+ interruptIfRequired()
Review Comment:
Is it a safeguard that the caller may mistakenly call `pendingInterrupt` even if the task is interruptable?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1911977257
##########
core/src/test/scala/org/apache/spark/JobCancellationSuite.scala:
##########
@@ -712,6 +713,140 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
assert(executionOfInterruptibleCounter.get() < numElements)
}
+ Seq(true, false).foreach { interruptible =>
+
+ val (hint1, hint2) = if (interruptible) {
+ (" not", "")
+ } else {
+ ("", " not")
+ }
+
+ val testName = s"SPARK-50768:$hint1 use TaskContext.createResourceUninterruptibly " +
+ s"would$hint2 cause stream leak on task interruption"
+
+ test(testName) {
+ import org.apache.spark.JobCancellationSuite._
+ withTempDir { dir =>
+
+ // `InterruptionSensitiveInputStream` is designed to easily leak the underlying stream
+ // when task thread interruption happens during its initialization.
+ class InterruptionSensitiveInputStream(fileHint: String) extends InputStream {
+ private var underlying: InputStream = _
+
+ def initialize(): InputStream = {
+ val in: InputStream = new InputStream {
+
+ open()
+
+ private def dumpFile(typeName: String): Unit = {
+ var fileOut: FileOutputStream = null
+ var objOut: ObjectOutputStream = null
+ try {
+ val file = new File(dir, s"$typeName.$fileHint")
+ fileOut = new FileOutputStream(file)
+ objOut = new ObjectOutputStream(fileOut)
+ objOut.writeBoolean(true)
+ objOut.flush()
+ } finally {
+ if (fileOut != null) {
+ fileOut.close()
+ }
+ if (objOut != null) {
+ objOut.close()
+ }
+ }
+
+ }
+
+ private def open(): Unit = {
+ dumpFile("open")
+ }
+
+ override def close(): Unit = {
+ dumpFile("close")
+ }
+
+ override def read(): Int = -1
+ }
+
+ // Leave some time for the task to be interrupted during the
+ // creation of `InterruptionSensitiveInputStream`.
+ Thread.sleep(5000)
Review Comment:
Good point
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591642713
Please hold one backporting for now, @Ngone51 .
It's because this could introduce a flakiness into `master` branch. `branch-3.5` can wait because we don't have Apache Spark 3.5.5 release schedule yet.
In addition, if you can, please give us a confirmation that the failures of current master branch are irrelevant to this PR.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591709652
nono this PR is fine. I will make a PR soon with a PR description with 20 mins
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591728628
here: https://github.com/apache/spark/pull/49500
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1907368576
##########
core/src/main/scala/org/apache/spark/scheduler/Task.scala:
##########
@@ -231,10 +232,23 @@ private[spark] abstract class Task[T](
require(reason != null)
_reasonIfKilled = reason
if (context != null) {
- context.markInterrupted(reason)
- }
- if (interruptThread && taskThread != null) {
- taskThread.interrupt()
+ TaskContext.synchronized {
+ if (context.interruptible()) {
+ context.markInterrupted(reason)
+ if (interruptThread && taskThread != null) {
+ taskThread.interrupt()
+ }
+ } else {
+ val threadToInterrupt = if (interruptThread && taskThread != null) {
Review Comment:
This can be simplified to `if (interruptThread) Option(taskThread) else None`.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -296,4 +304,42 @@ private[spark] class TaskContextImpl(
private[spark] override def fetchFailed: Option[FetchFailedException] = _fetchFailedException
private[spark] override def getLocalProperties: Properties = localProperties
+
+
+ override def interruptible(): Boolean = TaskContext.synchronized(_interruptible)
+
+ override def pendingInterrupt(threadToInterrupt: Option[Thread], reason: String): Unit = {
+ TaskContext.synchronized {
+ pendingInterruptRequest = Some((threadToInterrupt, reason))
+ }
+ }
+
+ def createResourceUninterruptibly[T <: Closeable](resourceBuilder: => T): T = {
+
+ @inline def interruptIfRequired(): Unit = {
+ pendingInterruptRequest.foreach { case (threadToInterrupt, reason) =>
+ markInterrupted(reason)
+ threadToInterrupt.foreach(_.interrupt())
+ }
+ killTaskIfInterrupted()
+ }
+
+ TaskContext.synchronized {
+ interruptIfRequired()
+
+ if (_interruptible) {
Review Comment:
Can we skip the condition check and directly assign the value here?
##########
core/src/test/scala/org/apache/spark/JobCancellationSuite.scala:
##########
@@ -712,6 +713,128 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
assert(executionOfInterruptibleCounter.get() < numElements)
}
+ Seq(true, false).foreach { interruptible =>
+
+ val (hint1, hint2) = if (interruptible) {
+ (" not", "")
+ } else {
+ ("", " not")
+ }
+
+ val testName = s"SPARK-50768:$hint1 use TaskContext.createResourceUninterruptibly " +
+ s"would$hint2 cause stream leak on task interruption"
+
+ test(testName) {
+ import org.apache.spark.JobCancellationSuite._
+ withTempDir { dir =>
+
+ // `InterruptionSensitiveInputStream` is designed to easily leak the underlying stream
+ // when task thread interruption happens during its initialization.
+ class InterruptionSensitiveInputStream(fileHint: String) extends InputStream {
+ private var underlying: InputStream = _
+
+ def initialize(): InputStream = {
+ val in: InputStream = new InputStream {
+
+ open()
+
+ private def dumpFile(typeName: String): Unit = {
+ val file = new File(dir, s"$typeName.$fileHint")
Review Comment:
The `dumpFile` method is best implemented using the `try-with-resources` to ensure the output stream is closed.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1913362187
##########
core/src/main/scala/org/apache/spark/TaskContext.scala:
##########
@@ -305,4 +305,20 @@ abstract class TaskContext extends Serializable {
/** Gets local properties set upstream in the driver. */
private[spark] def getLocalProperties: Properties
+
+
+ /** Whether the current task is allowed to interrupt. */
+ private[spark] def interruptible(): Boolean
+
+ /**
+ * Pending the interruption request until the task is able to
+ * interrupt after creating the resource uninterruptibly.
+ */
+ private[spark] def pendingInterrupt(threadToInterrupt: Option[Thread], reason: String): Unit
+
+ /**
+ * Creating a closeable resource uninterruptibly. A task is not allowed to interrupt in this
+ * state until the resource creation finishes.
+ */
+ private[spark] def createResourceUninterruptibly[T <: Closeable](resourceBuilder: => T): T
Review Comment:
Updated the comment with an example.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1916744335
##########
core/src/main/scala/org/apache/spark/BarrierTaskContext.scala:
##########
@@ -273,6 +274,18 @@ class BarrierTaskContext private[spark] (
}
override private[spark] def getLocalProperties: Properties = taskContext.getLocalProperties
+
+ override private[spark] def interruptible(): Boolean = taskContext.interruptible()
+
+ override private[spark] def pendingInterrupt(threadToInterrupt: Option[Thread], reason: String)
+ : Unit = {
Review Comment:
Fixed at https://github.com/apache/spark/pull/49508
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591574286
Merged into master for Spark 4.0. Thanks @Ngone51 @xuanyuanking and @HyukjinKwon .
@Ngone51 Due to code conflicts, it cannot be directly merged into branch-3.5. If needed, please submit a separate pr. Thanks ~
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang closed pull request #49413: [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption
URL: https://github.com/apache/spark/pull/49413
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591637840
please go ahead. I think it's fien
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591570698
@Ngone51 Does branch-3.5 also need this bug fix?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2593001456
FYI I created a followup PR (https://github.com/apache/spark/pull/49508) to use `TaskContext.createResourceUninterruptibly()` where it applies.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591707206
@HyukjinKwon Are you saying this pr caused the test to fail?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591702196
found the cuase. working on the fix
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on PR #49413:
URL: https://github.com/apache/spark/pull/49413#issuecomment-2591556388
Could someone help merge this PR if we're good to proceed? Or if I can also merge myself with enough approvals?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50768][CORE] Introduce TaskContext.createResourceUninterruptibly to avoid stream leak by task interruption [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #49413:
URL: https://github.com/apache/spark/pull/49413#discussion_r1916037805
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -296,4 +304,39 @@ private[spark] class TaskContextImpl(
private[spark] override def fetchFailed: Option[FetchFailedException] = _fetchFailedException
private[spark] override def getLocalProperties: Properties = localProperties
+
+
+ override def interruptible(): Boolean = TaskContext.synchronized(_interruptible)
+
+ override def pendingInterrupt(threadToInterrupt: Option[Thread], reason: String): Unit = {
Review Comment:
This API looks weird if `threadToInterrupt` is `None`, as there is nothing to interrupt.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org