This is an automated email from the ASF dual-hosted git repository.
jackylee-ch pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 40dc7920e0 [GLUTEN-12616][CORE] Guard SparkResourceUtil.getTaskSlots
against non-positive task cpus (#12617)
40dc7920e0 is described below
commit 40dc7920e06f69fe9e656a781208ec461f0752e0
Author: YangJie <[email protected]>
AuthorDate: Tue Jul 28 10:31:55 2026 +0800
[GLUTEN-12616][CORE] Guard SparkResourceUtil.getTaskSlots against
non-positive task cpus (#12617)
---
.../org/apache/spark/util/SparkResourceUtil.scala | 11 +++-
.../apache/spark/util/SparkResourceUtilSuite.scala | 68 ++++++++++++++++++++++
2 files changed, 78 insertions(+), 1 deletion(-)
diff --git
a/gluten-core/src/main/scala/org/apache/spark/util/SparkResourceUtil.scala
b/gluten-core/src/main/scala/org/apache/spark/util/SparkResourceUtil.scala
index 6be1791fa2..b25a174c12 100644
--- a/gluten-core/src/main/scala/org/apache/spark/util/SparkResourceUtil.scala
+++ b/gluten-core/src/main/scala/org/apache/spark/util/SparkResourceUtil.scala
@@ -77,8 +77,17 @@ object SparkResourceUtil extends Logging {
def getTaskSlots(conf: SparkConf): Int = {
val executorCores = SparkResourceUtil.getExecutorCores(conf)
+ // spark.task.cpus is read raw here, which bypasses Spark's own
checkValue(_ > 0) (and on
+ // Spark < 4.2 that positivity check does not exist at all). getTaskSlots
runs during driver
+ // plugin init, before Spark validates the value, so fail fast on a
non-positive setting rather
+ // than dividing by it: a zero would throw an opaque "/ by zero"
ArithmeticException and a
+ // negative would silently produce negative task slots and off-heap
budgets.
val taskCores = conf.getInt("spark.task.cpus", 1)
- executorCores / taskCores
+ require(taskCores > 0, s"spark.task.cpus should be positive, but was
$taskCores")
+ // Floor at one slot. spark.task.cpus > executor cores is an invalid
combination that Spark
+ // rejects later with a dedicated message, but callers divide by the slot
count during init, so
+ // a 0 quotient would throw before Spark can report the real
misconfiguration.
+ Math.max(executorCores / taskCores, 1)
}
def isLocalMaster(conf: SparkConf): Boolean = {
diff --git
a/gluten-core/src/test/scala/org/apache/spark/util/SparkResourceUtilSuite.scala
b/gluten-core/src/test/scala/org/apache/spark/util/SparkResourceUtilSuite.scala
new file mode 100644
index 0000000000..bad2153ddc
--- /dev/null
+++
b/gluten-core/src/test/scala/org/apache/spark/util/SparkResourceUtilSuite.scala
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.util
+
+import org.apache.spark.SparkConf
+
+import org.scalatest.funsuite.AnyFunSuite
+
+class SparkResourceUtilSuite extends AnyFunSuite {
+
+ test("getTaskSlots floors at one when task cpus exceed executor cores") {
+ // spark.task.cpus > executor cores is an invalid config that Spark
rejects later, but Gluten
+ // reads task slots during plugin init and divides by the result.
Returning 0 here makes that
+ // division throw ArithmeticException before Spark's
validateTaskCpusLargeEnough can report the
+ // real misconfiguration.
+ val conf = new SparkConf(false)
+ .set("spark.master", "local[1]")
+ .set("spark.task.cpus", "2")
+ assert(SparkResourceUtil.getTaskSlots(conf) == 1)
+ }
+
+ test("getTaskSlots fails fast when task cpus is zero") {
+ // spark.task.cpus is read via raw conf.getInt, which bypasses Spark's
checkValue(_ > 0) (a
+ // check that only exists on Spark >= 4.2), so a zero value must not reach
the division. Fail
+ // fast with a clear message instead of an opaque "/ by zero"
ArithmeticException.
+ val conf = new SparkConf(false)
+ .set("spark.master", "local[8]")
+ .set("spark.task.cpus", "0")
+ val e =
intercept[IllegalArgumentException](SparkResourceUtil.getTaskSlots(conf))
+ assert(e.getMessage.contains("spark.task.cpus should be positive"))
+ }
+
+ test("getTaskSlots fails fast when task cpus is negative") {
+ // A negative spark.task.cpus would otherwise divide to a negative slot
count, silently yielding
+ // negative per-task off-heap budgets. Fail fast rather than propagate the
bad value.
+ val conf = new SparkConf(false)
+ .set("spark.master", "local[8]")
+ .set("spark.task.cpus", "-2")
+ val e =
intercept[IllegalArgumentException](SparkResourceUtil.getTaskSlots(conf))
+ assert(e.getMessage.contains("spark.task.cpus should be positive"))
+ }
+
+ test("getTaskSlots divides executor cores by task cpus") {
+ val conf = new SparkConf(false)
+ .set("spark.master", "local[8]")
+ .set("spark.task.cpus", "2")
+ assert(SparkResourceUtil.getTaskSlots(conf) == 4)
+ }
+
+ test("getTaskSlots returns one slot per core when task cpus defaults to
one") {
+ val conf = new SparkConf(false).set("spark.master", "local[8]")
+ assert(SparkResourceUtil.getTaskSlots(conf) == 8)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
