fix(sparkapplication): backfill terminationTime when missing so TTL applies

Terminal FAILED or COMPLETED apps could omit status.terminationTime, so timeToLiveSeconds never triggered deletion. Backfill from lastSubmissionAttemptTime (or now) before IsExpired and TTL requeue.

Signed-off-by: Amir Alavi <amiralavi7@gmail.com>

Author: a7i

internal/controller/sparkapplication/controller.go

@@ -675,6 +675,8 @@ func (r *Reconciler) reconcileTerminatedSparkApplication(ctx context.Context, re
 		return ctrl.Result{}, nil
 	}
 
+	util.EnsureTerminationTimeForTerminalState(app)
+
 	if util.IsExpired(app) {
 		logger.Info("Deleting expired SparkApplication", "state", app.Status.AppState.State)
 		if err := r.client.Delete(ctx, app); err != nil {

pkg/util/sparkapplication.go

@@ -60,6 +60,22 @@ func IsTerminated(app *v1beta2.SparkApplication) bool {
 		app.Status.AppState.State == v1beta2.ApplicationStateFailed
 }
 
+// EnsureTerminationTimeForTerminalState sets status.terminationTime when the app is
+// Completed or Failed but terminationTime was never recorded (e.g. submission failed
+// with FAILED set and terminationTime omitted). Without this, timeToLiveSeconds never
+// applies because expiry is computed from terminationTime.
+// It prefers lastSubmissionAttemptTime when set.
+func EnsureTerminationTimeForTerminalState(app *v1beta2.SparkApplication) {
+	if !IsTerminated(app) || !app.Status.TerminationTime.IsZero() {
+		return
+	}
+	if !app.Status.LastSubmissionAttemptTime.IsZero() {
+		app.Status.TerminationTime = app.Status.LastSubmissionAttemptTime
+	} else {
+		app.Status.TerminationTime = metav1.Now()
+	}
+}
+
 // IsExpired returns whether the given SparkApplication is expired.
 func IsExpired(app *v1beta2.SparkApplication) bool {
 	// The application has no TTL defined and will never expire.

pkg/util/sparkapplication_test.go

@@ -194,6 +194,46 @@ var _ = Describe("IsExpired", func() {
 	})
 })
 
+var _ = Describe("EnsureTerminationTimeForTerminalState", func() {
+	It("does nothing when not terminated", func() {
+		app := &v1beta2.SparkApplication{
+			Status: v1beta2.SparkApplicationStatus{
+				AppState: v1beta2.ApplicationState{State: v1beta2.ApplicationStateSubmitted},
+			},
+		}
+		util.EnsureTerminationTimeForTerminalState(app)
+		Expect(app.Status.TerminationTime.IsZero()).To(BeTrue())
+	})
+
+	It("does nothing when termination time is already set", func() {
+		t := metav1.NewTime(time.Now().Add(-time.Hour))
+		app := &v1beta2.SparkApplication{
+			Status: v1beta2.SparkApplicationStatus{
+				AppState:                    v1beta2.ApplicationState{State: v1beta2.ApplicationStateFailed},
+				TerminationTime:             t,
+				LastSubmissionAttemptTime: metav1.NewTime(time.Now()),
+			},
+		}
+		util.EnsureTerminationTimeForTerminalState(app)
+		Expect(app.Status.TerminationTime).To(Equal(t))
+	})
+
+	It("sets termination time from last submission attempt when failed and termination time is missing", func() {
+		submitAt := metav1.NewTime(time.Date(2026, 4, 26, 16, 40, 39, 0, time.UTC))
+		app := &v1beta2.SparkApplication{
+			Spec: v1beta2.SparkApplicationSpec{
+				TimeToLiveSeconds: ptr.To[int64](3600),
+			},
+			Status: v1beta2.SparkApplicationStatus{
+				AppState:                   v1beta2.ApplicationState{State: v1beta2.ApplicationStateFailed},
+				LastSubmissionAttemptTime: submitAt,
+			},
+		}
+		util.EnsureTerminationTimeForTerminalState(app)
+		Expect(app.Status.TerminationTime).To(Equal(submitAt))
+	})
+})
+
 var _ = Describe("IsDriverRunning", func() {
 	Context("SparkApplication with completed state", func() {
 		app := &v1beta2.SparkApplication{