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 1bd7786 [SPARK-58649] Allow SparkApplication drivers ingress to
SparkCluster workers
1bd7786 is described below
commit 1bd7786cf6e1408e792c6dd2e4c5e9d1cbeb7102
Author: yannick.goetschel <[email protected]>
AuthorDate: Fri Aug 7 13:43:35 2026 -0700
[SPARK-58649] Allow SparkApplication drivers ingress to SparkCluster workers
### What changes were proposed in this pull request?
This PR adds a second ingress peer to the worker `NetworkPolicy` generated
for every
`SparkCluster`, so that driver pods in the same namespace are admitted
alongside the cluster's
own pods.
### Why are the changes needed?
We run per-user Spark Connect sessions as SparkApplications attached to a
shared standalone
`SparkCluster`. Most jobs are fine, but every so often one dies like this:
```
Job aborted due to stage failure: Task 11 in stage 2.0 failed 4 times, most
recent failure:
Lost task 11.3 in stage 2.0 (TID 86) (<executor-ip> executor 3):
TaskResultLost (result lost from block manager)
```
What made this confusing is that the tasks themselves succeed. The History
Server lists them as
SUCCESS with TaskResultLost sitting in the Errors column, so the compute
finished and only the
result transfer failed. It also does not hit every task: in the run above,
3 of 25 tasks in the
stage died this way and the other 22 completed normally.
The driver log explains why only some of them:
```
java.io.IOException: Connecting to /<executor-ip>:<ephemeral-port> timed
out (120000 ms)
at
org.apache.spark.network.client.TransportClientFactory.createClient(...)
at
org.apache.spark.network.netty.NettyBlockTransferService$$anon$2.createAndStart(...)
at
org.apache.spark.network.shuffle.RetryingBlockTransferor.transferAllOutstanding(...)
at
org.apache.spark.storage.BlockManager.fetchRemoteManagedBuffer(...)
at org.apache.spark.scheduler.TaskResultGetter$$anon$3.run(...)
```
The driver cannot reach the executors' block manager. That only matters
once a task result goes
over `spark.task.maxDirectResultSize` (1 MB): below it the result rides
along inline in the RPC,
above it the executor parks the result in its block manager and hands the
driver an
`IndirectTaskResult` to go and fetch. The tasks that stayed under the limit
never noticed
anything. The three that went over timed out, retried three times and gave
up, and the stage
aborted on the fourth attempt.
The culprit is the worker NetworkPolicy generated in
`SparkClusterResourceSpec#buildWorkerNetworkPolicy`:
```yaml
podSelector:
matchLabels: {spark-role: worker, spark.operator/spark-cluster-name:
<cluster>}
ingress:
- from:
- podSelector:
matchLabels: {spark.operator/spark-cluster-name: <cluster>}
policyTypes: [Ingress]
```
Only pods carrying the cluster label may talk to the workers. But a
SparkApplication attached to
the cluster is a separate resource: its driver is labelled `spark-role:
driver` and
`spark.operator/spark-app-name: <app>`, never the cluster label. So the
driver is locked out of
its own executors.
None of this is visible from the outside. Worker egress is unrestricted and
no policy selects
driver pods, so the executor-to-driver direction works perfectly well —
executors register,
tasks get dispatched, tasks run. Only the return path for large results is
broken, so the
cluster looks healthy right up until a job happens to cross the 1 MB
threshold.
There is no way around it from the client side either. Executors read
`spark.task.maxDirectResultSize` from the driver SparkConf when they start,
so a Connect client
cannot raise it partway through a session.
Ports are left unrestricted on the new peer because block manager ports are
ephemeral, and
pinning `spark.blockManager.port` is not workable when several executors
share a worker pod.
### Does this PR introduce _any_ user-facing change?
Yes. The generated worker `NetworkPolicy` now also admits driver pods in
the same namespace.
Since NetworkPolicies are additive this can only widen ingress, so it
cannot deny traffic that
was previously allowed. Clusters running on a CNI that does not enforce
NetworkPolicy are
unaffected either way.
### How was this patch tested?
Extended `SparkClusterResourceSpecTest#testWorkerNetworkPolicy` to assert
both ingress peers.
```
./gradlew :spark-submission-worker:test --tests
'*SparkClusterResourceSpecTest*'
./gradlew :spark-submission-worker:spotlessCheck
```
Both pass. We also applied the equivalent policy by hand against operator
1.0.0, and the job
that reproduced this now completes.
### Was this patch authored or co-authored using generative AI tooling?
NO
Closes #787 from YGoetschel/worker-networkpolicy-allow-drivers.
Authored-by: yannick.goetschel <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../apache/spark/k8s/operator/SparkClusterResourceSpec.java | 10 ++++++++++
.../spark/k8s/operator/SparkClusterResourceSpecTest.java | 6 +++++-
2 files changed, 15 insertions(+), 1 deletion(-)
diff --git
a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterResourceSpec.java
b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterResourceSpec.java
index 70eb6c4..e7319b1 100644
---
a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterResourceSpec.java
+++
b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkClusterResourceSpec.java
@@ -462,6 +462,11 @@ public class SparkClusterResourceSpec {
/**
* Builds the NetworkPolicy for the SparkCluster.
*
+ * <p>Ingress is allowed from the cluster's own pods and from driver pods in
the same namespace.
+ * Drivers are admitted because a SparkApplication attaching to this cluster
is a separate
+ * resource and does not carry the cluster label, yet the driver must reach
the executors' block
+ * manager to fetch task results larger than {@code
spark.task.maxDirectResultSize}.
+ *
* @param clusterName The name of the SparkCluster.
* @param namespace The namespace of the SparkApplication.
* @return A NetworkPolicy object.
@@ -484,6 +489,11 @@ public class SparkClusterResourceSpec {
.addToMatchLabels(LABEL_SPARK_CLUSTER_NAME, clusterName)
.endPodSelector()
.endFrom()
+ .addNewFrom()
+ .withNewPodSelector()
+ .addToMatchLabels(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_DRIVER_VALUE)
+ .endPodSelector()
+ .endFrom()
.endIngress()
.endSpec()
.build();
diff --git
a/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkClusterResourceSpecTest.java
b/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkClusterResourceSpecTest.java
index 63c4171..50a9c55 100644
---
a/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkClusterResourceSpecTest.java
+++
b/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkClusterResourceSpecTest.java
@@ -329,8 +329,12 @@ class SparkClusterResourceSpecTest {
assertEquals(expected, policy.getSpec().getPodSelector().getMatchLabels());
assertTrue(policy.getSpec().getEgress().isEmpty());
assertEquals(1, policy.getSpec().getIngress().size());
+ var from = policy.getSpec().getIngress().get(0).getFrom();
+ assertEquals(2, from.size());
assertEquals(Map.of(LABEL_SPARK_CLUSTER_NAME, "cluster-name"),
-
policy.getSpec().getIngress().get(0).getFrom().get(0).getPodSelector().getMatchLabels());
+ from.get(0).getPodSelector().getMatchLabels());
+ assertEquals(Map.of(LABEL_SPARK_ROLE_NAME, LABEL_SPARK_ROLE_DRIVER_VALUE),
+ from.get(1).getPodSelector().getMatchLabels());
}
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
