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/20 09:14:55 UTC
[6/6] flink git commit: [FLINK-8935][tests] Implement
MiniClusterClient#listJobs
[FLINK-8935][tests] Implement MiniClusterClient#listJobs
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5fa84c28
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5fa84c28
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5fa84c28
Branch: refs/heads/master
Commit: 5fa84c28fc1bfc62fa2e1165e3407fc81b3d09a9
Parents: cfd8df4
Author: zentol <ch...@apache.org>
Authored: Wed Mar 7 11:05:12 2018 +0100
Committer: zentol <ch...@apache.org>
Committed: Tue Mar 20 10:14:26 2018 +0100
----------------------------------------------------------------------
.../flink/client/program/MiniClusterClient.java | 2 +-
.../flink/runtime/minicluster/MiniCluster.java | 16 ++++++++++++++++
2 files changed, 17 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5fa84c28/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 86e9279..961604f 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
@@ -125,7 +125,7 @@ public class MiniClusterClient extends ClusterClient<MiniClusterClient.MiniClust
@Override
public CompletableFuture<Collection<JobStatusMessage>> listJobs() throws Exception {
- throw new UnsupportedOperationException("MiniClusterClient does not yet support this operation.");
+ return guardWithSingleRetry(miniCluster::listJobs, scheduledExecutor);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/5fa84c28/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
index dfe30af..21b89ec 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
@@ -28,6 +28,7 @@ import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.blob.BlobCacheService;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.client.JobExecutionException;
+import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.clusterframework.FlinkResourceManager;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.concurrent.FutureUtils;
@@ -87,6 +88,7 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
@@ -473,6 +475,20 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
// Accessing jobs
// ------------------------------------------------------------------------
+ public CompletableFuture<Collection<JobStatusMessage>> listJobs() {
+ try {
+ return getDispatcherGateway().requestMultipleJobDetails(rpcTimeout)
+ .thenApply(jobs -> jobs.getJobs().stream()
+ .map(details -> new JobStatusMessage(details.getJobId(), details.getJobName(), details.getStatus(), details.getStartTime()))
+ .collect(Collectors.toList()));
+ } catch (LeaderRetrievalException | InterruptedException e) {
+ return FutureUtils.completedExceptionally(
+ new FlinkException(
+ "Could not retrieve job list.",
+ e));
+ }
+ }
+
public CompletableFuture<JobStatus> getJobStatus(JobID jobId) {
try {
return getDispatcherGateway().requestJobStatus(jobId, rpcTimeout);