Prevent issue handler from trying to handle issues on finished pods

internal/executor/service/pod_issue_handler.go

@@ -173,6 +173,11 @@ func (p *IssueHandler) detectPodIssues(allManagedPods []*v1.Pod) {
 		if p.hasIssue(util.ExtractJobRunId(pod)) {
 			continue
 		}
+		if util.IsInTerminalState(pod) && util.HasCurrentStateBeenReported(pod) {
+			// No need to detect issues on completed pods
+			// This prevents us sending updates on pods that are already finished and reported
+			continue
+		}
 		if pod.DeletionTimestamp != nil && pod.DeletionTimestamp.Add(p.stuckTerminatingPodExpiry).Before(p.clock.Now()) {
 			// pod is stuck in terminating phase, this sometimes happen on node failure
 			// it is safer to produce failed event than retrying as the job might have run already

internal/executor/service/pod_issue_handler_test.go

@@ -91,6 +91,10 @@ func TestPodIssueService_DeletesPodAndReportsFailed_IfStuckTerminating(t *testin
 func TestPodIssueService_DeletesPodAndReportsFailed_IfExceedsActiveDeadline(t *testing.T) {
 	startTime := time.Now().Add(-time.Minute * 10)
 
+	completedPodPastDeadline := makePodWithDeadline(startTime, 300, 0)
+	completedPodPastDeadline.Status.Phase = v1.PodFailed
+	completedPodPastDeadline.Annotations[string(v1.PodFailed)] = "true"
+
 	tests := map[string]struct {
 		expectIssueDetected bool
 		pod                 *v1.Pod
@@ -100,14 +104,18 @@ func TestPodIssueService_DeletesPodAndReportsFailed_IfExceedsActiveDeadline(t *t
 			// Created 10 mins ago, 5 min deadline
 			pod: makePodWithDeadline(startTime, 300, 0),
 		},
+		"PodPastDeadline - Completed pod": {
+			expectIssueDetected: false,
+			pod:                 completedPodPastDeadline,
+		},
 		"PodPastDeadlineWithinTerminationGracePeriod": {
 			expectIssueDetected: false,
 			// Created 10 mins ago, 5 min deadline, 10 minute grace period
 			pod: makePodWithDeadline(startTime, 300, 600),
 		},
 		"PodWithNoStartTime": {
 			expectIssueDetected: false,
-			// Created 10 mins ago, 5 min deadline, no start time
+			// No start time so cannot determine if past its deadline
 			pod: makePodWithDeadline(time.Time{}, 300, 0),
 		},
 	}