fix NPE introduce on kubeflow#1280 (kubeflow#1325)

Author: ImpSy

pkg/controller/sparkapplication/controller.go

@@ -376,7 +376,13 @@ func (c *Controller) getAndUpdateExecutorState(app *v1beta2.SparkApplication) er
 			if !exists || newState != oldState {
 				if newState == v1beta2.ExecutorFailedState {
 					execContainerState := getExecutorContainerTerminatedState(pod.Status)
-					c.recordExecutorEvent(app, newState, pod.Name, execContainerState.ExitCode, execContainerState.Reason)
+					if execContainerState != nil {
+						c.recordExecutorEvent(app, newState, pod.Name, execContainerState.ExitCode, execContainerState.Reason)
+					} else {
+						// If we can't find the container state,
+						// we need to set the exitCode and the Reason to unambiguous values.
+						c.recordExecutorEvent(app, newState, pod.Name, -1, "Unknown (Container not Found)")
+					}
 				} else {
 					c.recordExecutorEvent(app, newState, pod.Name)
 				}

pkg/controller/sparkapplication/sparkapp_util.go

@@ -138,7 +138,11 @@ func getDriverContainerTerminatedState(podStatus apiv1.PodStatus) *apiv1.Contain
 }
 
 func getExecutorContainerTerminatedState(podStatus apiv1.PodStatus) *apiv1.ContainerStateTerminated {
-	return getContainerTerminatedState(config.SparkExecutorContainerName, podStatus)
+	state := getContainerTerminatedState(config.Spark3DefaultExecutorContainerName, podStatus)
+	if state == nil {
+		state = getContainerTerminatedState(config.SparkExecutorContainerName, podStatus)
+	}
+	return state
 }
 
 func getContainerTerminatedState(name string, podStatus apiv1.PodStatus) *apiv1.ContainerStateTerminated {