You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2018/03/05 11:13:04 UTC
[2/2] 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/56fef58a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/56fef58a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/56fef58a
Branch: refs/heads/master
Commit: 56fef58a3edd167b4bf2afc7b2dd63a94b4b478f
Parents: 7d1c6b4
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Mon Feb 26 11:53:47 2018 +0100
Committer: zentol <ch...@apache.org>
Committed: Mon Mar 5 12:12:09 2018 +0100
----------------------------------------------------------------------
.../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/56fef58a/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 7817f8f..18000b7 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/56fef58a/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 dd99f0d..67e49fa 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
@@ -120,6 +120,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 miniCluster.getJobStatus(jobId);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/56fef58a/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.