From c9b084da2efab7719c195401aa00f34d1bf23cbf Mon Sep 17 00:00:00 2001 From: Vikas Saxena Date: Tue, 20 Oct 2026 21:36:54 +1100 Subject: [PATCH 1/6] fix(sparkconnect): use server.cores as server pod CPU request Signed-off-by: Vikas Saxena --- api/v1alpha1/sparkconnect_types.go | 2 +- internal/controller/sparkconnect/options.go | 6 ------ internal/controller/sparkconnect/options_test.go | 6 +++--- internal/controller/sparkconnect/reconciler.go | 3 ++- internal/webhook/sparkconnect_validator.go | 2 +- pkg/util/container.go | 11 +++++++++++ 6 files changed, 18 insertions(+), 12 deletions(-) diff --git a/api/v1alpha1/sparkconnect_types.go b/api/v1alpha1/sparkconnect_types.go index 7bba39b782..f88088ccd1 100644 --- a/api/v1alpha1/sparkconnect_types.go +++ b/api/v1alpha1/sparkconnect_types.go @@ -111,7 +111,7 @@ type ExecutorSpec struct { // SparkPodSpec defines common things that can be customized for a Spark driver or executor pod. type SparkPodSpec struct { - // Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively. + // Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors. // +optional // +kubebuilder:validation:Minimum=1 Cores *int32 `json:"cores,omitempty"` diff --git a/internal/controller/sparkconnect/options.go b/internal/controller/sparkconnect/options.go index d06d980dcc..f8dc7199c9 100644 --- a/internal/controller/sparkconnect/options.go +++ b/internal/controller/sparkconnect/options.go @@ -196,12 +196,6 @@ func driverConfOption(conn *v1alpha1.SparkConnect) ([]string, error) { args = append(args, "--conf", fmt.Sprintf("%s=%s", property, value)) } - // Driver cores - if conn.Spec.Server.Cores != nil { - args = append(args, "--conf", - fmt.Sprintf("%s=%d", common.SparkDriverCores, *conn.Spec.Server.Cores)) - } - // Driver memory if conn.Spec.Server.Memory != nil { args = append(args, "--conf", diff --git a/internal/controller/sparkconnect/options_test.go b/internal/controller/sparkconnect/options_test.go index 1a22301809..be7df326ad 100644 --- a/internal/controller/sparkconnect/options_test.go +++ b/internal/controller/sparkconnect/options_test.go @@ -131,7 +131,7 @@ var _ = Describe("Options functions", func() { }) Context("driverConfOption and executorConfOption with CPU resources", func() { - It("does not emit driver SparkConf keys for server CoreRequest and CoreLimit", func() { + It("does not emit driver CPU SparkConf keys for server Cores, CoreRequest and CoreLimit", func() { cores := int32(4) coreRequest := resource.MustParse("3500m") coreLimit := resource.MustParse("4") @@ -157,7 +157,7 @@ var _ = Describe("Options functions", func() { Expect(err).NotTo(HaveOccurred()) config := parsedSparkConfig(args) - Expect(config).To(HaveKeyWithValue("spark.driver.cores", "4")) + Expect(config).NotTo(HaveKey(common.SparkDriverCores)) Expect(config).NotTo(HaveKey(common.SparkKubernetesDriverRequestCores)) Expect(config).NotTo(HaveKey(common.SparkKubernetesDriverLimitCores)) }) @@ -217,7 +217,7 @@ var _ = Describe("Options functions", func() { Expect(err).NotTo(HaveOccurred()) driverConfig := parsedSparkConfig(driverArgs) - Expect(driverConfig).To(HaveKeyWithValue("spark.driver.cores", "4")) + Expect(driverConfig).NotTo(HaveKey(common.SparkDriverCores)) Expect(driverConfig).NotTo(HaveKey(common.SparkKubernetesDriverRequestCores)) Expect(driverConfig).NotTo(HaveKey(common.SparkKubernetesDriverLimitCores)) }) diff --git a/internal/controller/sparkconnect/reconciler.go b/internal/controller/sparkconnect/reconciler.go index ba5d251f23..045c7c9f27 100644 --- a/internal/controller/sparkconnect/reconciler.go +++ b/internal/controller/sparkconnect/reconciler.go @@ -397,9 +397,10 @@ func (r *Reconciler) mutateServerPod(ctx context.Context, conn *v1alpha1.SparkCo ) // Setup Kubernetes CPU resources for the Connect server container. + // The server runs in client mode, where spark.driver.cores is inert, so cores sets the pod CPU request instead. container = util.SetContainerCPUResources( container, - conn.Spec.Server.CoreRequest, + util.CPURequestOrCores(conn.Spec.Server.CoreRequest, conn.Spec.Server.Cores), conn.Spec.Server.CoreLimit, ) diff --git a/internal/webhook/sparkconnect_validator.go b/internal/webhook/sparkconnect_validator.go index 9a9b93be97..228b4ae03d 100644 --- a/internal/webhook/sparkconnect_validator.go +++ b/internal/webhook/sparkconnect_validator.go @@ -431,7 +431,7 @@ func validateCPURequestLELimit(request, limit *resource.Quantity) error { } func effectiveServerCPUResources(sc *v1alpha1.SparkConnect) (request, limit *resource.Quantity) { - request = sc.Spec.Server.CoreRequest + request = util.CPURequestOrCores(sc.Spec.Server.CoreRequest, sc.Spec.Server.Cores) limit = sc.Spec.Server.CoreLimit if template := sc.Spec.Server.Template; template != nil { diff --git a/pkg/util/container.go b/pkg/util/container.go index 6c40355c95..e61257e0dc 100644 --- a/pkg/util/container.go +++ b/pkg/util/container.go @@ -63,6 +63,17 @@ func SetContainerCPUResources( return container } +// CPURequestOrCores returns coreRequest when set, otherwise cores as a whole-CPU quantity, otherwise nil. +func CPURequestOrCores(coreRequest *resource.Quantity, cores *int32) *resource.Quantity { + if coreRequest != nil { + return coreRequest + } + if cores != nil { + return resource.NewQuantity(int64(*cores), resource.DecimalSI) + } + return nil +} + // SetGPUResources sets the GPU resource identified by name to quantity in both // the requests and limits of container. Kubernetes requires extended resources // such as GPUs to have equal requests and limits. From 743bb7e1923087344c75878ceb34fd17f2c19c87 Mon Sep 17 00:00:00 2001 From: Vikas Saxena Date: Tue, 20 Oct 2026 21:36:54 +1100 Subject: [PATCH 2/6] chore: regenerate SparkConnect API artifacts Signed-off-by: Vikas Saxena --- api/openapi-spec/swagger.json | 6 +++--- .../models/spark_v1alpha1_executor_spec.py | 2 +- .../models/spark_v1alpha1_server_spec.py | 2 +- .../models/spark_v1alpha1_spark_pod_spec.py | 2 +- api/v1alpha1/zz_generated.openapi.go | 6 +++--- .../crds/sparkoperator.k8s.io_sparkconnects.yaml | 8 ++++---- config/crd/bases/sparkoperator.k8s.io_sparkconnects.yaml | 8 ++++---- docs/api-docs.md | 2 +- 8 files changed, 18 insertions(+), 18 deletions(-) diff --git a/api/openapi-spec/swagger.json b/api/openapi-spec/swagger.json index 721fa817f8..f4acbe31cb 100644 --- a/api/openapi-spec/swagger.json +++ b/api/openapi-spec/swagger.json @@ -12688,7 +12688,7 @@ ] }, "cores": { - "description": "Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively.", + "description": "Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors.", "type": "integer", "format": "int32" }, @@ -12761,7 +12761,7 @@ ] }, "cores": { - "description": "Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively.", + "description": "Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors.", "type": "integer", "format": "int32" }, @@ -13044,7 +13044,7 @@ ] }, "cores": { - "description": "Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively.", + "description": "Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors.", "type": "integer", "format": "int32" }, diff --git a/api/python_api/kubeflow_spark_api/models/spark_v1alpha1_executor_spec.py b/api/python_api/kubeflow_spark_api/models/spark_v1alpha1_executor_spec.py index 0fdfdcc08b..5f3d9b3496 100644 --- a/api/python_api/kubeflow_spark_api/models/spark_v1alpha1_executor_spec.py +++ b/api/python_api/kubeflow_spark_api/models/spark_v1alpha1_executor_spec.py @@ -45,7 +45,7 @@ class SparkV1alpha1ExecutorSpec(BaseModel): """ # noqa: E501 core_limit: Optional[IoK8sApimachineryPkgApiResourceQuantity] = Field(default=None, description="CoreLimit is the physical CPU core limit for the pod.", alias="coreLimit") core_request: Optional[IoK8sApimachineryPkgApiResourceQuantity] = Field(default=None, description="CoreRequest is the physical CPU core request for the pod.", alias="coreRequest") - cores: Optional[StrictInt] = Field(default=None, description="Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively.") + cores: Optional[StrictInt] = Field(default=None, description="Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors.") gpu: Optional[SparkV1alpha1GPUSpec] = Field(default=None, description="GPU specifies GPU resources for the pod and Spark resource scheduler. GPU discovery and per-task resource settings are configured through SparkConf.") instances: Optional[StrictInt] = Field(default=None, description="Instances is the number of executor instances.") memory: Optional[StrictStr] = Field(default=None, description="Memory is the amount of memory to request for the pod.") diff --git a/api/python_api/kubeflow_spark_api/models/spark_v1alpha1_server_spec.py b/api/python_api/kubeflow_spark_api/models/spark_v1alpha1_server_spec.py index 15b4866884..672d6fb063 100644 --- a/api/python_api/kubeflow_spark_api/models/spark_v1alpha1_server_spec.py +++ b/api/python_api/kubeflow_spark_api/models/spark_v1alpha1_server_spec.py @@ -46,7 +46,7 @@ class SparkV1alpha1ServerSpec(BaseModel): """ # noqa: E501 core_limit: Optional[IoK8sApimachineryPkgApiResourceQuantity] = Field(default=None, description="CoreLimit is the physical CPU core limit for the pod.", alias="coreLimit") core_request: Optional[IoK8sApimachineryPkgApiResourceQuantity] = Field(default=None, description="CoreRequest is the physical CPU core request for the pod.", alias="coreRequest") - cores: Optional[StrictInt] = Field(default=None, description="Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively.") + cores: Optional[StrictInt] = Field(default=None, description="Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors.") gpu: Optional[SparkV1alpha1GPUSpec] = Field(default=None, description="GPU specifies GPU resources for the pod and Spark resource scheduler. GPU discovery and per-task resource settings are configured through SparkConf.") memory: Optional[StrictStr] = Field(default=None, description="Memory is the amount of memory to request for the pod.") service: Optional[IoK8sApiCoreV1Service] = Field(default=None, description="Service exposes the Spark connect server.") diff --git a/api/python_api/kubeflow_spark_api/models/spark_v1alpha1_spark_pod_spec.py b/api/python_api/kubeflow_spark_api/models/spark_v1alpha1_spark_pod_spec.py index 7ea992d447..403359bcfb 100644 --- a/api/python_api/kubeflow_spark_api/models/spark_v1alpha1_spark_pod_spec.py +++ b/api/python_api/kubeflow_spark_api/models/spark_v1alpha1_spark_pod_spec.py @@ -45,7 +45,7 @@ class SparkV1alpha1SparkPodSpec(BaseModel): """ # noqa: E501 core_limit: Optional[IoK8sApimachineryPkgApiResourceQuantity] = Field(default=None, description="CoreLimit is the physical CPU core limit for the pod.", alias="coreLimit") core_request: Optional[IoK8sApimachineryPkgApiResourceQuantity] = Field(default=None, description="CoreRequest is the physical CPU core request for the pod.", alias="coreRequest") - cores: Optional[StrictInt] = Field(default=None, description="Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively.") + cores: Optional[StrictInt] = Field(default=None, description="Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors.") gpu: Optional[SparkV1alpha1GPUSpec] = Field(default=None, description="GPU specifies GPU resources for the pod and Spark resource scheduler. GPU discovery and per-task resource settings are configured through SparkConf.") memory: Optional[StrictStr] = Field(default=None, description="Memory is the amount of memory to request for the pod.") template: Optional[IoK8sApiCoreV1PodTemplateSpec] = Field(default=None, description="Template is a pod template that can be used to define the driver or executor pod configurations that Spark configurations do not support. Spark version >= 3.0.0 is required. Ref: https://spark.apache.org/docs/latest/running-on-kubernetes.html#pod-template.") diff --git a/api/v1alpha1/zz_generated.openapi.go b/api/v1alpha1/zz_generated.openapi.go index 2519f86bf4..c364e57fe6 100644 --- a/api/v1alpha1/zz_generated.openapi.go +++ b/api/v1alpha1/zz_generated.openapi.go @@ -510,7 +510,7 @@ func schema_spark_operator_v2_api_v1alpha1_ExecutorSpec(ref common.ReferenceCall Properties: map[string]spec.Schema{ "cores": { SchemaProps: spec.SchemaProps{ - Description: "Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively.", + Description: "Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors.", Type: []string{"integer"}, Format: "int32", }, @@ -600,7 +600,7 @@ func schema_spark_operator_v2_api_v1alpha1_ServerSpec(ref common.ReferenceCallba Properties: map[string]spec.Schema{ "cores": { SchemaProps: spec.SchemaProps{ - Description: "Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively.", + Description: "Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors.", Type: []string{"integer"}, Format: "int32", }, @@ -960,7 +960,7 @@ func schema_spark_operator_v2_api_v1alpha1_SparkPodSpec(ref common.ReferenceCall Properties: map[string]spec.Schema{ "cores": { SchemaProps: spec.SchemaProps{ - Description: "Cores maps to `spark.driver.cores` or `spark.executor.cores` for the driver and executors, respectively.", + Description: "Cores is a shorthand for coreRequest on the server, and maps to `spark.executor.cores` for executors.", Type: []string{"integer"}, Format: "int32", }, diff --git a/charts/spark-operator-chart/crds/sparkoperator.k8s.io_sparkconnects.yaml b/charts/spark-operator-chart/crds/sparkoperator.k8s.io_sparkconnects.yaml index 3fc868f08f..678c28898b 100644 --- a/charts/spark-operator-chart/crds/sparkoperator.k8s.io_sparkconnects.yaml +++ b/charts/spark-operator-chart/crds/sparkoperator.k8s.io_sparkconnects.yaml @@ -159,8 +159,8 @@ spec: pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$ x-kubernetes-int-or-string: true cores: - description: Cores maps to `spark.driver.cores` or `spark.executor.cores` - for the driver and executors, respectively. + description: Cores is a shorthand for coreRequest on the server, + and maps to `spark.executor.cores` for executors. format: int32 minimum: 1 type: integer @@ -236,8 +236,8 @@ spec: pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$ x-kubernetes-int-or-string: true cores: - description: Cores maps to `spark.driver.cores` or `spark.executor.cores` - for the driver and executors, respectively. + description: Cores is a shorthand for coreRequest on the server, + and maps to `spark.executor.cores` for executors. format: int32 minimum: 1 type: integer diff --git a/config/crd/bases/sparkoperator.k8s.io_sparkconnects.yaml b/config/crd/bases/sparkoperator.k8s.io_sparkconnects.yaml index 3fc868f08f..678c28898b 100644 --- a/config/crd/bases/sparkoperator.k8s.io_sparkconnects.yaml +++ b/config/crd/bases/sparkoperator.k8s.io_sparkconnects.yaml @@ -159,8 +159,8 @@ spec: pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$ x-kubernetes-int-or-string: true cores: - description: Cores maps to `spark.driver.cores` or `spark.executor.cores` - for the driver and executors, respectively. + description: Cores is a shorthand for coreRequest on the server, + and maps to `spark.executor.cores` for executors. format: int32 minimum: 1 type: integer @@ -236,8 +236,8 @@ spec: pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$ x-kubernetes-int-or-string: true cores: - description: Cores maps to `spark.driver.cores` or `spark.executor.cores` - for the driver and executors, respectively. + description: Cores is a shorthand for coreRequest on the server, + and maps to `spark.executor.cores` for executors. format: int32 minimum: 1 type: integer diff --git a/docs/api-docs.md b/docs/api-docs.md index cbdb53dd6a..bb763b4473 100644 --- a/docs/api-docs.md +++ b/docs/api-docs.md @@ -926,7 +926,7 @@ int32 (Optional) -

Cores maps to spark.driver.cores or spark.executor.cores for the driver and executors, respectively.

+

Cores is a shorthand for coreRequest on the server, and maps to spark.executor.cores for executors.

From 8c338c1085454ef2b9bccfebb67fcd8fc08144ad Mon Sep 17 00:00:00 2001 From: Vikas Saxena Date: Tue, 20 Oct 2026 21:38:18 +1100 Subject: [PATCH 3/6] test(sparkconnect): cover server.cores as CPU request Signed-off-by: Vikas Saxena --- .../sparkconnect/reconciler_test.go | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/internal/controller/sparkconnect/reconciler_test.go b/internal/controller/sparkconnect/reconciler_test.go index 75507e18f5..6b02630b14 100644 --- a/internal/controller/sparkconnect/reconciler_test.go +++ b/internal/controller/sparkconnect/reconciler_test.go @@ -537,7 +537,7 @@ var _ = Describe("mutateServerPod with CPU resources", func() { Expect(cpuMilliValue(container.Resources.Requests[corev1.ResourceCPU])).To(Equal(int64(2000))) }) - It("keeps spec.server.cores independent from the Kubernetes CPU resources", func() { + It("prefers server.coreRequest over server.cores for the CPU request", func() { cores := int32(4) coreRequest := resource.MustParse("500m") conn.Spec.Server.Cores = &cores @@ -555,6 +555,23 @@ var _ = Describe("mutateServerPod with CPU resources", func() { Expect(cpuMilliValue(container.Resources.Requests[corev1.ResourceCPU])).To(Equal(int64(500))) }) + It("uses server.cores as the CPU request when server.coreRequest is omitted", func() { + cores := int32(2) + conn.Spec.Server.Cores = &cores + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: conn.Namespace, + }, + } + + err := reconciler.mutateServerPod(context.TODO(), conn, pod) + Expect(err).NotTo(HaveOccurred()) + + container := pod.Spec.Containers[0] + Expect(cpuMilliValue(container.Resources.Requests[corev1.ResourceCPU])).To(Equal(int64(2000))) + Expect(container.Resources.Limits).NotTo(HaveKey(corev1.ResourceCPU)) + }) + It("does not mutate the server pod template", func() { coreRequest := resource.MustParse("500m") conn.Spec.Server.CoreRequest = &coreRequest From 21ef55842d23867d8a8278dc9d81268a41c75b47 Mon Sep 17 00:00:00 2001 From: Vikas Saxena Date: Tue, 20 Oct 2026 21:43:50 +1100 Subject: [PATCH 4/6] test: cover server.cores validation Signed-off-by: Vikas Saxena --- .../sparkconnect/reconciler_test.go | 50 +++++++++++++++++ .../webhook/sparkconnect_validator_test.go | 53 +++++++++++++++++++ pkg/util/container_test.go | 22 ++++++++ 3 files changed, 125 insertions(+) diff --git a/internal/controller/sparkconnect/reconciler_test.go b/internal/controller/sparkconnect/reconciler_test.go index 6b02630b14..ce1566f5f7 100644 --- a/internal/controller/sparkconnect/reconciler_test.go +++ b/internal/controller/sparkconnect/reconciler_test.go @@ -572,6 +572,56 @@ var _ = Describe("mutateServerPod with CPU resources", func() { Expect(container.Resources.Limits).NotTo(HaveKey(corev1.ResourceCPU)) }) + It("overrides template CPU request with server.cores when server.coreRequest is omitted", func() { + cores := int32(2) + conn.Spec.Server.Cores = &cores + conn.Spec.Server.Template = &corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{ + { + Name: common.SparkDriverContainerName, + Image: image, + Resources: corev1.ResourceRequirements{ + Requests: corev1.ResourceList{ + corev1.ResourceCPU: resource.MustParse("1"), + }, + Limits: corev1.ResourceList{ + corev1.ResourceCPU: resource.MustParse("3"), + }, + }, + }, + }, + }, + } + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: conn.Namespace, + }, + } + + err := reconciler.mutateServerPod(context.TODO(), conn, pod) + Expect(err).NotTo(HaveOccurred()) + + container := pod.Spec.Containers[0] + Expect(cpuMilliValue(container.Resources.Requests[corev1.ResourceCPU])).To(Equal(int64(2000))) + Expect(cpuMilliValue(container.Resources.Limits[corev1.ResourceCPU])).To(Equal(int64(3000))) + }) + + It("does not pass spark.driver.cores to the server", func() { + cores := int32(2) + conn.Spec.Server.Cores = &cores + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: conn.Namespace, + }, + } + + err := reconciler.mutateServerPod(context.TODO(), conn, pod) + Expect(err).NotTo(HaveOccurred()) + + Expect(pod.Spec.Containers[0].Args).NotTo(ContainElement(ContainSubstring(common.SparkDriverCores))) + }) + It("does not mutate the server pod template", func() { coreRequest := resource.MustParse("500m") conn.Spec.Server.CoreRequest = &coreRequest diff --git a/internal/webhook/sparkconnect_validator_test.go b/internal/webhook/sparkconnect_validator_test.go index 385536f072..fc0d45b2cf 100644 --- a/internal/webhook/sparkconnect_validator_test.go +++ b/internal/webhook/sparkconnect_validator_test.go @@ -704,6 +704,7 @@ func TestSparkConnectValidatorValidateCreate_ServerCoreLimitBelowTemplateRequest validator := newTestSparkConnectValidator(t) sc := newSparkConnect() + sc.Spec.Server.Cores = nil sc.Spec.Server.CoreLimit = ptr.To(resource.MustParse("1")) sc.Spec.Server.Template = &corev1.PodTemplateSpec{ Spec: corev1.PodSpec{ @@ -731,6 +732,7 @@ func TestSparkConnectValidatorValidateCreate_TemplateRequestExceedsTemplateLimit validator := newTestSparkConnectValidator(t) sc := newSparkConnect() + sc.Spec.Server.Cores = nil sc.Spec.Server.Template = &corev1.PodTemplateSpec{ Spec: corev1.PodSpec{ Containers: []corev1.Container{ @@ -783,6 +785,57 @@ func TestSparkConnectValidatorValidateCreate_ValidEffectiveCPUCombination(t *tes } } +func TestSparkConnectValidatorValidateCreate_ServerCoresExceedsLimit(t *testing.T) { + validator := newTestSparkConnectValidator(t) + + sc := newSparkConnect() + sc.Spec.Server.Cores = ptr.To[int32](2) + sc.Spec.Server.CoreLimit = ptr.To(resource.MustParse("1")) + + if _, err := validator.ValidateCreate(context.Background(), sc); err == nil || !strings.Contains(err.Error(), "coreRequest") { + t.Fatalf("expected server cores vs coreLimit validation error, got %v", err) + } +} + +func TestSparkConnectValidatorValidateCreate_ServerCoresExceedsTemplateLimit(t *testing.T) { + validator := newTestSparkConnectValidator(t) + + sc := newSparkConnect() + sc.Spec.Server.Cores = ptr.To[int32](2) + sc.Spec.Server.Template = &corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{ + { + Name: common.SparkDriverContainerName, + Image: "spark:3.5.0", + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{ + corev1.ResourceCPU: resource.MustParse("1"), + }, + }, + }, + }, + }, + } + + if _, err := validator.ValidateCreate(context.Background(), sc); err == nil || !strings.Contains(err.Error(), "coreRequest") { + t.Fatalf("expected server cores vs template coreLimit validation error, got %v", err) + } +} + +func TestSparkConnectValidatorValidateCreate_ServerCoreRequestTakesPrecedenceOverCores(t *testing.T) { + validator := newTestSparkConnectValidator(t) + + sc := newSparkConnect() + sc.Spec.Server.Cores = ptr.To[int32](4) + sc.Spec.Server.CoreRequest = ptr.To(resource.MustParse("500m")) + sc.Spec.Server.CoreLimit = ptr.To(resource.MustParse("1")) + + if _, err := validator.ValidateCreate(context.Background(), sc); err != nil { + t.Fatalf("expected success when coreRequest is within coreLimit despite larger cores, got %v", err) + } +} + func TestSparkConnectValidatorValidateCreate_ExecutorCoreRequestExceedsLimit(t *testing.T) { validator := newTestSparkConnectValidator(t) diff --git a/pkg/util/container_test.go b/pkg/util/container_test.go index b8921a1623..10cb3d578c 100644 --- a/pkg/util/container_test.go +++ b/pkg/util/container_test.go @@ -175,3 +175,25 @@ var _ = Describe("SetGPUResources", func() { Expect(container.Resources.Limits.Name(gpu, resource.DecimalSI).Value()).To(Equal(int64(1))) }) }) + +var _ = Describe("CPURequestOrCores", func() { + It("returns coreRequest when both are set", func() { + coreRequest := resource.MustParse("500m") + cores := int32(4) + + Expect(util.CPURequestOrCores(&coreRequest, &cores)).To(Equal(&coreRequest)) + }) + + It("returns cores as whole CPUs when coreRequest is nil", func() { + cores := int32(2) + + request := util.CPURequestOrCores(nil, &cores) + + Expect(request).NotTo(BeNil()) + Expect(request.MilliValue()).To(Equal(int64(2000))) + }) + + It("returns nil when both are nil", func() { + Expect(util.CPURequestOrCores(nil, nil)).To(BeNil()) + }) +}) From 1573caf0815f231b0c2a0aa84a3fb7ead4fa8d6e Mon Sep 17 00:00:00 2001 From: Vikas Saxena Date: Tue, 20 Oct 2026 22:01:28 +1100 Subject: [PATCH 5/6] docs(sparkconnect): document server.cores as coreRequest shorthand. The documentation change was generated with help from Claude. Signed-off-by: Vikas Saxena --- docs/website/user-guide/spark-connect.md | 23 ++++++++++++------- .../spark-connect-custom-resource.yaml | 3 ++- 2 files changed, 17 insertions(+), 9 deletions(-) diff --git a/docs/website/user-guide/spark-connect.md b/docs/website/user-guide/spark-connect.md index 60146d3bfd..42639c28ed 100644 --- a/docs/website/user-guide/spark-connect.md +++ b/docs/website/user-guide/spark-connect.md @@ -113,7 +113,8 @@ The main `SparkConnect` fields are: | --- | --- | | `.spec.sparkVersion` | Declares the Spark version used by the server. | | `.spec.image` | Sets a common server and executor image when templates do not provide component-specific images. | -| `.spec.server.cores` and `.spec.server.memory` | Set the server's Spark CPU and memory settings, passed as `spark.driver.cores` and `spark.driver.memory`. | +| `.spec.server.cores` | Shorthand for `.spec.server.coreRequest` in whole CPUs; `coreRequest` takes precedence when both are set. | +| `.spec.server.memory` | Sets the server's Spark memory setting, passed as `spark.driver.memory`. | | `.spec.server.coreRequest` and `.spec.server.coreLimit` | Set the physical Kubernetes CPU request and limit on the container of the operator-created server pod. | | `.spec.server.template` | Customizes the server pod, including its image, service account, volumes, and security context. If no service account is set here, the operator falls back to the `--default-service-account` controller flag (`controller.defaultServiceAccount` in the Helm chart). | | `.spec.executor.instances` | Sets the number of static executors. | @@ -172,18 +173,24 @@ for the full rules. ## CPU request and limit -`cores` and `coreRequest` / `coreLimit` are independent. `cores` is the number of -Spark task slots, while `coreRequest` and `coreLimit` are the physical Kubernetes -CPU request and limit on the container, so setting `cores: 2` together with -`coreRequest: 500m` is valid and common. +`coreRequest` and `coreLimit` are the physical Kubernetes CPU request and limit on +the container. On executors, `cores` is independent of them: it is the number of +Spark task slots, so setting `cores: 2` together with `coreRequest: 500m` is valid +and common. + +On the server, `cores` is a shorthand for `coreRequest` in whole CPUs. The server +runs as the Spark driver in client mode, where `spark.driver.cores` has no effect, +so the operator does not pass it to Spark. When both `cores` and `coreRequest` are +set, `coreRequest` wins. The two are applied differently, because the server and executor pods are created by different actors: - **Server** — the operator creates the server pod, so - `.spec.server.coreRequest` and `.spec.server.coreLimit` are written directly onto - its container's `resources`. A CPU value in `.spec.server.template` is used only - where the corresponding field is unset. + `.spec.server.coreRequest` (or `.spec.server.cores` when `coreRequest` is unset) + and `.spec.server.coreLimit` are written directly onto its container's + `resources`. A CPU value in `.spec.server.template` is used only where the + corresponding field is unset. - **Executor** — Spark creates the executor pods. The operator passes the fields to Spark as `spark.kubernetes.executor.request.cores` and `spark.kubernetes.executor.limit.cores`, and Spark sets the container resources diff --git a/examples/sparkconnect/spark-connect-custom-resource.yaml b/examples/sparkconnect/spark-connect-custom-resource.yaml index 1128dcd33e..6240db93a1 100644 --- a/examples/sparkconnect/spark-connect-custom-resource.yaml +++ b/examples/sparkconnect/spark-connect-custom-resource.yaml @@ -41,7 +41,8 @@ spec: resources: requests: # Overridden by `spec.server.coreRequest` above. Remove that field - # and this value becomes the server pod's CPU request. + # and `spec.server.cores`, and this value becomes the server pod's + # CPU request. cpu: 100m # Not overridden: the operator only manages the server pod's CPU, # so this memory request is applied to the pod as-is. From cdb58322c7ca8674aea8ac39be3ac3fad4a29475 Mon Sep 17 00:00:00 2001 From: Vikas Saxena <90456542+vikas-saxena02@users.noreply.github.com> Date: Wed, 28 Oct 2026 09:37:59 +1100 Subject: [PATCH 6/6] Update internal/controller/sparkconnect/reconciler.go Co-authored-by: Manabu McCloskey Signed-off-by: Vikas Saxena <90456542+vikas-saxena02@users.noreply.github.com> --- internal/controller/sparkconnect/reconciler.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/controller/sparkconnect/reconciler.go b/internal/controller/sparkconnect/reconciler.go index 045c7c9f27..5da49165ad 100644 --- a/internal/controller/sparkconnect/reconciler.go +++ b/internal/controller/sparkconnect/reconciler.go @@ -397,7 +397,7 @@ func (r *Reconciler) mutateServerPod(ctx context.Context, conn *v1alpha1.SparkCo ) // Setup Kubernetes CPU resources for the Connect server container. - // The server runs in client mode, where spark.driver.cores is inert, so cores sets the pod CPU request instead. + // spark.driver.cores has no effect because the Connect server runs in client mode. container = util.SetContainerCPUResources( container, util.CPURequestOrCores(conn.Spec.Server.CoreRequest, conn.Spec.Server.Cores),