This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/spark-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new d5acafd [SPARK-58835] Enable `spark.shuffle.service.enabled` by
default in `SparkCluster`
d5acafd is described below
commit d5acafd24a0cd9cf225026bbdbcb0775d73ffe8f
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Tue Aug 18 07:16:39 2026 -0700
[SPARK-58835] Enable `spark.shuffle.service.enabled` by default in
`SparkCluster`
### What changes were proposed in this pull request?
This PR aims to enable `spark.shuffle.service.enabled` by default in
`SparkCluster` resources. The default is applied before the user-provided
`sparkConf` and `confOverrides`, so any user-specified value takes precedence.
### Why are the changes needed?
In Standalone mode, dynamic allocation requires the external shuffle
service, which must be enabled at Worker launch time. Defaulting it to `true`
makes every `SparkCluster` dynamic-allocation ready out of the box, while users
can still opt out by setting `spark.shuffle.service.enabled=false` in
`spec.sparkConf`.
### Does this PR introduce _any_ user-facing change?
Yes. `SparkCluster` Workers now start the external shuffle service by
default. Setting `spark.shuffle.service.enabled=false` in `spec.sparkConf`
restores the previous behavior.
### How was this patch tested?
Pass the CIs with two new test cases in `SparkClusterSubmissionWorkerTest`
covering the default and the user-override case.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #795 from dongjoon-hyun/SPARK-58835.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../k8s/operator/SparkClusterSubmissionWorker.java | 4 ++++
.../operator/SparkClusterSubmissionWorkerTest.java | 24 ++++++++++++++++++++++
2 files changed, 28 insertions(+)
diff --git
a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterSubmissionWorker.java
b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterSubmissionWorker.java
index f72f171..29bd55b 100644
---
a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterSubmissionWorker.java
+++
b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterSubmissionWorker.java
@@ -39,6 +39,10 @@ public class SparkClusterSubmissionWorker {
String sparkVersion = (versions != null) ? versions.getSparkVersion() :
"UNKNOWN";
SparkConf effectiveSparkConf = new SparkConf();
+ // Default to enabling the external shuffle service on workers so that
clusters are
+ // dynamic-allocation ready; users can override this via sparkConf or
confOverrides.
+ effectiveSparkConf.set("spark.shuffle.service.enabled", "true");
+
cluster.getSpec().getSparkConf().forEach((key, value) -> {
if ("spark.kubernetes.container.image".equals(key)) {
value = value.replace("{{SPARK_VERSION}}", sparkVersion);
diff --git
a/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkClusterSubmissionWorkerTest.java
b/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkClusterSubmissionWorkerTest.java
index 694e02b..d70bf66 100644
---
a/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkClusterSubmissionWorkerTest.java
+++
b/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkClusterSubmissionWorkerTest.java
@@ -77,6 +77,30 @@ class SparkClusterSubmissionWorkerTest {
assertNotNull(spec.getHorizontalPodAutoscaler());
}
+ @Test
+ void enableShuffleServiceByDefault() {
+ SparkClusterSubmissionWorker worker = new SparkClusterSubmissionWorker();
+ SparkClusterResourceSpec spec = worker.getResourceSpec(cluster, Map.of());
+
assertTrue(getWorkerOpts(spec).contains("-Dspark.shuffle.service.enabled=\"true\""));
+ }
+
+ @Test
+ void respectUserProvidedShuffleServiceConf() {
+ clusterSpec.getSparkConf().put("spark.shuffle.service.enabled", "false");
+ SparkClusterSubmissionWorker worker = new SparkClusterSubmissionWorker();
+ SparkClusterResourceSpec spec = worker.getResourceSpec(cluster, Map.of());
+
assertTrue(getWorkerOpts(spec).contains("-Dspark.shuffle.service.enabled=\"false\""));
+ }
+
+ private static String getWorkerOpts(SparkClusterResourceSpec spec) {
+ return
spec.getWorkerStatefulSet().getSpec().getTemplate().getSpec().getContainers().get(0)
+ .getEnv().stream()
+ .filter(e -> "SPARK_WORKER_OPTS".equals(e.getName()))
+ .findFirst()
+ .orElseThrow()
+ .getValue();
+ }
+
@Test
void supportSparkVersionPlaceHolder() {
SparkClusterSubmissionWorker worker = new SparkClusterSubmissionWorker();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
