Merged branch k8s-executor-resources

Squashed commit of the following:

commit 253082b86748476a70415698514a8a897ca793f5
Author: Enrico Minack <github@enrico.minack.dev>
Date:   Tue Feb 17 16:46:00 2026 +0100

    Allow executor feature steps to create resources

Author: EnricoMi

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala

@@ -475,7 +475,8 @@ class ExecutorPodsAllocator(
         if (driverResources.nonEmpty && driverPod.nonEmpty) {
           addOwnerReference(driverPod.get, driverResources)
         }
-        kubernetesClient.resourceList(resources: _*).forceConflicts().serverSideApply()
+        Option(kubernetesClient.resourceList(resources: _*))
+          .map(_.forceConflicts().serverSideApply())
         resources
           .filter(_.getKind == "PersistentVolumeClaim")
           .foreach { resource =>

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala

@@ -781,6 +781,36 @@ class ExecutorPodsAllocatorSuite extends SparkFunSuite with BeforeAndAfter {
     assert(!podsAllocatorUnderTest.isDeleted("7"))
   }
 
+  test("executor feature steps can create resources") {
+    val service = new ServiceBuilder()
+      .withNewMetadata()
+      .withName("servicename")
+      .endMetadata()
+      .build()
+
+    when(executorBuilder.buildFromFeatures(any(classOf[KubernetesExecutorConf]), meq(secMgr),
+      // have the feature step define a kubernetes service (resource)
+      meq(kubernetesClient), any(classOf[ResourceProfile])))
+      .thenAnswer((invocation: InvocationOnMock) => {
+        val k8sConf: KubernetesExecutorConf = invocation.getArgument(0)
+        KubernetesExecutorSpec(
+          executorPodWithId(k8sConf.executorId.toInt, k8sConf.resourceProfileId),
+          Seq(service))
+      })
+
+    val startTime = Instant.now.toEpochMilli
+    waitForExecutorPodsClock.setTime(startTime)
+
+    // Scale up to one executor
+    podsAllocatorUnderTest.setTotalExpectedExecutors(
+      Map(defaultProfile -> 1))
+    assert(podsAllocatorUnderTest.invokePrivate(numOutstandingPods).get() == 1)
+    verify(podsWithNamespace).resource(podWithAttachedContainerForId(1))
+
+    // service is considered for creation
+    verify(kubernetesClient, times(1)).resourceList(meq(service))
+  }
+
   test("executor feature steps resources ownership") {
     val executorMetadata = mock[ObjectMeta]
     when(executorMetadata.getName).thenReturn("executor-name")