Skip to content

[SPARK-34018][K8S] NPE in ExecutorPodsSnapshot #31071

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
holdenk wants to merge 4 commits into from
Closed

[SPARK-34018][K8S] NPE in ExecutorPodsSnapshot #31071

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
45ae069
Currently the test passes because we match the null on the container …
holdenk Jan 6, 2021
949b856
Add comments on the code
holdenk Jan 6, 2021
98ed3e8
Switch to using flatMaps
holdenk Jan 7, 2021
86b09a6
Identation cleanup
holdenk Jan 7, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 12 additions & 15 deletions ...tes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshot.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ private[spark] case class ExecutorPodsSnapshot(executorPods: Map[Long, ExecutorP

object ExecutorPodsSnapshot extends Logging {
private var shouldCheckAllContainers: Boolean = _
private var sparkContainerName: String = _
private var sparkContainerName: String = DEFAULT_EXECUTOR_CONTAINER_NAME

def apply(executorPods: Seq[Pod]): ExecutorPodsSnapshot = {
ExecutorPodsSnapshot(toStatesByExecutorId(executorPods))
Expand Down Expand Up @@ -80,24 +80,21 @@ object ExecutorPodsSnapshot extends Logging {
.anyMatch(t => t != null && t.getExitCode != 0)) {
PodFailed(pod)
} else {
// Otherwise look for the Spark container
val sparkContainerStatusOpt = pod.getStatus.getContainerStatuses.asScala
.find(_.getName() == sparkContainerName)
sparkContainerStatusOpt match {
case Some(sparkContainerStatus) =>
sparkContainerStatus.getState.getTerminated match {
case t if t.getExitCode != 0 =>
PodFailed(pod)
case t if t.getExitCode == 0 =>
// Otherwise look for the Spark container and get the exit code if present.
val sparkContainerExitCode = pod.getStatus.getContainerStatuses.asScala
.find(_.getName() == sparkContainerName).flatMap(x => Option(x.getState))
.flatMap(x => Option(x.getTerminated)).flatMap(x => Option(x.getExitCode))
.map(_.toInt)
sparkContainerExitCode match {
case Some(t) =>
t match {
case 0 =>
PodSucceeded(pod)
case _ =>
Copy link
Member

@dongjoon-hyun dongjoon-hyun Jan 6, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we have three case statements before this line, t is null and t is not null and exit code is non-zero and t is not null and exist code is zero` is covered. When we fall into here?

Copy link
Contributor Author

@holdenk holdenk Jan 7, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We shouldn't but this is to prevent the spurious "non-exhaustive match" warning.

Copy link
Member

@dongjoon-hyun dongjoon-hyun Jan 7, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, got it.

PodRunning(pod)
PodFailed(pod)
}
Copy link
Member

@dongjoon-hyun dongjoon-hyun Jan 6, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What about the following?

sparkContainerStatus.getState.getTerminated match {
  case t if t != null and t.getExitCode != 0 =>
    PodFailed(pod)
  case t if t != null and t.getExitCode == 0 =>
    PodSucceeded(pod)
  case _ =>
    PodRunning(pod)
}

Copy link
Contributor Author

@holdenk holdenk Jan 7, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure that would work to. Do you prefer this one?

Copy link
Member

@dongjoon-hyun dongjoon-hyun Jan 7, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I do because it's shorter. :)

Copy link
Contributor Author

@holdenk holdenk Jan 7, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I refactored this into using flatMaps so we can just pattern match on the result, WDYT?

// If we can't find the Spark container status, fall back to the pod status. This is
// expected to occur during pod startup and other situations.
// No exit code means we are running.
case _ =>
logDebug(s"Unable to find container ${sparkContainerName} in pod ${pod} " +
"defaulting to entire pod status (running).")
PodRunning(pod)
}
}
Expand Down
3 changes: 3 additions & 0 deletions ...re/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorLifecycleTestUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,13 +115,16 @@ object ExecutorLifecycleTestUtils {
.editOrNewStatus()
.withPhase("running")
.addNewContainerStatus()
.withName(DEFAULT_EXECUTOR_CONTAINER_NAME)
.withNewState()
.withNewTerminated()
.withMessage("message")
.withExitCode(exitCode)
.endTerminated()
.endState()
.endContainerStatus()
.addNewContainerStatus()
.withName("SIDECARFRIEND")
.withNewState()
.withNewRunning()
.endRunning()
Expand Down