You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2022/03/24 10:41:37 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-26832] Output more status info for JobObserver
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new a8551e7 [FLINK-26832] Output more status info for JobObserver
a8551e7 is described below
commit a8551e703ce49acbfc7443dbbc06e3a9304e67b0
Author: Biao Geng <gb...@gmail.com>
AuthorDate: Thu Mar 24 14:51:01 2022 +0800
[FLINK-26832] Output more status info for JobObserver
---
.../kubernetes/operator/observer/BaseObserver.java | 3 ++-
.../kubernetes/operator/observer/JobObserver.java | 23 ++++++++++++++++------
2 files changed, 19 insertions(+), 7 deletions(-)
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/BaseObserver.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/BaseObserver.java
index 85d47fd..8d4ee77 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/BaseObserver.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/BaseObserver.java
@@ -63,7 +63,8 @@ public abstract class BaseObserver implements Observer {
JobManagerDeploymentStatus previousJmStatus =
deploymentStatus.getJobManagerDeploymentStatus();
- logger.info("Observing JobManager deployment status");
+ logger.info(
+ "Observing JobManager deployment. Previous status: {}", previousJmStatus.name());
if (JobManagerDeploymentStatus.DEPLOYED_NOT_READY == previousJmStatus) {
deploymentStatus.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobObserver.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobObserver.java
index 0802dfe..163a9bb 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobObserver.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobObserver.java
@@ -63,7 +63,7 @@ public class JobObserver extends BaseObserver {
FlinkDeployment flinkApp, Context context, Configuration effectiveConfig) {
logger.info("Observing job status");
FlinkDeploymentStatus flinkAppStatus = flinkApp.getStatus();
-
+ String previousJobStatus = flinkAppStatus.getJobStatus().getState();
Collection<JobStatusMessage> clusterJobStatuses;
try {
clusterJobStatuses = flinkService.listJobs(effectiveConfig);
@@ -81,14 +81,24 @@ public class JobObserver extends BaseObserver {
flinkAppStatus.getJobStatus().setState(JOB_STATE_UNKNOWN);
return false;
}
-
- updateJobStatus(flinkAppStatus.getJobStatus(), new ArrayList<>(clusterJobStatuses));
- logger.info("Job status successfully updated");
+ String targetJobStatus =
+ updateJobStatus(flinkAppStatus.getJobStatus(), new ArrayList<>(clusterJobStatuses));
+ if (targetJobStatus.equals(previousJobStatus)) {
+ logger.info("Job status ({}) unchanged", previousJobStatus);
+ } else {
+ logger.info(
+ "Job status successfully updated from {} to {}",
+ previousJobStatus,
+ targetJobStatus);
+ }
return true;
}
- /** Update previous job status based on the job list from the cluster. */
- private void updateJobStatus(JobStatus status, List<JobStatusMessage> clusterJobStatuses) {
+ /**
+ * Update previous job status based on the job list from the cluster and return the target
+ * status.
+ */
+ private String updateJobStatus(JobStatus status, List<JobStatusMessage> clusterJobStatuses) {
Collections.sort(
clusterJobStatuses, (j1, j2) -> Long.compare(j2.getStartTime(), j1.getStartTime()));
JobStatusMessage newJob = clusterJobStatuses.get(0);
@@ -98,6 +108,7 @@ public class JobObserver extends BaseObserver {
status.setJobId(newJob.getJobId().toHexString());
status.setStartTime(String.valueOf(newJob.getStartTime()));
status.setUpdateTime(String.valueOf(System.currentTimeMillis()));
+ return status.getState();
}
private void observeSavepointStatus(FlinkDeployment flinkApp, Configuration effectiveConfig) {