about the "RetrieveSparkAppConfig" msg

[pierric]

I often find the executor pod reports that the connection to driver is not active and will shutdown in 120s. And looking into the driver pod's log, I find that driver is waiting for "RegisterExecutor" msg from the executor. That's sound very odd, driver and executor are waiting for each other!

I traced and find something does not sound right in the code, but I am not really sure. Could someone take a look?

• core\src\main\scala\org\apache\spark\executor\CoarseGrainedExecutorBackend.scala. I think it contains the main loop of the executor. Here is one line: val cfg = driver.askSync[SparkAppConfig](RetrieveSparkAppConfig(executorId)) It is unconditional executed, asking the driver for a configuration.

• \resource-managers\kubernetes\core\src\main\scala\org\apache\spark\scheduler\cluster\k8s\KubernetesClusterSchedulerBackend.scala. I think it contains the main loop of the driver pod. Here is how it handles RetrieveSparkAppConfig

  override def receiveAndReply(
    context: RpcCallContext): PartialFunction[Any, Unit] = {
    new PartialFunction[Any, Unit]() {
      override def isDefinedAt(msg: Any): Boolean = {
        msg match {
          case RetrieveSparkAppConfig(_) =>
            shuffleManager.isDefined
          case _ => false
        }
      }
  
      override def apply(msg: Any): Unit = {
        msg match {
          case RetrieveSparkAppConfig(executorId) if shuffleManager.isDefined =>
            val runningExecutorPod = RUNNING_EXECUTOR_PODS_LOCK.synchronized {
              kubernetesClient
                .pods()
                .withName(runningExecutorsToPods(executorId).getMetadata.getName)
                .get()
            }
            val shuffleSpecificProperties = shuffleManager.get
                .getShuffleServiceConfigurationForExecutor(runningExecutorPod)
            val reply = SparkAppConfig(
                sparkProperties ++ shuffleSpecificProperties,
                SparkEnv.get.securityManager.getIOEncryptionKey())
            context.reply(reply)
        }
      }
    }.orElse(super.receiveAndReply(context))
  }

  It seems that the driver pod reply the RetrieveSparkAppConfig only if shuffle service is configured.

I didn't enable the shuffle service. I guess that leads to my strange issue.

Shouldn't a configuration always be constructed and sent back?

[foxish]

I don't recall off the top of my head what the intended behavior was with RetrieveSparkAppConfig. Can you detail a case where this happens? Maybe post the spark-submit commandline you're using as well? We have tested runs that last 10+ minutes and haven't seen that issue with the applications.

[pierric]

ok, I will get some information of my k8s and the submit command on Friday. don't have access to them right now.

[liyinan926]

Reply of RetrieveSparkAppConfig is triggered through orElse(super.receiveAndReply(context)) when shuffleManager.isDefined is false. So a reply is always expected.

[pierric]

hmm, that's right. I missed that orElse part. Now I find the default handling of RetrieveSparkAppConfig. then I will need to debug my issue a little bit more.