You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "anchovYu (via GitHub)" <gi...@apache.org> on 2023/11/21 06:44:44 UTC
[PR] [SPARK-46021] Support cancel future jobs belonging to a job group [spark]
anchovYu opened a new pull request, #43926:
URL: https://github.com/apache/spark/pull/43926
<!--
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
'core/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.
-->
### 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.
-->
### 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'.
-->
### 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.
-->
### 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.
-->
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1402846847
##########
core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala:
##########
@@ -221,6 +221,18 @@ private[spark] object SparkCoreErrors {
new NoSuchElementException(id)
}
+ def sparkJobCancelled(jobId: Int, reason: String, e: Exception): SparkException = {
+ new SparkException(
+ errorClass = "SPARK_JOB_CANCELLED",
+ messageParameters = Map("jobId" -> jobId.toString, "reason" -> reason),
+ cause = e
+ )
+ }
+
+ def sparkJobCancelledAsPartOfJobGroupError(jobId: Int, jobGroupId: String): SparkException = {
+ sparkJobCancelled(jobId, s"part of cancelled job group $jobGroupId", null)
Review Comment:
I'm a bit confused. Shall the driver exit if throw a spark exception?
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1400170021
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -1728,6 +1728,16 @@ package object config {
.checkValue(v => v > 0, "The max failures should be a positive value.")
.createWithDefault(40)
+ private[spark] val CANCELLED_JOB_GROUP_SET_SIZE =
+ ConfigBuilder("spark.scheduler.job.cancelledJobGroupSet.size")
+ .doc("The size of the set to store cancelled job groups, if the job group is cancelled " +
+ "with cancelFutureJobs = true. If the size of the set exceeds this value, the oldest job " +
+ "group will be removed from the set.")
Review Comment:
`The maximum number of job groups to track for canceling job groups and their future jobs.`
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "JoshRosen (via GitHub)" <gi...@apache.org>.
JoshRosen commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1402712326
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1264,14 +1280,24 @@ private[spark] class DAGScheduler(
listenerBus.post(SparkListenerTaskGettingResult(taskInfo))
}
- private[scheduler] def handleJobSubmitted(jobId: Int,
+ private[scheduler] def handleJobSubmitted(
+ jobId: Int,
finalRDD: RDD[_],
func: (TaskContext, Iterator[_]) => _,
partitions: Array[Int],
callSite: CallSite,
listener: JobListener,
artifacts: JobArtifactSet,
properties: Properties): Unit = {
+ // If this job belongs to a cancelled job group, skip running it
+ val jobGroupIdOpt = Option(properties).map(_.getProperty(SparkContext.SPARK_JOB_GROUP_ID))
+ if (jobGroupIdOpt.exists(cancelledJobGroups.contains(_))) {
Review Comment:
+1, I agree with @anchovYu's assessment:
- This `handleJobSubmitted` method and the `handleJobGroupCancelled` method both run in the single-threaded event thread.
- Consider both potential interleavings:
- If `handleJobGroupCancelled` runs first then the group will be marked as cancelled and `handleJobSubmitted` will skip the job.
- If `handleJobSubmitted` runs first then `handleJobGroupCancelled` will see a running job and cancel it.
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1400170021
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -1728,6 +1728,16 @@ package object config {
.checkValue(v => v > 0, "The max failures should be a positive value.")
.createWithDefault(40)
+ private[spark] val CANCELLED_JOB_GROUP_SET_SIZE =
+ ConfigBuilder("spark.scheduler.job.cancelledJobGroupSet.size")
+ .doc("The size of the set to store cancelled job groups, if the job group is cancelled " +
+ "with cancelFutureJobs = true. If the size of the set exceeds this value, the oldest job " +
+ "group will be removed from the set.")
Review Comment:
`The maximum number of job groups to track for cancelling job groups and their future jobs. If ...`
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1400179808
##########
core/src/test/scala/org/apache/spark/JobCancellationSuite.scala:
##########
@@ -153,6 +153,91 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
assert(jobB.get() === 100)
}
+ test("if cancel job group and future jobs, skip running jobs in the same job group") {
+ sc = new SparkContext("local[2]", "test")
+
+ val sem = new Semaphore(0)
+ sc.addSparkListener(new SparkListener {
+ override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
+ sem.release()
+ }
+ })
+
+ val jobGroupName = "job-group"
+
+ val jobA = Future {
+ sc.setJobGroup(jobGroupName, "")
+ sc.parallelize(1 to 1000).map { i => Thread.sleep (100); i}.count()
+ }
+ // Block until jobA starts
+ sem.acquire(1)
+ // Cancel the job group and future jobs
+ sc.cancelJobGroupAndFutureJobs(jobGroupName)
+ ThreadUtils.awaitReady(jobA, Duration.Inf).failed.foreach { case e: SparkException =>
+ checkError(
+ exception = e,
+ errorClass = "SPARK_JOB_CANCELLED",
+ sqlState = "XXKDA",
+ parameters = scala.collection.immutable.Map(
+ "jobId" -> "0",
+ "reason" -> s"part of cancelled job group $jobGroupName")
+ )
+ }
+
+ // jobB in the same job group will not run
+ val jobB = Future {
+ sc.setJobGroup(jobGroupName, "")
+ sc.parallelize(1 to 100).count()
+ }
+ ThreadUtils.awaitReady(jobB, Duration.Inf).failed.foreach { case e: SparkException =>
+ checkError(
+ exception = e,
+ errorClass = "SPARK_JOB_CANCELLED",
+ sqlState = "XXKDA",
+ parameters = scala.collection.immutable.Map(
+ "jobId" -> "1",
+ "reason" -> s"part of cancelled job group $jobGroupName")
+ )
+ }
+
+ // jobC in a different job group should run
+ val jobC = Future {
+ sc.setJobGroup("another-job-group", "")
+ sc.parallelize(1 to 100).count()
+ }
+ assert(ThreadUtils.awaitResult(jobC, Duration.Inf) == 100)
+ }
+
+ test("only keeps limited number of cancelled job groups") {
+ sc = new SparkContext("local[2]", "test")
Review Comment:
to make this test run faster, can we set `CANCELLED_JOB_GROUP_SET_SIZE` to a small numner?
##########
core/src/test/scala/org/apache/spark/JobCancellationSuite.scala:
##########
@@ -153,6 +153,91 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
assert(jobB.get() === 100)
}
+ test("if cancel job group and future jobs, skip running jobs in the same job group") {
+ sc = new SparkContext("local[2]", "test")
+
+ val sem = new Semaphore(0)
+ sc.addSparkListener(new SparkListener {
+ override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
+ sem.release()
+ }
+ })
+
+ val jobGroupName = "job-group"
+
+ val jobA = Future {
+ sc.setJobGroup(jobGroupName, "")
+ sc.parallelize(1 to 1000).map { i => Thread.sleep (100); i}.count()
+ }
+ // Block until jobA starts
+ sem.acquire(1)
+ // Cancel the job group and future jobs
+ sc.cancelJobGroupAndFutureJobs(jobGroupName)
+ ThreadUtils.awaitReady(jobA, Duration.Inf).failed.foreach { case e: SparkException =>
+ checkError(
+ exception = e,
+ errorClass = "SPARK_JOB_CANCELLED",
+ sqlState = "XXKDA",
+ parameters = scala.collection.immutable.Map(
+ "jobId" -> "0",
+ "reason" -> s"part of cancelled job group $jobGroupName")
+ )
+ }
+
+ // jobB in the same job group will not run
+ val jobB = Future {
+ sc.setJobGroup(jobGroupName, "")
+ sc.parallelize(1 to 100).count()
+ }
+ ThreadUtils.awaitReady(jobB, Duration.Inf).failed.foreach { case e: SparkException =>
+ checkError(
+ exception = e,
+ errorClass = "SPARK_JOB_CANCELLED",
+ sqlState = "XXKDA",
+ parameters = scala.collection.immutable.Map(
+ "jobId" -> "1",
+ "reason" -> s"part of cancelled job group $jobGroupName")
+ )
+ }
+
+ // jobC in a different job group should run
+ val jobC = Future {
+ sc.setJobGroup("another-job-group", "")
+ sc.parallelize(1 to 100).count()
+ }
+ assert(ThreadUtils.awaitResult(jobC, Duration.Inf) == 100)
+ }
+
+ test("only keeps limited number of cancelled job groups") {
+ sc = new SparkContext("local[2]", "test")
Review Comment:
to make this test run faster, can we set `CANCELLED_JOB_GROUP_SET_SIZE` to a small number?
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1401555960
##########
common/utils/src/main/resources/error/error-classes.json:
##########
@@ -2961,6 +2961,12 @@
],
"sqlState" : "42601"
},
+ "SPARK_JOB_CANCELLED" : {
+ "message" : [
+ "Job <jobId> cancelled <reason>"
+ ],
+ "sqlState" : "XXKDA"
Review Comment:
This is my last concern. We need this error class as users can trigger it using RDD APIs. But RDD API is kind of internal now, and we don't expect users to call them directly.
To confirm, SQL users won't hit it, right? Image the user canceled a query, but the AQE loop is still running and submitted some more jobs, this failure will be swallowed and not propagated to users, rigjt? also cc @liuzqt
##########
common/utils/src/main/resources/error/error-classes.json:
##########
@@ -2961,6 +2961,12 @@
],
"sqlState" : "42601"
},
+ "SPARK_JOB_CANCELLED" : {
+ "message" : [
+ "Job <jobId> cancelled <reason>"
+ ],
+ "sqlState" : "XXKDA"
Review Comment:
This is my last concern. We need this error class as users can trigger it using RDD APIs. But RDD API is kind of internal now, and we don't expect users to call them directly.
To confirm, SQL users won't hit it, right? Image the user canceled a query, but the AQE loop is still running and submitted some more jobs, this failure will be swallowed and not propagated to users, right? also cc @liuzqt
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1402846847
##########
core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala:
##########
@@ -221,6 +221,18 @@ private[spark] object SparkCoreErrors {
new NoSuchElementException(id)
}
+ def sparkJobCancelled(jobId: Int, reason: String, e: Exception): SparkException = {
+ new SparkException(
+ errorClass = "SPARK_JOB_CANCELLED",
+ messageParameters = Map("jobId" -> jobId.toString, "reason" -> reason),
+ cause = e
+ )
+ }
+
+ def sparkJobCancelledAsPartOfJobGroupError(jobId: Int, jobGroupId: String): SparkException = {
+ sparkJobCancelled(jobId, s"part of cancelled job group $jobGroupId", null)
Review Comment:
I'm a bit confused. Does driver exit if throw a spark exception?
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1402846242
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1264,14 +1280,24 @@ private[spark] class DAGScheduler(
listenerBus.post(SparkListenerTaskGettingResult(taskInfo))
}
- private[scheduler] def handleJobSubmitted(jobId: Int,
+ private[scheduler] def handleJobSubmitted(
+ jobId: Int,
finalRDD: RDD[_],
func: (TaskContext, Iterator[_]) => _,
partitions: Array[Int],
callSite: CallSite,
listener: JobListener,
artifacts: JobArtifactSet,
properties: Properties): Unit = {
+ // If this job belongs to a cancelled job group, skip running it
+ val jobGroupIdOpt = Option(properties).map(_.getProperty(SparkContext.SPARK_JOB_GROUP_ID))
+ if (jobGroupIdOpt.exists(cancelledJobGroups.contains(_))) {
Review Comment:
@JoshRosen Thank you for the description. I got it now.
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1400170021
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -1728,6 +1728,16 @@ package object config {
.checkValue(v => v > 0, "The max failures should be a positive value.")
.createWithDefault(40)
+ private[spark] val CANCELLED_JOB_GROUP_SET_SIZE =
+ ConfigBuilder("spark.scheduler.job.cancelledJobGroupSet.size")
+ .doc("The size of the set to store cancelled job groups, if the job group is cancelled " +
+ "with cancelFutureJobs = true. If the size of the set exceeds this value, the oldest job " +
+ "group will be removed from the set.")
Review Comment:
`The maximum number of job groups to track for canceling job groups and their future jobs. If the maximum number is hit, the oldest job group will no longer be tracked.`
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "anchovYu (via GitHub)" <gi...@apache.org>.
anchovYu commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1401370271
##########
common/utils/src/main/resources/error/error-classes.json:
##########
@@ -2961,6 +2961,12 @@
],
"sqlState" : "42601"
},
+ "SPARK_JOB_CANCELLED" : {
+ "message" : [
+ "Job <jobId> cancelled <reason>"
+ ],
+ "sqlState" : "XXKDA"
Review Comment:
In the engine with cancellation it's straightforward, that if user cancel the command, the error could be designed to pop to the top.
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1400169105
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -1728,6 +1728,16 @@ package object config {
.checkValue(v => v > 0, "The max failures should be a positive value.")
.createWithDefault(40)
+ private[spark] val CANCELLED_JOB_GROUP_SET_SIZE =
+ ConfigBuilder("spark.scheduler.job.cancelledJobGroupSet.size")
Review Comment:
maybe `spark.scheduler.numJobGroupsToTrackForCancel`?
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1400176679
##########
core/src/test/scala/org/apache/spark/JobCancellationSuite.scala:
##########
@@ -153,6 +153,91 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
assert(jobB.get() === 100)
}
+ test("if cancel job group and future jobs, skip running jobs in the same job group") {
+ sc = new SparkContext("local[2]", "test")
+
+ val sem = new Semaphore(0)
+ sc.addSparkListener(new SparkListener {
+ override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
+ sem.release()
+ }
+ })
+
+ val jobGroupName = "job-group"
+
+ val jobA = Future {
Review Comment:
why do we run this job in a separated 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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1400176679
##########
core/src/test/scala/org/apache/spark/JobCancellationSuite.scala:
##########
@@ -153,6 +153,91 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
assert(jobB.get() === 100)
}
+ test("if cancel job group and future jobs, skip running jobs in the same job group") {
+ sc = new SparkContext("local[2]", "test")
+
+ val sem = new Semaphore(0)
+ sc.addSparkListener(new SparkListener {
+ override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
+ sem.release()
+ }
+ })
+
+ val jobGroupName = "job-group"
+
+ val jobA = Future {
Review Comment:
why do we run this job in a separated 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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1400172725
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -1728,6 +1728,16 @@ package object config {
.checkValue(v => v > 0, "The max failures should be a positive value.")
.createWithDefault(40)
+ private[spark] val CANCELLED_JOB_GROUP_SET_SIZE =
+ ConfigBuilder("spark.scheduler.job.cancelledJobGroupSet.size")
+ .doc("The size of the set to store cancelled job groups, if the job group is cancelled " +
+ "with cancelFutureJobs = true. If the size of the set exceeds this value, the oldest job " +
+ "group will be removed from the set.")
+ .version("3.5.0")
Review Comment:
```suggestion
.version("4.0.0")
```
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "anchovYu (via GitHub)" <gi...@apache.org>.
anchovYu commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1402477697
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1264,14 +1280,24 @@ private[spark] class DAGScheduler(
listenerBus.post(SparkListenerTaskGettingResult(taskInfo))
}
- private[scheduler] def handleJobSubmitted(jobId: Int,
+ private[scheduler] def handleJobSubmitted(
+ jobId: Int,
finalRDD: RDD[_],
func: (TaskContext, Iterator[_]) => _,
partitions: Array[Int],
callSite: CallSite,
listener: JobListener,
artifacts: JobArtifactSet,
properties: Properties): Unit = {
+ // If this job belongs to a cancelled job group, skip running it
+ val jobGroupIdOpt = Option(properties).map(_.getProperty(SparkContext.SPARK_JOB_GROUP_ID))
+ if (jobGroupIdOpt.exists(cancelledJobGroups.contains(_))) {
Review Comment:
The `handler` function are run in the synchronized block, so the "`cancelJobGroup(cancelFutureJobs = true)` => the current active jobs will be cancelled + this job group is added to the set" should be run atomically. And later the future job belonging to this job group will be cancelled on submission.
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1400170021
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -1728,6 +1728,16 @@ package object config {
.checkValue(v => v > 0, "The max failures should be a positive value.")
.createWithDefault(40)
+ private[spark] val CANCELLED_JOB_GROUP_SET_SIZE =
+ ConfigBuilder("spark.scheduler.job.cancelledJobGroupSet.size")
+ .doc("The size of the set to store cancelled job groups, if the job group is cancelled " +
+ "with cancelFutureJobs = true. If the size of the set exceeds this value, the oldest job " +
+ "group will be removed from the set.")
Review Comment:
`The maximum number of job groups to track for canceling job groups and their future jobs. If the maximum number is hit, the oldest job group will no longer be tracked and its future jobs may not be cancelled. `
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "JoshRosen (via GitHub)" <gi...@apache.org>.
JoshRosen commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1400125810
##########
common/utils/src/main/resources/error/README.md:
##########
@@ -1347,7 +1347,7 @@ The following SQLSTATEs are collated from:
|XX001 |XX |Internal Error |001 |data_corrupted |PostgreSQL |N |PostgreSQL Redshift |
|XX002 |XX |Internal Error |002 |index_corrupted |PostgreSQL |N |PostgreSQL Redshift |
|XXKD0 |XX |Internal Error |KD0 |Analysis - Bad plan |Databricks |N |Databricks |
-|XXKDA |XX |Internal Error |KAS |Scheduler (Aether Scheduler) |Databricks |N |Databricks |
Review Comment:
Given that this is adding an error class for an existing OSS Apache Spark error, we should probably use a non-Databricks error code here.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -169,6 +169,12 @@ private[spark] class DAGScheduler(
private[scheduler] val activeJobs = new HashSet[ActiveJob]
+ // Job groups that are canceled with `cancelFutureJobs` as true, with at most
+ // `CANCELLED_JOB_GROUP_SET_SIZE` stored. On a new job submission, if the job group is in this
+ // set, the job will be immediately canceled.
Review Comment:
Spark isn't perfectly consistent about "cancelled" vs "canceled" (two 'L's or one) (and it [seems this is a British vs. American english difference](https://www.merriam-webster.com/grammar/canceled-or-cancelled)), but it looks like Spark's existing user-facing messages generally use the two-L version, e.g. at
https://github.com/apache/spark/blob/d908a212ef6144338f075354b4636a13f1172478/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L2730
Given this, I think we should aim for consistency with the existing user-facing messages and use "cancelled" throughout.
##########
core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala:
##########
@@ -221,6 +221,18 @@ private[spark] object SparkCoreErrors {
new NoSuchElementException(id)
}
+ def sparkJobCancelled(jobId: Int, reason: String, e: Exception): SparkException = {
+ new SparkException(
+ errorClass = "SPARK_JOB_CANCELLED",
+ messageParameters = Map("jobId" -> jobId.toString, "reason" -> reason),
+ cause = e
+ )
+ }
+
+ def sparkJobCancelledAsPartOfJobGroupError(jobId: Int, jobGroupId: String): SparkException = {
+ sparkJobCancelled(jobId, s"part of cancelled job group $jobGroupId", null)
Review Comment:
It looks like this is consistent with the message thrown in the other code path at
https://github.com/apache/spark/blob/021a5e66f530d6a5621782ea6caaf7f8f3e11ae3/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L1207-L1209
which is important in case end-user code is matching on these exception messages (since we previously didn't provide a special exception or error class for distinguishing these errors) 👍
##########
core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala:
##########
@@ -221,6 +221,18 @@ private[spark] object SparkCoreErrors {
new NoSuchElementException(id)
}
+ def sparkJobCancelled(jobId: Int, reason: String, e: Exception): SparkException = {
+ new SparkException(
+ errorClass = "SPARK_JOB_CANCELLED",
+ messageParameters = Map("jobId" -> jobId.toString, "reason" -> reason),
+ cause = e
+ )
+ }
+
+ def sparkJobCancelledAsPartOfJobGroupError(jobId: Int, jobGroupId: String): SparkException = {
+ sparkJobCancelled(jobId, s"part of cancelled job group $jobGroupId", null)
Review Comment:
It looks like this is consistent with the message thrown in the other code path at
https://github.com/apache/spark/blob/021a5e66f530d6a5621782ea6caaf7f8f3e11ae3/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L1207-L1209
which is important in case end-user code is matching on these exception messages (since we previously didn't provide a special exception or error class for distinguishing these errors) 👍
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "anchovYu (via GitHub)" <gi...@apache.org>.
anchovYu commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1401373298
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -169,6 +169,12 @@ private[spark] class DAGScheduler(
private[scheduler] val activeJobs = new HashSet[ActiveJob]
+ // Job groups that are canceled with `cancelFutureJobs` as true, with at most
+ // `CANCELLED_JOB_GROUP_SET_SIZE` stored. On a new job submission, if the job group is in this
+ // set, the job will be immediately canceled.
Review Comment:
I will update the PR to all use double l.
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1401569798
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1264,14 +1280,24 @@ private[spark] class DAGScheduler(
listenerBus.post(SparkListenerTaskGettingResult(taskInfo))
}
- private[scheduler] def handleJobSubmitted(jobId: Int,
+ private[scheduler] def handleJobSubmitted(
+ jobId: Int,
finalRDD: RDD[_],
func: (TaskContext, Iterator[_]) => _,
partitions: Array[Int],
callSite: CallSite,
listener: JobListener,
artifacts: JobArtifactSet,
properties: Properties): Unit = {
+ // If this job belongs to a cancelled job group, skip running it
+ val jobGroupIdOpt = Option(properties).map(_.getProperty(SparkContext.SPARK_JOB_GROUP_ID))
+ if (jobGroupIdOpt.exists(cancelledJobGroups.contains(_))) {
Review Comment:
Because the schedule thread and event thread are asynchronous, it seems doesn't cancel the future job not always effective.
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1400178753
##########
core/src/test/scala/org/apache/spark/JobCancellationSuite.scala:
##########
@@ -153,6 +153,91 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
assert(jobB.get() === 100)
}
+ test("if cancel job group and future jobs, skip running jobs in the same job group") {
+ sc = new SparkContext("local[2]", "test")
+
+ val sem = new Semaphore(0)
+ sc.addSparkListener(new SparkListener {
+ override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
+ sem.release()
+ }
+ })
+
+ val jobGroupName = "job-group"
+
+ val jobA = Future {
+ sc.setJobGroup(jobGroupName, "")
+ sc.parallelize(1 to 1000).map { i => Thread.sleep (100); i}.count()
+ }
+ // Block until jobA starts
+ sem.acquire(1)
+ // Cancel the job group and future jobs
+ sc.cancelJobGroupAndFutureJobs(jobGroupName)
+ ThreadUtils.awaitReady(jobA, Duration.Inf).failed.foreach { case e: SparkException =>
+ checkError(
+ exception = e,
+ errorClass = "SPARK_JOB_CANCELLED",
+ sqlState = "XXKDA",
+ parameters = scala.collection.immutable.Map(
+ "jobId" -> "0",
+ "reason" -> s"part of cancelled job group $jobGroupName")
+ )
+ }
+
+ // jobB in the same job group will not run
+ val jobB = Future {
Review Comment:
for the jobB test, I think we can run it with the current thread and check its exception:
```
val e = intercept[SparkException] {
// job b...
}
checkError...
```
##########
core/src/test/scala/org/apache/spark/JobCancellationSuite.scala:
##########
@@ -153,6 +153,91 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
assert(jobB.get() === 100)
}
+ test("if cancel job group and future jobs, skip running jobs in the same job group") {
+ sc = new SparkContext("local[2]", "test")
+
+ val sem = new Semaphore(0)
+ sc.addSparkListener(new SparkListener {
+ override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
+ sem.release()
+ }
+ })
+
+ val jobGroupName = "job-group"
+
+ val jobA = Future {
+ sc.setJobGroup(jobGroupName, "")
+ sc.parallelize(1 to 1000).map { i => Thread.sleep (100); i}.count()
+ }
+ // Block until jobA starts
+ sem.acquire(1)
+ // Cancel the job group and future jobs
+ sc.cancelJobGroupAndFutureJobs(jobGroupName)
+ ThreadUtils.awaitReady(jobA, Duration.Inf).failed.foreach { case e: SparkException =>
+ checkError(
+ exception = e,
+ errorClass = "SPARK_JOB_CANCELLED",
+ sqlState = "XXKDA",
+ parameters = scala.collection.immutable.Map(
+ "jobId" -> "0",
+ "reason" -> s"part of cancelled job group $jobGroupName")
+ )
+ }
+
+ // jobB in the same job group will not run
+ val jobB = Future {
+ sc.setJobGroup(jobGroupName, "")
+ sc.parallelize(1 to 100).count()
+ }
+ ThreadUtils.awaitReady(jobB, Duration.Inf).failed.foreach { case e: SparkException =>
+ checkError(
+ exception = e,
+ errorClass = "SPARK_JOB_CANCELLED",
+ sqlState = "XXKDA",
+ parameters = scala.collection.immutable.Map(
+ "jobId" -> "1",
+ "reason" -> s"part of cancelled job group $jobGroupName")
+ )
+ }
+
+ // jobC in a different job group should run
Review Comment:
ditto
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "anchovYu (via GitHub)" <gi...@apache.org>.
anchovYu commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1401368945
##########
common/utils/src/main/resources/error/README.md:
##########
@@ -1347,7 +1347,7 @@ The following SQLSTATEs are collated from:
|XX001 |XX |Internal Error |001 |data_corrupted |PostgreSQL |N |PostgreSQL Redshift |
|XX002 |XX |Internal Error |002 |index_corrupted |PostgreSQL |N |PostgreSQL Redshift |
|XXKD0 |XX |Internal Error |KD0 |Analysis - Bad plan |Databricks |N |Databricks |
-|XXKDA |XX |Internal Error |KAS |Scheduler (Aether Scheduler) |Databricks |N |Databricks |
Review Comment:
I think 'Databricks' here means the 'origin' and 'used by', similar to errors that originates from other engines. Taken an example of
```
|XXKD0 |XX |Internal Error |KD0 |Analysis - Bad plan |Databricks |N |Databricks
```
, it's used in open source error class:
```
"PLAN_VALIDATION_FAILED_RULE_IN_BATCH" : {
"message" : [
"Rule <rule> in batch <batch> generated an invalid plan: <reason>"
],
"sqlState" : "XXKD0"
},
```
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "JoshRosen (via GitHub)" <gi...@apache.org>.
JoshRosen commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1402709319
##########
common/utils/src/main/resources/error/error-classes.json:
##########
@@ -2961,6 +2961,12 @@
],
"sqlState" : "42601"
},
+ "SPARK_JOB_CANCELLED" : {
+ "message" : [
+ "Job <jobId> cancelled <reason>"
+ ],
+ "sqlState" : "XXKDA"
Review Comment:
In theory a SQL user might see this if they are submitting a query to the ThriftServer and then the query's underlying Spark job is canceled via the Web UI (unless the ThriftServer wraps the resulting query failure into some other exception or error code).
Regardless of the likelihood of this error occurring in SQL, I think it's still beneficial to migrate it to the error classes framework so that it is easier to match on cancellation exceptions without having to do string matching.
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1400170021
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -1728,6 +1728,16 @@ package object config {
.checkValue(v => v > 0, "The max failures should be a positive value.")
.createWithDefault(40)
+ private[spark] val CANCELLED_JOB_GROUP_SET_SIZE =
+ ConfigBuilder("spark.scheduler.job.cancelledJobGroupSet.size")
+ .doc("The size of the set to store cancelled job groups, if the job group is cancelled " +
+ "with cancelFutureJobs = true. If the size of the set exceeds this value, the oldest job " +
+ "group will be removed from the set.")
Review Comment:
`The maximum number of job groups to track for cancelling job groups and their future jobs.`
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #43926:
URL: https://github.com/apache/spark/pull/43926#discussion_r1400167030
##########
common/utils/src/main/resources/error/error-classes.json:
##########
@@ -2961,6 +2961,12 @@
],
"sqlState" : "42601"
},
+ "SPARK_JOB_CANCELLED" : {
+ "message" : [
+ "Job <jobId> cancelled <reason>"
+ ],
+ "sqlState" : "XXKDA"
Review Comment:
How can a user hit this bug using SQL?
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #43926: [SPARK-46021][CORE] Support cancel future jobs belonging to a job group
URL: https://github.com/apache/spark/pull/43926
--
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-46021][CORE] Support cancel future jobs belonging to a job group [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #43926:
URL: https://github.com/apache/spark/pull/43926#issuecomment-1824349611
thanks, merging to master!
--
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