[FLINK-14762][client] Implement JobClient#getJobStatus

This closes #10311 .

flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterClientJobClientAdapter.java

@@ -20,6 +20,7 @@ package org.apache.flink.client.deployment;
 
 import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
 import org.apache.flink.client.program.ClusterClient;
 import org.apache.flink.client.program.ProgramInvocationException;
 import org.apache.flink.core.execution.JobClient;
@@ -59,6 +60,11 @@ public class ClusterClientJobClientAdapter<ClusterID> implements JobClient {
 		return jobID;
 	}
 
+	@Override
+	public CompletableFuture<JobStatus> getJobStatus() {
+		return clusterClient.getJobStatus(jobID);
+	}
+
 	@Override
 	public CompletableFuture<Void> cancel() {
 		return clusterClient.cancel(jobID).thenApply(FunctionUtils.nullFn());

flink-core/src/main/java/org/apache/flink/core/execution/JobClient.java

@@ -21,6 +21,7 @@ package org.apache.flink.core.execution;
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
 
 import javax.annotation.Nullable;
 
@@ -38,6 +39,11 @@ public interface JobClient extends AutoCloseable {
 	 */
 	JobID getJobID();
 
+	/**
+	 * Requests the {@link JobStatus} of the associated job.
+	 */
+	CompletableFuture<JobStatus> getJobStatus();
+
 	/**
 	 * Cancels the associated job.
 	 */

flink-java/src/test/java/org/apache/flink/api/java/TestingJobClient.java

@@ -20,6 +20,7 @@ package org.apache.flink.api.java;
 
 import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
 import org.apache.flink.core.execution.JobClient;
 
 import javax.annotation.Nullable;
@@ -38,6 +39,11 @@ public class TestingJobClient implements JobClient {
 		return new JobID();
 	}
 
+	@Override
+	public CompletableFuture<JobStatus> getJobStatus() {
+		return CompletableFuture.completedFuture(JobStatus.FINISHED);
+	}
+
 	@Override
 	public CompletableFuture<JobExecutionResult> getJobExecutionResult(ClassLoader userClassloader) {
 		return CompletableFuture.completedFuture(new JobExecutionResult(new JobID(), 0L, Collections.emptyMap()));

flink-streaming-java/src/test/java/org/apache/flink/streaming/environment/TestingJobClient.java

@@ -20,6 +20,7 @@ package org.apache.flink.streaming.environment;
 
 import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
 import org.apache.flink.core.execution.JobClient;
 
 import javax.annotation.Nullable;
@@ -38,6 +39,11 @@ public class TestingJobClient implements JobClient {
 		return new JobID();
 	}
 
+	@Override
+	public CompletableFuture<JobStatus> getJobStatus() {
+		return CompletableFuture.completedFuture(JobStatus.FINISHED);
+	}
+
 	@Override
 	public CompletableFuture<JobExecutionResult> getJobExecutionResult(ClassLoader userClassloader) {
 		return CompletableFuture.completedFuture(new JobExecutionResult(new JobID(), 0L, Collections.emptyMap()));