You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2018/03/11 15:34:21 UTC
[4/6] flink git commit: [FLINK-8700] Add ClusterClient.getJobStatus()
[FLINK-8700] Add ClusterClient.getJobStatus()
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9e253b5d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9e253b5d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9e253b5d
Branch: refs/heads/release-1.5
Commit: 9e253b5d1ecaf0066d7aeae2828eee4f4671b2e8
Parents: dbb8d2f
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Mon Feb 26 11:53:47 2018 +0100
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Sun Mar 11 08:31:56 2018 -0700
----------------------------------------------------------------------
.../flink/client/program/ClusterClient.java | 29 ++++++++++++++++++++
.../flink/client/program/MiniClusterClient.java | 1 +
.../client/program/rest/RestClusterClient.java | 14 ++++++++++
3 files changed, 44 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/9e253b5d/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
index a4880db..1a783fc 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
@@ -49,6 +49,7 @@ import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils;
import org.apache.flink.runtime.instance.ActorGateway;
import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalException;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
@@ -584,6 +585,34 @@ public abstract class ClusterClient<T> {
}
/**
+ * Requests the {@link JobStatus} of the job with the given {@link JobID}.
+ */
+ public CompletableFuture<JobStatus> getJobStatus(JobID jobId) {
+ final ActorGateway jobManager;
+ try {
+ jobManager = getJobManagerGateway();
+ } catch (FlinkException e) {
+ throw new RuntimeException("Could not retrieve JobManage gateway.", e);
+ }
+
+ Future<Object> response = jobManager.ask(JobManagerMessages.getRequestJobStatus(jobId), timeout);
+
+ CompletableFuture<Object> javaFuture = FutureUtils.toJava(response);
+
+ return javaFuture.thenApply((responseMessage) -> {
+ if (responseMessage instanceof JobManagerMessages.CurrentJobStatus) {
+ return ((JobManagerMessages.CurrentJobStatus) responseMessage).status();
+ } else if (responseMessage instanceof JobManagerMessages.JobNotFound) {
+ throw new CompletionException(
+ new IllegalStateException("Could not find job with JobId " + jobId));
+ } else {
+ throw new CompletionException(
+ new IllegalStateException("Unknown JobManager response of type " + responseMessage.getClass()));
+ }
+ });
+ }
+
+ /**
* Cancels a job identified by the job id.
* @param jobId the job id
* @throws Exception In case an error occurred.
http://git-wip-us.apache.org/repos/asf/flink/blob/9e253b5d/flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java
index bbb5d49..7475071 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java
@@ -137,6 +137,7 @@ public class MiniClusterClient extends ClusterClient<MiniClusterClient.MiniClust
throw new UnsupportedOperationException("MiniClusterClient does not yet support this operation.");
}
+ @Override
public CompletableFuture<JobStatus> getJobStatus(JobID jobId) {
return guardWithSingleRetry(() -> miniCluster.getJobStatus(jobId), scheduledExecutor);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/9e253b5d/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
index 18ff099..8cf0d2c 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
@@ -35,6 +35,7 @@ import org.apache.flink.runtime.clusterframework.messages.GetClusterStatusRespon
import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.concurrent.ScheduledExecutorServiceAdapter;
import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalException;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
@@ -61,6 +62,8 @@ import org.apache.flink.runtime.rest.messages.RequestBody;
import org.apache.flink.runtime.rest.messages.ResponseBody;
import org.apache.flink.runtime.rest.messages.TerminationModeQueryParameter;
import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.job.JobDetailsHeaders;
+import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
import org.apache.flink.runtime.rest.messages.job.JobExecutionResultHeaders;
import org.apache.flink.runtime.rest.messages.job.JobSubmitHeaders;
import org.apache.flink.runtime.rest.messages.job.JobSubmitRequestBody;
@@ -254,6 +257,17 @@ public class RestClusterClient<T> extends ClusterClient<T> {
}
}
+ @Override
+ public CompletableFuture<JobStatus> getJobStatus(JobID jobId) {
+ JobDetailsHeaders detailsHeaders = JobDetailsHeaders.getInstance();
+ final JobMessageParameters params = new JobMessageParameters();
+ params.jobPathParameter.resolve(jobId);
+
+ CompletableFuture<JobDetailsInfo> responseFuture = sendRequest(detailsHeaders, params);
+
+ return responseFuture.thenApply(JobDetailsInfo::getJobStatus);
+ }
+
/**
* Requests the {@link JobResult} for the given {@link JobID}. The method retries multiple
* times to poll the {@link JobResult} before giving up.