@Override
  public DeleteControl cleanup(
      SparkApplication sparkApplication, Context<SparkApplication> context) {
    LoggingUtils.TrackedMDC trackedMDC = new LoggingUtils.TrackedMDC();
    try {
      trackedMDC.set(sparkApplication);
      log.info("Cleaning up resources for SparkApp.");
      SparkAppContext ctx = new SparkAppContext(sparkApplication, context, submissionWorker);
      List<AppReconcileStep> cleanupSteps = new ArrayList<>();
      cleanupSteps.add(new AppValidateStep());
      cleanupSteps.add(new AppTerminatedStep());
      cleanupSteps.add(new AppCleanUpStep(SparkAppStatusUtils::appCancelled));
      for (AppReconcileStep step : cleanupSteps) {
        ReconcileProgress progress = step.reconcile(ctx, sparkAppStatusRecorder);
        if (progress.isCompleted()) {
          if (progress.isRequeue()) {
            return DeleteControl.noFinalizerRemoval()
                .rescheduleAfter(progress.getRequeueAfterDuration());
          } else {
            break;
          }
        }
      }
    } finally {
      log.info("Cleanup completed");
      trackedMDC.reset();
    }
    sparkAppStatusRecorder.removeCachedStatus(sparkApplication);
    return DeleteControl.defaultDelete();
  }
@Test
  void testCleanupRunningApp() {
    try (MockedConstruction<SparkAppContext> mockAppContext =
            mockConstruction(
                SparkAppContext.class,
                (mock, context) -> {
                  when(mock.getResource()).thenReturn(app);
                  when(mock.getClient()).thenReturn(mockClient);
                  when(mock.getDriverPod()).thenReturn(Optional.of(mockDriver));
                  when(mock.getDriverPodSpec()).thenReturn(mockDriver);
                  when(mock.getDriverPreResourcesSpec()).thenReturn(Collections.emptyList());
                  when(mock.getDriverResourcesSpec()).thenReturn(Collections.emptyList());
                });
        MockedStatic<ReconcilerUtils> utils = Mockito.mockStatic(ReconcilerUtils.class)) {
      // delete running app
      app.setStatus(
          app.getStatus()
              .appendNewState(new ApplicationState(ApplicationStateSummary.RunningHealthy, "")));
      DeleteControl deleteControl = reconciler.cleanup(app, mockContext);
      assertFalse(deleteControl.isRemoveFinalizer());
      utils.verify(() -> ReconcilerUtils.deleteResourceIfExists(mockClient, mockDriver, false));
      assertEquals(
          ApplicationStateSummary.ResourceReleased,
          app.getStatus().getCurrentState().getCurrentStateSummary());

      // proceed delete for terminated app
      deleteControl = reconciler.cleanup(app, mockContext);
      assertTrue(deleteControl.isRemoveFinalizer());
    }
  }