[FLINK-2373] Add configuration parameter to createRemoteEnvironment m…

…ethod

This closes apache#1066

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

@@ -54,30 +54,47 @@ public class RemoteExecutor extends PlanExecutor {
  private static final Logger LOG = LoggerFactory.getLogger(RemoteExecutor.class);
 
  private final List<String> jarFiles;
- private final Configuration configuration;
+
+ private final Configuration clientConfiguration;
 
  public RemoteExecutor(String hostname, int port) {
- this(hostname, port, Collections.<String>emptyList());
+ this(hostname, port, Collections.<String>emptyList(), new Configuration());
  }
 
  public RemoteExecutor(String hostname, int port, String jarFile) {
- this(hostname, port, Collections.singletonList(jarFile));
+ this(hostname, port, Collections.singletonList(jarFile), new Configuration());
  }
 
  public RemoteExecutor(String hostport, String jarFile) {
- this(getInetFromHostport(hostport), Collections.singletonList(jarFile));
+ this(getInetFromHostport(hostport), Collections.singletonList(jarFile), new Configuration());
  }
 
  public RemoteExecutor(String hostname, int port, List<String> jarFiles) {
- this(new InetSocketAddress(hostname, port), jarFiles);
+ this(new InetSocketAddress(hostname, port), jarFiles, new Configuration());
+ }
+
+ public RemoteExecutor(String hostname, int port, Configuration clientConfiguration) {
+ this(hostname, port, Collections.<String>emptyList(), clientConfiguration);
+ }
+
+ public RemoteExecutor(String hostname, int port, String jarFile, Configuration clientConfiguration) {
+ this(hostname, port, Collections.singletonList(jarFile), clientConfiguration);
+ }
+
+ public RemoteExecutor(String hostport, String jarFile, Configuration clientConfiguration) {
+ this(getInetFromHostport(hostport), Collections.singletonList(jarFile), clientConfiguration);
+ }
+
+ public RemoteExecutor(String hostname, int port, List<String> jarFiles, Configuration clientConfiguration) {
+ this(new InetSocketAddress(hostname, port), jarFiles, clientConfiguration);
  }
 
- public RemoteExecutor(InetSocketAddress inet, List<String> jarFiles) {
+ public RemoteExecutor(InetSocketAddress inet, List<String> jarFiles, Configuration clientConfiguration) {
  this.jarFiles = jarFiles;
- configuration = new Configuration();
+ this.clientConfiguration = clientConfiguration;
 
- configuration.setString(ConfigConstants.JOB_MANAGER_IPC_ADDRESS_KEY, inet.getHostName());
- configuration.setInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY, inet.getPort());
+ clientConfiguration.setString(ConfigConstants.JOB_MANAGER_IPC_ADDRESS_KEY, inet.getHostName());
+ clientConfiguration.setInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY, inet.getPort());
  }
 
  @Override
