[
https://issues.apache.org/jira/browse/FLINK-40162?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Milind L updated FLINK-40162:
-----------------------------
Description:
On creating a FlinkSessionJob which runs quickly (1-5s for me, I suppose the
exact time depends on the cluster), the `status.jobStatus.state` is stuck at
`RECONCILING` and `status.state` is stuck at `DEPLOYED`, even though the Flink
Web UI shows the job state as "FINISHED"
Exact reproduction:
The FlinkDeployment:
```
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: sessioncluster
namespace: default
spec:
image: sleep-job:1.0 # this image has my job jar on the classpath
imagePullPolicy: Never
flinkVersion: v2_1
serviceAccount: flink
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
execution.checkpointing.interval: "2000"
jobManager:
resource:
cpu: 0.5
memory: 1024m
taskManager:
resource:
cpu: 0.5
memory: 1024m
```
The FlinkSessionJob and the code for the job:
```
apiVersion: flink.apache.org/v1beta1
kind: FlinkSessionJob
metadata:
name: sleep-job
spec:
deploymentName: sessioncluster
job:
entryClass: whatever.sleepjob.SleepJob
args:
- "1"
parallelism: 1
```
```
package whatever.sleepjob;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class SleepJob {
public static void main(String[] args) throws Exception {
if (args.length < 1)
{ System.err.println("Usage: SleepJob <seconds>");
System.exit(1); }
long sleepMs = Long.parseLong(args[0]) * 1000L;
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.fromElements("sleep")
.map(value ->
{ Thread.sleep(sleepMs); return "done";
}
)
.print();
env.execute("SleepJob");
}
}
```
To reproduce this, the execution.checkpointing.interval must be set to any
value and state.checkpoints.dir should be unset. The `status.error` field gives
more information:
```
{\"type\":\"org.apache.flink.kubernetes.operator.exception.UpgradeFailureException\",\"message\":\"Latest
checkpoint not externally addressable, Manual restore
required.\",\"additionalMetadata\":{},\"throwableList\":[]}",
```
was:
On creating a FlinkSessionJob which runs quickly (1-5s for me, I suppose the
exact time depends on the cluster), the `status.jobStatus.state` is stuck at
`RECONCILING` and `status.state` is stuck at `DEPLOYED`, even though the Flink
Web UI shows the job state as "FINISHED"
Exact reproduction:
The FlinkDeployment:
```
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: sessioncluster
namespace: default
spec:
image: sleep-job:1.0 # this img contains my job jar on the classpath
imagePullPolicy: Never
flinkVersion: v2_1
serviceAccount: flink
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
execution.checkpointing.interval: "600000"
jobManager:
resource:
cpu: 0.5
memory: 1024m
taskManager:
resource:
cpu: 0.5
memory: 1024m
```
The FlinkSessionJob and the code for the job:
```
apiVersion: flink.apache.org/v1beta1
kind: FlinkSessionJob
metadata:
name: sleep-job
spec:
deploymentName: sessioncluster
job:
entryClass: whatever.sleepjob.SleepJob
args:
- "1"
parallelism: 1
```
```
package whatever.sleepjob;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class SleepJob {
public static void main(String[] args) throws Exception {
if (args.length < 1) {
System.err.println("Usage: SleepJob <seconds>");
System.exit(1);
}
long sleepMs = Long.parseLong(args[0]) * 1000L;
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.fromElements("sleep")
.map(value -> {
Thread.sleep(sleepMs);
return "done";
})
.print();
env.execute("SleepJob");
}
}
```
To reproduce this, the execution.checkpointing.interval must be set to any
value and state.checkpoints.dir should be unset. The `status.error` field gives
more information:
```
{\"type\":\"org.apache.flink.kubernetes.operator.exception.UpgradeFailureException\",\"message\":\"Latest
checkpoint not externally addressable, Manual restore
required.\",\"additionalMetadata\":{},\"throwableList\":[]}",
```
> Fast-running FlinkSessionJob does not reflect actual state of the job
> ---------------------------------------------------------------------
>
> Key: FLINK-40162
> URL: https://issues.apache.org/jira/browse/FLINK-40162
> Project: Flink
> Issue Type: Bug
> Components: Kubernetes Operator
> Affects Versions: kubernetes-operator-1.15.0
> Reporter: Milind L
> Priority: Major
>
> On creating a FlinkSessionJob which runs quickly (1-5s for me, I suppose the
> exact time depends on the cluster), the `status.jobStatus.state` is stuck at
> `RECONCILING` and `status.state` is stuck at `DEPLOYED`, even though the
> Flink Web UI shows the job state as "FINISHED"
> Exact reproduction:
> The FlinkDeployment:
> ```
> apiVersion: flink.apache.org/v1beta1
> kind: FlinkDeployment
> metadata:
> name: sessioncluster
> namespace: default
> spec:
> image: sleep-job:1.0 # this image has my job jar on the classpath
> imagePullPolicy: Never
> flinkVersion: v2_1
> serviceAccount: flink
> flinkConfiguration:
> taskmanager.numberOfTaskSlots: "2"
> execution.checkpointing.interval: "2000"
> jobManager:
> resource:
> cpu: 0.5
> memory: 1024m
> taskManager:
> resource:
> cpu: 0.5
> memory: 1024m
> ```
> The FlinkSessionJob and the code for the job:
> ```
> apiVersion: flink.apache.org/v1beta1
> kind: FlinkSessionJob
> metadata:
> name: sleep-job
> spec:
> deploymentName: sessioncluster
> job:
> entryClass: whatever.sleepjob.SleepJob
> args:
> - "1"
> parallelism: 1
> ```
> ```
> package whatever.sleepjob;
> import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
> public class SleepJob {
> public static void main(String[] args) throws Exception {
> if (args.length < 1)
> { System.err.println("Usage: SleepJob <seconds>");
> System.exit(1); }
> long sleepMs = Long.parseLong(args[0]) * 1000L;
> StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> env.fromElements("sleep")
> .map(value ->
> { Thread.sleep(sleepMs); return "done";
> }
> )
> .print();
> env.execute("SleepJob");
> }
> }
> ```
> To reproduce this, the execution.checkpointing.interval must be set to any
> value and state.checkpoints.dir should be unset. The `status.error` field
> gives more information:
> ```
> {\"type\":\"org.apache.flink.kubernetes.operator.exception.UpgradeFailureException\",\"message\":\"Latest
> checkpoint not externally addressable, Manual restore
> required.\",\"additionalMetadata\":{},\"throwableList\":[]}",
> ```
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
