LuciferYang commented on code in PR #58489: URL: https://github.com/apache/spark/pull/58489#discussion_r3939086190
########## resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/InformerManager.scala: ########## @@ -0,0 +1,86 @@ +/* + * 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.scheduler.cluster.k8s + +import io.fabric8.kubernetes.api.model.Pod +import io.fabric8.kubernetes.client.KubernetesClient +import io.fabric8.kubernetes.client.informers.SharedIndexInformer + +import org.apache.spark.SparkConf +import org.apache.spark.deploy.k8s.Config.KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL +import org.apache.spark.deploy.k8s.Constants.{SPARK_APP_ID_LABEL, SPARK_EXECUTOR_INACTIVE_LABEL, SPARK_POD_EXECUTOR_ROLE, SPARK_ROLE_LABEL} +import org.apache.spark.internal.Logging +import org.apache.spark.util.Utils + +/** + * Owns the shared [[SharedIndexInformer]] used by executor pod snapshot sources when the + * informer-based mode is enabled. The informer is scoped server-side to the current + * application's executor pods that are not marked inactive, matching the filter set used by + * [[ExecutorPodsWatchSnapshotSource]] and [[ExecutorPodsPollingSnapshotSource]]. + */ +class InformerManager(kubernetesClient: KubernetesClient, conf: SparkConf) + extends Logging { + + private val resyncInterval = conf.get(KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL) + // VisibleForTesting + private[k8s] var informer: SharedIndexInformer[Pod] = _ + private var stopped = false + + def initInformer(applicationId: String): Unit = { + if (informer == null) { + logInfo(s"Initializing executor pods informer for application $applicationId") + informer = kubernetesClient.pods() + .withLabel(SPARK_APP_ID_LABEL, applicationId) + .withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE) + .withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL, "true") + .runnableInformer(resyncInterval) + } + } + + def getInformer(): SharedIndexInformer[Pod] = { + if (informer == null) { + throw new IllegalStateException( + "Informer has not been initialized. Call initInformer() first.") + } + informer + } + + def startInformer(): Unit = { + if (informer == null) { + throw new IllegalStateException( + "Informer has not been initialized. Call initInformer() first.") + } + if (stopped) { + throw new IllegalStateException("Cannot run informer after stopInformer() has been called.") + } + if (!informer.isRunning) { + informer.run() Review Comment: **1. startInformer() runs the blocking, unbounded run() on the SparkContext creation thread** `InformerManager.startInformer()` calls `informer.run()` directly; in fabric8 7.x `run()` blocks until the initial LIST completes and the watch is established, and this runs on the SparkContext creation thread, whereas the legacy watch/polling sources start fully asynchronously. From the v7.8.0 Reflector source, the default exception handler declines to retry any error before the first successful sync, so an apiserver hiccup at startup (throttling, transient error) makes run() throw on the calling thread and SparkContext creation fail outright; a slow-but-progressing initial LIST (large namespace) blocks startup with no timeout and no log. A cold start lists ~0 matching pods, so manual verification wouldn't show it. Please switch to `start()` (which doesn't block the caller), but this must come with an exceptionHandler that forces retries (see my other comment) — otherwise you've only traded a startup crash for an informer that dies silently in the background. The lister poll also needs a `hasSynced()` check. ########## resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/InformerManager.scala: ########## @@ -0,0 +1,86 @@ +/* + * 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.scheduler.cluster.k8s + +import io.fabric8.kubernetes.api.model.Pod +import io.fabric8.kubernetes.client.KubernetesClient +import io.fabric8.kubernetes.client.informers.SharedIndexInformer + +import org.apache.spark.SparkConf +import org.apache.spark.deploy.k8s.Config.KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL +import org.apache.spark.deploy.k8s.Constants.{SPARK_APP_ID_LABEL, SPARK_EXECUTOR_INACTIVE_LABEL, SPARK_POD_EXECUTOR_ROLE, SPARK_ROLE_LABEL} +import org.apache.spark.internal.Logging +import org.apache.spark.util.Utils + +/** + * Owns the shared [[SharedIndexInformer]] used by executor pod snapshot sources when the + * informer-based mode is enabled. The informer is scoped server-side to the current + * application's executor pods that are not marked inactive, matching the filter set used by + * [[ExecutorPodsWatchSnapshotSource]] and [[ExecutorPodsPollingSnapshotSource]]. + */ +class InformerManager(kubernetesClient: KubernetesClient, conf: SparkConf) + extends Logging { + + private val resyncInterval = conf.get(KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL) + // VisibleForTesting + private[k8s] var informer: SharedIndexInformer[Pod] = _ + private var stopped = false + + def initInformer(applicationId: String): Unit = { + if (informer == null) { + logInfo(s"Initializing executor pods informer for application $applicationId") + informer = kubernetesClient.pods() + .withLabel(SPARK_APP_ID_LABEL, applicationId) + .withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE) + .withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL, "true") + .runnableInformer(resyncInterval) Review Comment: **2. No exceptionHandler: startup errors are not retried, and a mid-run informer death is silent** `InformerManager` sets no `exceptionHandler` and never consumes `stopped()`. In fabric8 7.x the ExceptionHandler decides retry-vs-stop, and the default chooses stop for any error before the first successful sync (it already logs internally — what's missing is the retry decision), and for non-GONE WatcherExceptions after startup too. So with the current code, a mid-run informer death is silent: events stop, the lister keeps replacing snapshots from a cache that no longer updates, and dead executors wait for RPC timeouts to be noticed. And if start becomes async per my startup comment without a handler, any startup error turns into a silent background death that the hasSynced check would then skip past forever; strictly worse than crashing. Please set `exceptionHandler((b, t) => { logError(...); true })` in `initInformer` (it can only be set before start) to log and force a retry, or consume the exceptional completion of `stopped()` and fail the driver. ########## resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsInformerSnapshotSource.scala: ########## @@ -0,0 +1,63 @@ +/* + * 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.scheduler.cluster.k8s + +import io.fabric8.kubernetes.api.model.Pod +import io.fabric8.kubernetes.client.informers.ResourceEventHandler + +import org.apache.spark.internal.Logging +import org.apache.spark.util.Utils + +/** + * Publishes executor pod updates to [[ExecutorPodsSnapshotsStore]] using the shared informer + * owned by [[InformerManager]]. Event-driven counterpart of [[ExecutorPodsListerSnapshotSource]], + * which periodically snapshots the same informer's local cache. + */ +class ExecutorPodsInformerSnapshotSource( + snapshotsStore: ExecutorPodsSnapshotsStore, + informerManager: InformerManager) + extends ExecutorPodsSnapshotSource with Logging { + + override def start(applicationId: String): Unit = { Review Comment: **8. start() has no double-start guard, and a misleading INFO fires on every normal startup** Neither new source's `start()` has the double-start guard the legacy ones have ("Cannot start the watcher twice." / "Cannot start polling more than once."); starting the informer source twice would add a second handler and duplicate events, and a second lister start would overwrite and leak the first pollingFuture. Also, since the informer source runs the informer first and the lister source then sees `isRunning=true`, the `logInfo("Informer is already running.")` in `InformerManager` fires on every normal startup and reads like something went wrong. Adding the `require` guards and demoting that log to debug would match the existing sources. ########## resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala: ########## @@ -668,6 +681,24 @@ private[spark] object Config extends Logging { .booleanConf .createWithDefault(true) + val KUBERNETES_EXECUTOR_LISTER_POLLING_INTERVAL = Review Comment: **10. The new config toggle lands ~90 lines away from its family, and this PR splits the legacy family too** `KUBERNETES_EXECUTOR_ENABLE_INFORMER` lands at line 579 (between POD_DELETION_COST and ALLOCATION_BATCH_SIZE), while its semantic siblings `ENABLE_API_POLLING`/`ENABLE_API_WATCHER` and the two new interval configs sit ~90 lines further down. The two new intervals are also inserted between the legacy toggles and `API_POLLING_INTERVAL`, so this PR splits the legacy family that used to be contiguous as well. Moving the toggle down next to the two intervals, and placing the new intervals after `API_POLLING_INTERVAL`, would keep both families together. ########## resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/InformerManager.scala: ########## @@ -0,0 +1,86 @@ +/* + * 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.scheduler.cluster.k8s + +import io.fabric8.kubernetes.api.model.Pod +import io.fabric8.kubernetes.client.KubernetesClient +import io.fabric8.kubernetes.client.informers.SharedIndexInformer + +import org.apache.spark.SparkConf +import org.apache.spark.deploy.k8s.Config.KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL +import org.apache.spark.deploy.k8s.Constants.{SPARK_APP_ID_LABEL, SPARK_EXECUTOR_INACTIVE_LABEL, SPARK_POD_EXECUTOR_ROLE, SPARK_ROLE_LABEL} +import org.apache.spark.internal.Logging +import org.apache.spark.util.Utils + +/** + * Owns the shared [[SharedIndexInformer]] used by executor pod snapshot sources when the + * informer-based mode is enabled. The informer is scoped server-side to the current + * application's executor pods that are not marked inactive, matching the filter set used by + * [[ExecutorPodsWatchSnapshotSource]] and [[ExecutorPodsPollingSnapshotSource]]. + */ +class InformerManager(kubernetesClient: KubernetesClient, conf: SparkConf) + extends Logging { + + private val resyncInterval = conf.get(KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL) + // VisibleForTesting + private[k8s] var informer: SharedIndexInformer[Pod] = _ + private var stopped = false + + def initInformer(applicationId: String): Unit = { Review Comment: **6. initInformer ignores `stopped`, and the test's intercept scope masks it** After `stopInformer()`, `stopped=true` and `informer=null`, so a later `initInformer` silently builds an informer that can never start (blocked by `stopped`) and is never closed. Production code doesn't do this today, but the suite's "Calling startInformer after stopInformer should throw" wraps both `initInformer` and `startInformer` in one `intercept`, so it passes regardless of which one throws and hides exactly this gap. Could you make `initInformer` throw when stopped (or explicitly allow revival), with the test asserting the two steps separately? ########## resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManager.scala: ########## @@ -204,6 +200,28 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit allocatorInstance } + private def makeSnapshotSources( Review Comment: **5. The mode selection in makeSnapshotSources has no test, and the method is private** `makeSnapshotSources` is `private`, while the sibling `makeExecutorPodsAllocator` in the same file is `private[k8s]` for testability, and `KubernetesClusterManagerSuite` never references it or `enableInformer`. So the core wiring of this PR (which pair of sources the flag selects) has no coverage, and the informer-mode sources are never run through `KubernetesClusterSchedulerBackend`'s start/stop; a wrong branch or a missing listerExecutor would pass every existing test. Could you make it `private[k8s]` and add a test asserting the selected source types per flag value? ########## resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala: ########## @@ -576,6 +576,19 @@ private[spark] object Config extends Logging { .intConf .createOptional + val KUBERNETES_EXECUTOR_ENABLE_INFORMER = Review Comment: **4. The three new user-facing configs are undocumented** The three new user-facing configs (`spark.kubernetes.executor.enableInformer`, `listerPollingInterval`, `informerResyncInterval`) have no entries anywhere under docs/, while the closest precedent `spark.kubernetes.executor.apiPollingInterval` is documented in the config table in docs/running-on-kubernetes.md. The mutual exclusion with the two legacy switches currently lives only in the config doc string, so users can't discover the switch or the migration notes from the docs; it's also worth noting there that the informer path requires both list and watch permissions on pods. Could you add the three entries to running-on-kubernetes.md? ########## resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsInformerSnapshotSourceSuite.scala: ########## @@ -0,0 +1,114 @@ +/* + * 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.scheduler.cluster.k8s + +import io.fabric8.kubernetes.api.model.{Pod, PodBuilder} +import io.fabric8.kubernetes.client.KubernetesClient +import io.fabric8.kubernetes.client.informers.{ResourceEventHandler, SharedIndexInformer} +import org.mockito.{ArgumentCaptor, Mock, Mockito, MockitoAnnotations} +import org.mockito.Mockito._ +import org.scalatest.BeforeAndAfterEach +import org.scalatestplus.mockito.MockitoSugar + +import org.apache.spark.{SparkConf, SparkFunSuite} +import org.apache.spark.deploy.k8s.Config.KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL +import org.apache.spark.deploy.k8s.Constants.{SPARK_APP_ID_LABEL, SPARK_EXECUTOR_INACTIVE_LABEL, SPARK_POD_EXECUTOR_ROLE, SPARK_ROLE_LABEL} +import org.apache.spark.deploy.k8s.Fabric8Aliases.{LABELED_PODS, PODS} +import org.apache.spark.scheduler.cluster.k8s.ExecutorLifecycleTestUtils.{runningExecutor, TEST_SPARK_APP_ID} + +class ExecutorPodsInformerSnapshotSourceSuite + extends SparkFunSuite + with BeforeAndAfterEach + with MockitoSugar { + + private var snapshotSource: ExecutorPodsInformerSnapshotSource = _ + private var informerManager: InformerManager = _ + + private val sparkConf = new SparkConf() + private val resyncInterval = sparkConf.get(KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL) + private val handlerCaptor: ArgumentCaptor[ResourceEventHandler[Pod]] = + ArgumentCaptor.forClass(classOf[ResourceEventHandler[Pod]]) + + @Mock + private var kubernetesClient: KubernetesClient = _ + + @Mock + private var snapshotsStore: ExecutorPodsSnapshotsStore = _ + + @Mock + private var informer: SharedIndexInformer[Pod] = _ + + @Mock + private var podOperations: PODS = _ + + @Mock + private var scopedPods: LABELED_PODS = _ + + override def beforeEach(): Unit = { + MockitoAnnotations.initMocks(this) Review Comment: **9. The three new test suites diverge from the conventions of the neighboring suites** A few conventions diverge from the neighboring suites: all three use the deprecated `MockitoAnnotations.initMocks` while every existing suite here uses `openMocks(this).close()`; the `Mockito.spy[InformerManager]` in the informer suite uses no spy feature (the lister suite just calls new); `with MockitoSugar` is unused; `handlerCaptor` is a class-level val instead of being rebuilt in beforeEach; InformerManagerSuite has a redundant `import org.mockito.Mockito.verify` (the wildcard import already provides it); and the five-line label-filter mock chain is copy-pasted verbatim in all three suites, which would suit a small shared helper. Also, "getInformer should throw if the informer has not been initialized" actually exercises the init→start→stop path rather than a fresh manager. None of this blocks — fine as a quick cleanup. ########## resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsListerSnapshotSource.scala: ########## @@ -0,0 +1,76 @@ +/* + * 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.scheduler.cluster.k8s + +import java.util.concurrent.{Future, ScheduledExecutorService, TimeUnit} + +import scala.jdk.CollectionConverters._ + +import io.fabric8.kubernetes.api.model.Pod +import io.fabric8.kubernetes.client.KubernetesClient +import io.fabric8.kubernetes.client.informers.cache.Lister + +import org.apache.spark.SparkConf +import org.apache.spark.deploy.k8s.Config.KUBERNETES_EXECUTOR_LISTER_POLLING_INTERVAL +import org.apache.spark.internal.Logging +import org.apache.spark.util.{ThreadUtils, Utils} + +/** + * Periodically snapshots the local cache of the shared [[InformerManager]] and replaces the + * contents of the [[ExecutorPodsSnapshotsStore]] with the result. Companion to + * [[ExecutorPodsInformerSnapshotSource]], which pushes updates as informer events arrive. + */ +class ExecutorPodsListerSnapshotSource( + conf: SparkConf, + kubernetesClient: KubernetesClient, + snapshotsStore: ExecutorPodsSnapshotsStore, + informerManager: InformerManager, + pollingExecutor: ScheduledExecutorService) + extends ExecutorPodsSnapshotSource with Logging { + + private val pollingInterval = conf.get(KUBERNETES_EXECUTOR_LISTER_POLLING_INTERVAL) + + private var pollingFuture: Future[_] = _ + + override def start(applicationId: String): Unit = { + informerManager.initInformer(applicationId) + informerManager.startInformer() + val lister = new Lister[Pod]( + informerManager.getInformer().getIndexer, kubernetesClient.getNamespace) + pollingFuture = pollingExecutor.scheduleWithFixedDelay( + new PollRunnable(lister), pollingInterval, pollingInterval, TimeUnit.MILLISECONDS) + } + + override def stop(): Unit = { + if (pollingFuture != null) { + pollingFuture.cancel(true) + pollingFuture = null + } + Utils.tryLogNonFatalError { + informerManager.stopInformer() + } + ThreadUtils.shutdown(pollingExecutor) + } + + private class PollRunnable(lister: Lister[Pod]) extends Runnable { + override def run(): Unit = Utils.tryLogNonFatalError { + // The informer is already scoped server-side to app-id + role=executor + non-inactive + // pods, so we can hand its snapshot to the store as-is. + snapshotsStore.replaceSnapshot(lister.list().asScala.toSeq) Review Comment: **3. The lister poll doesn't check hasSynced(), so an unsynced empty cache would wipe the snapshot store** `PollRunnable` unconditionally calls `replaceSnapshot(lister.list())`, but the informer's local cache is empty until the initial LIST finishes. `replaceSnapshot` replaces wholesale with a fresh `fullSnapshotTs`, so `ExecutorPodsLifecycleManager`'s missing-pod reconcile fires on every poll and removes every executor registered more than `missingPodDetectDelta` (30s) ago via `doRemoveExecutor`, while the allocator re-requests a full batch seeing zero known executors. Today the window is masked by the blocking `run()` (by the time backend.start() returns, the cache is synced), but it opens the moment start becomes async per my startup comment: any namespace where the initial sync is slower than `listerPollingInterval` hits it. Skipping the round when `!informer.hasSynced()` fixes it; the "Empty list of pods" test should then assert no replacement happens while unsynced. Note this check needs the force-retry exceptionHandler from my exceptionHandler comment as well — a never-synced, dead informer would otherwise mean polls skipped forever. ########## resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsInformerSnapshotSource.scala: ########## @@ -0,0 +1,63 @@ +/* + * 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.scheduler.cluster.k8s + +import io.fabric8.kubernetes.api.model.Pod +import io.fabric8.kubernetes.client.informers.ResourceEventHandler + +import org.apache.spark.internal.Logging +import org.apache.spark.util.Utils + +/** + * Publishes executor pod updates to [[ExecutorPodsSnapshotsStore]] using the shared informer + * owned by [[InformerManager]]. Event-driven counterpart of [[ExecutorPodsListerSnapshotSource]], + * which periodically snapshots the same informer's local cache. + */ +class ExecutorPodsInformerSnapshotSource( + snapshotsStore: ExecutorPodsSnapshotsStore, + informerManager: InformerManager) + extends ExecutorPodsSnapshotSource with Logging { + + override def start(applicationId: String): Unit = { + informerManager.initInformer(applicationId) + informerManager.getInformer().addEventHandler(new ExecutorPodsInformer()) + informerManager.startInformer() + } + + override def stop(): Unit = { + Utils.tryLogNonFatalError { + informerManager.stopInformer() + } + } + + private class ExecutorPodsInformer extends ResourceEventHandler[Pod] { + override def onAdd(pod: Pod): Unit = { + logDebug(s"Received add executor pod event for pod named ${pod.getMetadata.getName}") + snapshotsStore.updatePod(pod) + } + + override def onUpdate(oldPod: Pod, newPod: Pod): Unit = { Review Comment: **7. resync > 0 replays N snapshots per round (churn only; off by default)** With `informerResyncInterval > 0`, each resync replays onUpdate for every pod; the handler calls `updatePod` unconditionally and `ExecutorPodsSnapshot.withUpdate` doesn't dedup by resourceVersion, so N executors produce N snapshot objects per round and both subscribers rescan everything. It's idempotent (`fullSnapshotTs` is preserved), so this is churn rather than a correctness issue, and the default resync=0 avoids it — just noting a driver-side cost proportional to the executor count when the interval is set low. Comparing resourceVersion before `updatePod` in the handler would cap it. ########## resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotSource.scala: ########## @@ -0,0 +1,27 @@ +/* + * 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.scheduler.cluster.k8s + +/** + * Publishes snapshots of the set of executor pods that Kubernetes reports as running for an + * application. Built-in implementations are chosen by + * [[org.apache.spark.deploy.k8s.Config.KUBERNETES_EXECUTOR_ENABLE_INFORMER]]. + */ +trait ExecutorPodsSnapshotSource { Review Comment: **11. The four new types are public with no annotation, inconsistent with adjacent types** `ExecutorPodsSnapshotSource`, `InformerManager`, and the two new sources are all public with no annotation, while the types they sit next to differ: `ExecutorPodsSnapshotsStore` is `private[spark]`, and the two legacy sources carry `@Stable @DeveloperApi` on the classes and `@Since` on the methods. The new trait now sits in the public hierarchy of two @DeveloperApi classes without being @DeveloperApi itself, and `InformerManager` is pure internal wiring with no need to be public. Marking the trait @DeveloperApi and narrowing InformerManager to private[spark] (or the whole group) would avoid widening the API surface first and paying a breaking change to shrink it later. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