@@ -87,7 +104,7 @@ public JobExecutionResult executePlan(Plan plan) throws Exception {
  }
 
  public JobExecutionResult executePlanWithJars(JobWithJars p) throws Exception {
- Client c = new Client(configuration, p.getUserCodeClassLoader(), -1);
+ Client c = new Client(clientConfiguration, p.getUserCodeClassLoader(), -1);
  c.setPrintStatusDuringExecution(isPrintingStatusDuringExecution());
 
  JobSubmissionResult result = c.run(p, -1, true);
@@ -103,7 +120,7 @@ public JobExecutionResult executeJar(String jarPath, String assemblerClass, Stri
  File jarFile = new File(jarPath);
  PackagedProgram program = new PackagedProgram(jarFile, assemblerClass, args);
 
- Client c = new Client(configuration, program.getUserCodeClassLoader(), -1);
+ Client c = new Client(clientConfiguration, program.getUserCodeClassLoader(), -1);
  c.setPrintStatusDuringExecution(isPrintingStatusDuringExecution());
 
  JobSubmissionResult result = c.run(program.getPlanWithJars(), -1, true);
@@ -118,7 +135,7 @@ public JobExecutionResult executeJar(String jarPath, String assemblerClass, Stri
  @Override
  public String getOptimizerPlanAsJSON(Plan plan) throws Exception {
  JobWithJars p = new JobWithJars(plan, this.jarFiles);
- Client c = new Client(configuration, p.getUserCodeClassLoader(), -1);
+ Client c = new Client(clientConfiguration, p.getUserCodeClassLoader(), -1);
 
  OptimizedPlan op = (OptimizedPlan) c.getOptimizedPlan(p, -1);
  PlanJSONDumpGenerator jsonGen = new PlanJSONDumpGenerator();

flink-clients/src/test/java/org/apache/flink/client/RemoteExecutorHostnameResolutionTest.java

@@ -21,6 +21,7 @@
 import org.apache.flink.api.common.Plan;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.io.DiscardingOutputFormat;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.client.program.ProgramInvocationException;
 import org.junit.Test;
 
@@ -67,7 +68,7 @@ public void testUnresolvableHostname2() {
 
  try {
  InetSocketAddress add = new InetSocketAddress(nonExistingHostname, port);
- RemoteExecutor exec = new RemoteExecutor(add, Collections.<String>emptyList());
+ RemoteExecutor exec = new RemoteExecutor(add, Collections.<String>emptyList(), new Configuration());
  exec.executePlan(getProgram());
  fail("This should fail with an ProgramInvocationException");
  }

flink-core/src/main/java/org/apache/flink/api/common/PlanExecutor.java

@@ -105,11 +105,12 @@ public static PlanExecutor createLocalExecutor(Configuration configuration) {
  * 
  * @param hostname The address of the JobManager to send the program to.
  * @param port The port of the JobManager to send the program to.
+ * @param clientConfiguration The configuration for the client (Akka, default.parallelism).
  * @param jarFiles A list of jar files that contain the user-defined function (UDF) classes and all classes used
  * from within the UDFs.
  * @return A remote executor.
  */
- public static PlanExecutor createRemoteExecutor(String hostname, int port, String... jarFiles) {
+ public static PlanExecutor createRemoteExecutor(String hostname, int port, Configuration clientConfiguration, String... jarFiles) {
  if (hostname == null) {
  throw new IllegalArgumentException("The hostname must not be null.");
  }
@@ -123,7 +124,10 @@ public static PlanExecutor createRemoteExecutor(String hostname, int port, Strin
  : Arrays.asList(jarFiles); 
 
  try {
- return reClass.getConstructor(String.class, int.class, List.class).newInstance(hostname, port, files);
+ PlanExecutor executor = (clientConfiguration == null) ?
+ reClass.getConstructor(String.class, int.class, List.class).newInstance(hostname, port, files) :
+ reClass.getConstructor(String.class, int.class, List.class, Configuration.class).newInstance(hostname, port, files, clientConfiguration);
+ return executor;
  }
  catch (Throwable t) {
  throw new RuntimeException("An error occurred while loading the remote executor ("

flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java

@@ -1105,6 +1105,28 @@ public static ExecutionEnvironment createRemoteEnvironment(String host, int port
  return new RemoteEnvironment(host, port, jarFiles);
  }
 
+ /**
+ * Creates a {@link RemoteEnvironment}. The remote environment sends (parts of) the program
+ * to a cluster for execution. Note that all file paths used in the program must be accessible from the
+ * cluster. The custom configuration file is used to configure Akka specific configuration parameters
+ * for the Client only; Program parallelism can be set via {@link ExecutionEnvironment#setParallelism(int)}.
+ *
+ * Cluster configuration has to be done in the remotely running Flink instance.
+ *
+ * @param host The host name or address of the master (JobManager), where the program should be executed.
+ * @param port The port of the master (JobManager), where the program should be executed.
+ * @param clientConfiguration Pass a custom configuration to the Client.
+ * @param jarFiles The JAR files with code that needs to be shipped to the cluster. If the program uses
+ * user-defined functions, user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public static ExecutionEnvironment createRemoteEnvironment(String host, int port, Configuration clientConfiguration, String... jarFiles) {
+ RemoteEnvironment rec = new RemoteEnvironment(host, port, jarFiles);
+ rec.setClientConfiguration(clientConfiguration);
+ return rec;
+ }
+
  /**
  * Creates a {@link RemoteEnvironment}. The remote environment sends (parts of) the program 
  * to a cluster for execution. Note that all file paths used in the program must be accessible from the

flink-java/src/main/java/org/apache/flink/api/java/RemoteEnvironment.java

@@ -21,6 +21,7 @@
 import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.api.common.Plan;
 import org.apache.flink.api.common.PlanExecutor;
+import org.apache.flink.configuration.Configuration;
 
 /**
  * An {@link ExecutionEnvironment} that sends programs 
@@ -35,6 +36,8 @@ public class RemoteEnvironment extends ExecutionEnvironment {
  protected final int port;
 
  private final String[] jarFiles;
+
+ private Configuration clientConfiguration;
 
  /**
  * Creates a new RemoteEnvironment that points to the master (JobManager) described by the
@@ -65,7 +68,7 @@ public RemoteEnvironment(String host, int port, String... jarFiles) {
  public JobExecutionResult execute(String jobName) throws Exception {
  Plan p = createProgramPlan(jobName);
 
- PlanExecutor executor = PlanExecutor.createRemoteExecutor(host, port, jarFiles);
+ PlanExecutor executor = PlanExecutor.createRemoteExecutor(host, port, clientConfiguration, jarFiles);
  executor.setPrintStatusDuringExecution(p.getExecutionConfig().isSysoutLoggingEnabled());
 
  this.lastJobExecutionResult = executor.executePlan(p);
@@ -78,7 +81,7 @@ public String getExecutionPlan() throws Exception {
  p.setDefaultParallelism(getParallelism());
  registerCachedFilesWithPlan(p);
 
- PlanExecutor executor = PlanExecutor.createRemoteExecutor(host, port, jarFiles);
+ PlanExecutor executor = PlanExecutor.createRemoteExecutor(host, port, clientConfiguration, jarFiles);
  return executor.getOptimizerPlanAsJSON(p);
  }
 
@@ -87,4 +90,8 @@ public String toString() {
  return "Remote Environment (" + this.host + ":" + this.port + " - parallelism = " +
  (getParallelism() == -1 ? "default" : getParallelism()) + ") : " + getIdString();
  }
+
+ public void setClientConfiguration(Configuration clientConfiguration) {
+ this.clientConfiguration = clientConfiguration;
+ }
 }

flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala

@@ -134,7 +134,7 @@ class ExecutionEnvironment(javaEnv: JavaEnv) {
  def getId: UUID = {
  javaEnv.getId
  }
- 
+
  /**
  * Gets the JobExecutionResult of the last executed job.
  */
@@ -181,13 +181,13 @@ class ExecutionEnvironment(javaEnv: JavaEnv) {
  def addDefaultKryoSerializer(clazz: Class[_], serializer: Serializer[_]): Unit = {
  javaEnv.addDefaultKryoSerializer(clazz, serializer)
  }
- 
+
  /**
  * Registers the given type with the serialization stack. If the type is eventually
  * serialized as a POJO, then the type is registered with the POJO serializer. If the
  * type ends up being serialized with Kryo, then it will be registered at Kryo to make
  * sure that only tags are written.
- * 
+ *
  */
  def registerType(typeClass: Class[_]) {
  javaEnv.registerType(typeClass)
@@ -707,5 +707,32 @@ object ExecutionEnvironment {
  javaEnv.setParallelism(parallelism)
  new ExecutionEnvironment(javaEnv)
  }
+
+ /**
+ * Creates a remote execution environment. The remote environment sends (parts of) the program
+ * to a cluster for execution. Note that all file paths used in the program must be accessible
+ * from the cluster. The custom configuration file is used to configure Akka specific
+ * configuration parameters for the Client only; Program parallelism can be set via
+ * [[ExecutionEnvironment.setParallelism]].
+ *
+ * Cluster configuration has to be done in the remotely running Flink instance.
+ *
+ * @param host The host name or address of the master (JobManager), where the program should be
+ * executed.
+ * @param port The port of the master (JobManager), where the program should be executed.
+ * @param clientConfiguration Pass a custom configuration to the Client.
+ * @param jarFiles The JAR files with code that needs to be shipped to the cluster. If the
+ * program uses user-defined functions, user-defined input formats, or any
+ * libraries, those must be provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ def createRemoteEnvironment(
+ host: String,
+ port: Int,
+ clientConfiguration: Configuration,
+ jarFiles: String*): ExecutionEnvironment = {
+ val javaEnv = JavaEnv.createRemoteEnvironment(host, port, clientConfiguration, jarFiles: _*)
+ new ExecutionEnvironment(javaEnv)
+ }
 }
 

flink-staging/flink-scala-shell/src/main/java/org/apache/flink/api/java/ScalaShellRemoteEnvironment.java

@@ -25,6 +25,7 @@
 import org.apache.flink.api.common.PlanExecutor;
 
 import org.apache.flink.api.scala.FlinkILoop;
+import org.apache.flink.configuration.Configuration;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -78,7 +79,7 @@ public JobExecutionResult execute(String jobName) throws Exception {
  alljars.add(jarFile);
  String[] alljarsArr = new String[alljars.size()];
  alljarsArr = alljars.toArray(alljarsArr);
- PlanExecutor executor = PlanExecutor.createRemoteExecutor(host, port, alljarsArr);
+ PlanExecutor executor = PlanExecutor.createRemoteExecutor(host, port, new Configuration(), alljarsArr);
 
  executor.setPrintStatusDuringExecution(p.getExecutionConfig().isSysoutLoggingEnabled());
  return executor.executePlan(p);

flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/ExecutionEnvironmentITCase.java

@@ -80,9 +80,7 @@ public void mapPartition(Iterable<Integer> values, Collector<Integer> out) throw
  out.collect(getRuntimeContext().getIndexOfThisSubtask());
  }
  });
- List<Integer> resultCollection = new ArrayList<Integer>();
- result.output(new LocalCollectionOutputFormat<Integer>(resultCollection));
- env.execute();
+ List<Integer> resultCollection = result.collect();
  assertEquals(PARALLELISM, resultCollection.size());
  }