Milind L created FLINK-40164:
--------------------------------

             Summary: FlinkSessionJob's lifecycleState stays "UPGRADING" even 
after it's suspended successfully
                 Key: FLINK-40164
                 URL: https://issues.apache.org/jira/browse/FLINK-40164
             Project: Flink
          Issue Type: Bug
          Components: Kubernetes Operator
    Affects Versions: 1.15.0
            Reporter: Milind L


While running a FlinkSessionJob, if I change spec.job.state = suspended, the 
job is successfully suspended, but the status.lifecycleState says "UPGRADING" 
indefinitely. The Flink Web UI shows the job as canceled. 

 

Files to reproduce this issue:

The FlinkDeployment:
{code:java}
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: sessioncluster
  namespace: default
spec:
  image: idle-job:1.0
  imagePullPolicy: Never
  flinkVersion: v2_1
  serviceAccount: flink
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
  jobManager:
    resource:
      cpu: 0.5
      memory: 1024m
  taskManager:
    resource:
      cpu: 0.5
      memory: 1024m
{code}
 

 

The FlinkSessionJob and the code
{code:yaml}
apiVersion: flink.apache.org/v1beta1
kind: FlinkSessionJob
metadata:   name: idle-job
spec:   
  deploymentName: sessioncluster
  job:     
    entryClass: myclasspath.IdleJob
    parallelism: 1
    upgradeMode: stateless
    state: running # later, suspended
{code}
{code:java}
package myclasspath;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.legacy.SourceFunction;

public class IdleJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new IdleSource()).print();
        env.execute("IdleJob");
    }

    public static class IdleSource implements SourceFunction<String> {
        private volatile boolean running = true;

        @Override
        public void run(SourceContext<String> ctx) throws Exception {
            while (running) {
                Thread.sleep(10000);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}

{code}
The sequence of events:
1. Create the session cluster and the session job, wait for it to start running.
2. Change the state of the job to 'suspended', apply, and wait. It will show as 
canceled in the web UI, but kubectl get flinksessionjobs will not move from 
"UPGRADING". Here's what it looks like for me:
{code:java}


{
  "apiVersion": "flink.apache.org/v1beta1",
  "kind": "FlinkSessionJob",
  "metadata": {
    "annotations": {
      "kubectl.kubernetes.io/last-applied-configuration": 
"{\"apiVersion\":\"flink.apache.org/v1beta1\",\"kind\":\"FlinkSessionJob\",\"metadata\":{\"annotations\":{},\"name\":\"idle-job\",\"namespace\":\"default\"},\"spec\":{\"deploymentName\":\"sessioncluster\",\"job\":{\"entryClass\":\"myclasspath.IdleJob\",\"parallelism\":1,\"state\":\"suspended\",\"upgradeMode\":\"stateless\"}}}\n"
    },
    "creationTimestamp": "2026-07-17T12:11:05Z",
    "finalizers": [
      "flinksessionjobs.flink.apache.org/finalizer"
    ],
    "generation": 2,
    "labels": {
      "target.session": "sessioncluster"
    },
    "name": "idle-job",
    "namespace": "default",
    "resourceVersion": "26301",
    "uid": "aa1fdb02-3c70-486c-adf8-e0f557594b3c"
  },
  "spec": {
    "deploymentName": "sessioncluster",
    "flinkConfiguration": {},
    "job": {
      "args": [],
      "entryClass": "myclasspath.IdleJob",
      "parallelism": 1,
      "state": "suspended",
      "upgradeMode": "stateless"
    }
  },
  "status": {
    "jobStatus": {
      "checkpointInfo": {
        "lastPeriodicCheckpointTimestamp": 0
      },
      "jobId": "a610d66b1a3ec486b638ff617a02e819",
      "jobName": "IdleJob",
      "savepointInfo": {
        "lastPeriodicSavepointTimestamp": 0,
        "savepointHistory": []
      },
      "startTime": "1784290267673",
      "state": "CANCELED",
      "updateTime": "1784290339023"
    },
    "lifecycleState": "UPGRADING",
    "observedGeneration": 1,
    "reconciliationStatus": {
      "lastReconciledSpec": 
"{\"spec\":{\"job\":{\"jarURI\":null,\"parallelism\":1,\"entryClass\":\"myclasspath.IdleJob\",\"args\":[],\"state\":\"suspended\",\"savepointTriggerNonce\":null,\"initialSavepointPath\":null,\"checkpointTriggerNonce\":null,\"upgradeMode\":\"stateless\",\"allowNonRestoredState\":null,\"savepointRedeployNonce\":null,\"autoscalerResetNonce\":null},\"restartNonce\":null,\"flinkConfiguration\":{},\"deploymentName\":\"sessioncluster\"},\"resource_metadata\":{\"apiVersion\":\"flink.apache.org/v1beta1\",\"firstDeployment\":false}}",
      "lastStableSpec": 
"{\"spec\":{\"job\":{\"jarURI\":null,\"parallelism\":1,\"entryClass\":\"myclasspath.IdleJob\",\"args\":[],\"state\":\"running\",\"savepointTriggerNonce\":null,\"initialSavepointPath\":null,\"checkpointTriggerNonce\":null,\"upgradeMode\":\"stateless\",\"allowNonRestoredState\":null,\"savepointRedeployNonce\":null,\"autoscalerResetNonce\":null},\"restartNonce\":null,\"flinkConfiguration\":{},\"deploymentName\":\"sessioncluster\"},\"resource_metadata\":{\"apiVersion\":\"flink.apache.org/v1beta1\",\"firstDeployment\":true}}",
      "reconciliationTimestamp": 1784290269845,
      "state": "UPGRADING"
    }
  }
}
{code}



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

Reply via email to