This is an automated email from the ASF dual-hosted git repository. gyfora pushed a commit to branch release-1.16 in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
commit 8ad105d51fd14f742ce38c7a98acddf540a4a342 Author: Royston <[email protected]> AuthorDate: Thu Aug 27 11:16:36 2026 +0530 [FLINK-38290][operator] Keep terminal job state when job is not found A resource in a globally terminal state was moved back to RECONCILING when its job was not found. After a JobManager restart the job overview comes from an empty in-memory store, so a finished job reads as missing. Globally terminal states are absorbing, so a missing job cannot invalidate the recorded state. Keep it and skip the job not found event; this also restores the terminal JobManager cleanup, which needs a terminal state. Generated-by: Claude Code --- .../operator/observer/JobStatusObserver.java | 30 +++++--- .../operator/observer/JobStatusObserverTest.java | 89 +++++++++++++++++++++- 2 files changed, 105 insertions(+), 14 deletions(-) diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java index b1936481..423ab1b4 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java @@ -301,17 +301,23 @@ public class JobStatusObserver<R extends AbstractFlinkResource<?, ?>> { protected void onTargetJobNotFound(FlinkResourceContext<R> ctx) { var resource = ctx.getResource(); var jobStatus = resource.getStatus().getJobStatus(); - - eventRecorder.triggerEvent( - resource, - EventRecorder.Type.Warning, - EventRecorder.Reason.Missing, - EventRecorder.Component.Job, - JOB_NOT_FOUND_ERR, - ctx.getKubernetesClient()); + var alreadyTerminal = ReconciliationUtils.isJobInTerminalState(resource.getStatus()); + if (alreadyTerminal) { + LOG.info( + "Job not found but it was already observed in the {} state, keeping the observed state", + jobStatus.getState()); + } else { + eventRecorder.triggerEvent( + resource, + EventRecorder.Type.Warning, + EventRecorder.Reason.Missing, + EventRecorder.Component.Job, + JOB_NOT_FOUND_ERR, + ctx.getKubernetesClient()); + } if (resource instanceof FlinkSessionJob - && !ReconciliationUtils.isJobInTerminalState(resource.getStatus()) + && !alreadyTerminal && resource.getSpec().getJob().getUpgradeMode() == UpgradeMode.STATELESS) { // We also mark jobs that were previously not terminated as suspended if // stateless upgrade mode is used. In these cases we want to simply restart the job. @@ -321,8 +327,10 @@ public class JobStatusObserver<R extends AbstractFlinkResource<?, ?>> { // upgrading state and retry the upgrade (if possible) resource.getStatus().getReconciliationStatus().setState(ReconciliationState.DEPLOYED); } - jobStatus.setState(org.apache.flink.api.common.JobStatus.RECONCILING); - resource.getStatus().setError(JOB_NOT_FOUND_ERR); + if (!alreadyTerminal) { + jobStatus.setState(org.apache.flink.api.common.JobStatus.RECONCILING); + resource.getStatus().setError(JOB_NOT_FOUND_ERR); + } } /** diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserverTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserverTest.java index e0cca019..f7733010 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserverTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserverTest.java @@ -57,6 +57,7 @@ import java.util.stream.Stream; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; /** Tests for the {@link JobStatusObserver}. */ @@ -89,9 +90,17 @@ public class JobStatusObserverTest extends OperatorTestBase { observer.observe( getResourceContext( job, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient))); - assertEquals( - JobStatusObserver.JOB_NOT_FOUND_ERR, - flinkResourceEventCollector.events.poll().getMessage()); + if (fromStatus.isGloballyTerminalState()) { + assertTrue(flinkResourceEventCollector.events.isEmpty()); + assertEquals(fromStatus, jobStatus.getState()); + assertNull(status.getError()); + } else { + assertEquals( + JobStatusObserver.JOB_NOT_FOUND_ERR, + flinkResourceEventCollector.events.poll().getMessage()); + assertEquals(JobStatus.RECONCILING, jobStatus.getState()); + assertEquals(JobStatusObserver.JOB_NOT_FOUND_ERR, status.getError()); + } assertEquals( expectedAfter, status.getReconciliationStatus() @@ -911,6 +920,80 @@ public class JobStatusObserverTest extends OperatorTestBase { ctx.getExceptionCacheEntry().getLastTimestamp()); } + @Test + void testFinishedJobStateSurvivesJobManagerRestart() throws Exception { + // A restarted JobManager serves the job overview from an empty in-memory job store, so a + // job that already finished reads as missing even though the HA store still considers it + // completed and does not resubmit it. The observed terminal state must be kept, otherwise + // the deployment is stranded in RECONCILING and the terminal JobManager is never cleaned + // up. + var deployment = initDeployment(); + var status = deployment.getStatus(); + var jobStatus = status.getJobStatus(); + jobStatus.setState(JobStatus.RUNNING); + + FlinkResourceContext<AbstractFlinkResource<?, ?>> ctx = getResourceContext(deployment); + flinkService.submitApplicationCluster( + deployment.getSpec().getJob(), ctx.getDeployConfig(deployment.getSpec()), false); + var jobId = JobID.fromHexString(jobStatus.getJobId()); + + // Let the observer record the terminal state from the cluster + flinkService.cancelJob(jobId, true); + assertTrue(observer.observe(ctx)); + assertEquals(JobStatus.FINISHED, jobStatus.getState()); + flinkResourceEventCollector.events.clear(); + + // The JobManager restart drops the finished job from the store it is served from + flinkService.clearJobsInTerminalState(); + assertFalse(observer.observe(ctx)); + + assertEquals(JobStatus.FINISHED, jobStatus.getState()); + assertNull(status.getError()); + assertTrue(flinkResourceEventCollector.events.isEmpty()); + } + + @Test + void testFailedJobStateSurvivesJobManagerRestart() throws Exception { + // Same as the finished case, but a failed job must additionally keep the recorded failure + // reason instead of having it replaced by the job not found error. + var deployment = initDeployment(); + var status = deployment.getStatus(); + var jobStatus = status.getJobStatus(); + jobStatus.setState(JobStatus.RUNNING); + + FlinkResourceContext<AbstractFlinkResource<?, ?>> ctx = getResourceContext(deployment); + flinkService.submitApplicationCluster( + deployment.getSpec().getJob(), ctx.getDeployConfig(deployment.getSpec()), false); + var jobId = JobID.fromHexString(jobStatus.getJobId()); + + flinkService.markApplicationJobFailedWithError(jobId, "job failure"); + assertTrue(observer.observe(ctx)); + assertEquals(JobStatus.FAILED, jobStatus.getState()); + var failureError = status.getError(); + flinkResourceEventCollector.events.clear(); + + flinkService.clearJobsInTerminalState(); + assertFalse(observer.observe(ctx)); + + assertEquals(JobStatus.FAILED, jobStatus.getState()); + assertEquals(failureError, status.getError()); + assertTrue(flinkResourceEventCollector.events.isEmpty()); + } + + @Test + void testMissingTerminalJobStillUnblocksPendingUpgrade() throws Exception { + // Keeping the terminal job state must not keep a pending upgrade in the UPGRADING state, + // the reconciliation state is still reset so the upgrade can be retried. + var deployment = initDeployment(); + var status = deployment.getStatus(); + status.getJobStatus().setState(JobStatus.FINISHED); + status.getReconciliationStatus().setState(ReconciliationState.UPGRADING); + + observer.observe(getResourceContext(deployment)); + + assertEquals(ReconciliationState.DEPLOYED, status.getReconciliationStatus().getState()); + } + private static Stream<Arguments> cancellingArgs() { var args = new ArrayList<Arguments>(); for (var status : JobStatus.values()) {
