[FLINK-8705] [flip6] Add DispatcherRestEndpoint to MiniCluster

In order to properly support the RemoteEnvironment, the Flip-6 MiniCluster
needs a REST endpoint to receive requests from the RestClusterClient.

This closes #5527.

flink-clients/src/main/java/org/apache/flink/client/LocalExecutor.java

@@ -27,7 +27,7 @@
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
-import org.apache.flink.configuration.ResourceManagerOptions;
+import org.apache.flink.configuration.RestOptions;
 import org.apache.flink.optimizer.DataStatistics;
 import org.apache.flink.optimizer.Optimizer;
 import org.apache.flink.optimizer.dag.DataSinkNode;
@@ -125,14 +125,11 @@ public void start() throws Exception {
 	private JobExecutorService createJobExecutorService(Configuration configuration) throws Exception {
 		final JobExecutorService newJobExecutorService;
 		if (CoreOptions.FLIP6_MODE.equals(configuration.getString(CoreOptions.MODE))) {
+
+			configuration.setInteger(RestOptions.REST_PORT, 0);
+
 			final MiniClusterConfiguration miniClusterConfiguration = new MiniClusterConfiguration.Builder()
 				.setConfiguration(configuration)
-				.setNumJobManagers(
-					configuration.getInteger(
-						ConfigConstants.LOCAL_NUMBER_JOB_MANAGER,
-						ConfigConstants.DEFAULT_LOCAL_NUMBER_JOB_MANAGER))
-				.setNumResourceManagers(
-					configuration.getInteger(ResourceManagerOptions.LOCAL_NUMBER_RESOURCE_MANAGER))
 				.setNumTaskManagers(
 					configuration.getInteger(
 						ConfigConstants.LOCAL_NUMBER_TASK_MANAGER,
@@ -146,6 +143,8 @@ private JobExecutorService createJobExecutorService(Configuration configuration)
 			final MiniCluster miniCluster = new MiniCluster(miniClusterConfiguration);
 			miniCluster.start();
 
+			configuration.setInteger(RestOptions.REST_PORT, miniCluster.getRestAddress().getPort());
+
 			newJobExecutorService = miniCluster;
 		} else {
 			final LocalFlinkMiniCluster localFlinkMiniCluster = new LocalFlinkMiniCluster(configuration, true);
@@ -161,7 +160,7 @@ private JobExecutorService createJobExecutorService(Configuration configuration)
 	public void stop() throws Exception {
 		synchronized (lock) {
 			if (jobExecutorService != null) {
-				jobExecutorService.terminate().get();
+				jobExecutorService.close();
 				jobExecutorService = null;
 			}
 		}

flink-clients/src/main/java/org/apache/flink/client/RemoteExecutor.java

@@ -29,6 +29,7 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
 import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.RestOptions;
 import org.apache.flink.optimizer.DataStatistics;
 import org.apache.flink.optimizer.Optimizer;
 import org.apache.flink.optimizer.costs.DefaultCostEstimator;
@@ -112,6 +113,7 @@ public RemoteExecutor(InetSocketAddress inet, Configuration clientConfiguration,
 
 		clientConfiguration.setString(JobManagerOptions.ADDRESS, inet.getHostName());
 		clientConfiguration.setInteger(JobManagerOptions.PORT, inet.getPort());
+		clientConfiguration.setInteger(RestOptions.REST_PORT, inet.getPort());
 	}
 
 	// ------------------------------------------------------------------------

flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java

@@ -19,10 +19,8 @@
 package org.apache.flink.client.program.rest;
 
 import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.JobSubmissionResult;
-import org.apache.flink.api.common.accumulators.AccumulatorHelper;
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.client.program.ClusterClient;
 import org.apache.flink.client.program.ProgramInvocationException;
@@ -78,7 +76,6 @@
 import org.apache.flink.util.ExecutorUtils;
 import org.apache.flink.util.FlinkException;
 import org.apache.flink.util.Preconditions;
-import org.apache.flink.util.SerializedThrowable;
 import org.apache.flink.util.function.CheckedSupplier;
 
 import org.apache.flink.shaded.netty4.io.netty.channel.ConnectTimeoutException;
@@ -219,20 +216,11 @@ protected JobSubmissionResult submitJob(JobGraph jobGraph, ClassLoader classLoad
 				throw new ProgramInvocationException("Could not retrieve the execution result.", ExceptionUtils.stripExecutionException(e));
 			}
 
-			if (jobResult.getSerializedThrowable().isPresent()) {
-				final SerializedThrowable serializedThrowable = jobResult.getSerializedThrowable().get();
-				final Throwable throwable = serializedThrowable.deserializeError(classLoader);
-				throw new ProgramInvocationException(throwable);
-			}
-
 			try {
-				this.lastJobExecutionResult = new JobExecutionResult(
-					jobResult.getJobId(),
-					jobResult.getNetRuntime(),
-					AccumulatorHelper.deserializeAccumulators(
-						jobResult.getAccumulatorResults(),
-						classLoader));
+				this.lastJobExecutionResult = jobResult.toJobExecutionResult(classLoader);
 				return lastJobExecutionResult;
+			} catch (JobResult.WrappedJobException we) {
+				throw new ProgramInvocationException(we.getCause());
 			} catch (IOException | ClassNotFoundException e) {
 				throw new ProgramInvocationException(e);
 			}

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobResult.java

@@ -19,16 +19,20 @@
 package org.apache.flink.runtime.jobmaster;
 
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.accumulators.AccumulatorHelper;
 import org.apache.flink.runtime.dispatcher.Dispatcher;
 import org.apache.flink.runtime.executiongraph.AccessExecutionGraph;
 import org.apache.flink.runtime.executiongraph.ErrorInfo;
 import org.apache.flink.runtime.jobgraph.JobStatus;
+import org.apache.flink.util.FlinkException;
 import org.apache.flink.util.SerializedThrowable;
 import org.apache.flink.util.SerializedValue;
 
 import javax.annotation.Nullable;
 
+import java.io.IOException;
 import java.io.Serializable;
 import java.util.Collections;
 import java.util.Map;
@@ -98,6 +102,29 @@ public Optional<SerializedThrowable> getSerializedThrowable() {
 		return Optional.ofNullable(serializedThrowable);
 	}
 
+	/**
+	 * Converts the {@link JobResult} to a {@link JobExecutionResult}.
+	 *
+	 * @param classLoader to use for deserialization
+	 * @return JobExecutionResult
+	 * @throws WrappedJobException if the JobResult contains a serialized exception
+	 * @throws IOException if the accumulator could not be deserialized
+	 * @throws ClassNotFoundException if the accumulator could not deserialized
+	 */
+	public JobExecutionResult toJobExecutionResult(ClassLoader classLoader) throws WrappedJobException, IOException, ClassNotFoundException {
+		if (serializedThrowable != null) {
+			final Throwable throwable = serializedThrowable.deserializeError(classLoader);
+			throw new WrappedJobException(throwable);
+		}
+
+		return new JobExecutionResult(
+			jobId,
+			netRuntime,
+			AccumulatorHelper.deserializeAccumulators(
+				accumulatorResults,
+				classLoader));
+	}
+
 	/**
 	 * Builder for {@link JobResult}.
 	 */
@@ -175,4 +202,16 @@ public static JobResult createFrom(AccessExecutionGraph accessExecutionGraph) {
 		return builder.build();
 	}
 
+	/**
+	 * Exception which indicates that the job has finished with an {@link Exception}.
+	 */
+	public static final class WrappedJobException extends FlinkException {
+
+		private static final long serialVersionUID = 6535061898650156019L;
+
+		public WrappedJobException(Throwable cause) {
+			super(cause);
+		}
+	}
+
 }

flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/JobExecutorService.java

@@ -18,20 +18,10 @@
 
 package org.apache.flink.runtime.minicluster;
 
-import java.util.concurrent.CompletableFuture;
+import org.apache.flink.util.AutoCloseableAsync;
 
 /**
  * Interface to control {@link JobExecutor}.
  */
-public interface JobExecutorService extends JobExecutor {
-
-	/**
-	 * Terminate the given JobExecutorService.
-	 *
-	 * <p>This method can be implemented asynchronously. Therefore it returns a future
-	 * which is completed once the termination has been done.
-	 *
-	 * @return Termination future which can also contain an exception if the termination went wrong
-	 */
-	CompletableFuture<?> terminate();
+public interface JobExecutorService extends JobExecutor, AutoCloseableAsync {
 }