ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

[MINOR][K8S] Print the driver pod name instead of Some(name) if absent

Browse source 
Print the driver pod name instead of Some(name) if absent

### What changes were proposed in this pull request?

### Why are the changes needed?

fix error hint

### Does this PR introduce _any_ user-facing change?

no

### How was this patch tested?

new test

Closes #32889 from yaooqinn/minork8s.

Authored-by: Kent Yao <yao@apache.org>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>
This commit is contained in:
Kent Yao 2021-06-13 09:11:14 -07:00 committed by Dongjoon Hyun
parent 6272222bc0
commit 1125afd462
2 changed files with 14 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -77,7 +77,7 @@ private[spark] class ExecutorPodsAllocator(
.withName(name) .withName(name)
.get()) .get())
.getOrElse(throw new SparkException( .getOrElse(throw new SparkException(
s"No pod was found named $kubernetesDriverPodName in the cluster in the " + s"No pod was found named $name in the cluster in the " +
s"namespace $namespace (this was supposed to be the driver pod.)."))) s"namespace $namespace (this was supposed to be the driver pod.).")))
// Executor IDs that have been requested from Kubernetes but have not been detected in any // Executor IDs that have been requested from Kubernetes but have not been detected in any

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

@ -32,7 +32,7 @@ import org.mockito.stubbing.Answer
import org.scalatest.BeforeAndAfter import org.scalatest.BeforeAndAfter
import org.scalatest.PrivateMethodTester._ import org.scalatest.PrivateMethodTester._
import org.apache.spark.{SecurityManager, SparkConf, SparkFunSuite} import org.apache.spark.{SecurityManager, SparkConf, SparkException, SparkFunSuite}
import org.apache.spark.deploy.k8s.{KubernetesExecutorConf, KubernetesExecutorSpec} import org.apache.spark.deploy.k8s.{KubernetesExecutorConf, KubernetesExecutorSpec}
import org.apache.spark.deploy.k8s.Config._ import org.apache.spark.deploy.k8s.Config._
import org.apache.spark.deploy.k8s.Constants._ import org.apache.spark.deploy.k8s.Constants._
@ -653,6 +653,18 @@ class ExecutorPodsAllocatorSuite extends SparkFunSuite with BeforeAndAfter {
verify(persistentVolumeClaims, never()).create(any()) verify(persistentVolumeClaims, never()).create(any())
} }
test("print the pod name instead of Some(name) if pod is absent") {
val nonexistentPod = "i-do-not-exist"
val conf = new SparkConf().set(KUBERNETES_DRIVER_POD_NAME, nonexistentPod)
when(kubernetesClient.pods()).thenReturn(podOperations)
when(podOperations.withName(nonexistentPod)).thenReturn(driverPodOperations)
when(driverPodOperations.get()).thenReturn(null)
val e = intercept[SparkException](new ExecutorPodsAllocator(
conf, secMgr, executorBuilder, kubernetesClient, snapshotsStore, waitForExecutorPodsClock))
assert(e.getMessage.contains("No pod was found named i-do-not-exist in the cluster in the" +
" namespace default"))
}
private def executorPodAnswer(): Answer[KubernetesExecutorSpec] = private def executorPodAnswer(): Answer[KubernetesExecutorSpec] =
(invocation: InvocationOnMock) => { (invocation: InvocationOnMock) => {
val k8sConf: KubernetesExecutorConf = invocation.getArgument(0) val k8sConf: KubernetesExecutorConf = invocation.getArgument(0)