[web client] Fix webclient config forwarding

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

@@ -16,9 +16,9 @@
  * limitations under the License.
  */
 
-
 package org.apache.flink.client;
 
+import org.apache.flink.runtime.util.EnvironmentInformation;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.flink.client.web.WebInterfaceServer;
@@ -29,7 +29,6 @@
 /**
  * Main entry point for the web frontend. Creates a web server according to the configuration
  * in the given directory.
- * 
  */
 public class WebFrontend {
 	/**
@@ -38,28 +37,24 @@ public class WebFrontend {
 	private static final Logger LOG = LoggerFactory.getLogger(WebFrontend.class);
 
 	/**
-	 * Main method. accepts a single parameter, which is the config directory.
+	 * Main method. Accepts a single command line parameter, which is the config directory.
 	 * 
-	 * @param args
-	 *        The parameters to the entry point.
+	 * @param args The command line parameters.
 	 */
 	public static void main(String[] args) {
-		try {
-			// get the config directory first
-			String configDir = null;
 
-			if (args.length >= 2 && args[0].equals("--configDir")) {
-				configDir = args[1];
-			}
+		EnvironmentInformation.logEnvironmentInfo(LOG, "Web Client");
+		EnvironmentInformation.checkJavaVersion();
 
-			if (configDir == null) {
-				System.err
-					.println("Error: Configuration directory must be specified.\nWebFrontend --configDir <directory>\n");
-				System.exit(1);
-				return;
-			}
+		// check the arguments
+		if (args.length < 2 || !args[0].equals("--configDir")) {
+			LOG.error("Wrong command line arguments. Usage: WebFrontend --configDir <directory>");
+			System.exit(1);
+		}
 
+		try {
 			// load the global configuration
+			String configDir = args[1];
 			GlobalConfiguration.loadConfiguration(configDir);
 			Configuration config = GlobalConfiguration.getConfiguration();
 
@@ -68,15 +63,17 @@ public static void main(String[] args) {
 
 			// get the listening port
 			int port = config.getInteger(ConfigConstants.WEB_FRONTEND_PORT_KEY,
-				ConfigConstants.DEFAULT_WEBCLIENT_PORT);
+										ConfigConstants.DEFAULT_WEBCLIENT_PORT);
 
 			// start the server
 			WebInterfaceServer server = new WebInterfaceServer(config, port);
 			LOG.info("Starting web frontend server on port " + port + '.');
 			server.start();
 			server.join();
-		} catch (Throwable t) {
-			LOG.error("Unexpected exception: " + t.getMessage(), t);
+		}
+		catch (Throwable t) {
+			LOG.error("Exception while starting the web server: " + t.getMessage(), t);
+			System.exit(2);
 		}
 	}
 }

flink-clients/src/main/java/org/apache/flink/client/web/JobSubmissionServlet.java

@@ -89,11 +89,11 @@ public class JobSubmissionServlet extends HttpServlet {
 
 	private final Random rand;							// random number generator for UID
 
-	private final Configuration nepheleConfig;
+	private final Configuration config;
 
 
-	public JobSubmissionServlet(Configuration nepheleConfig, File jobDir, File planDir) {
-		this.nepheleConfig = nepheleConfig;
+	public JobSubmissionServlet(Configuration config, File jobDir, File planDir) {
+		this.config = config;
 		this.jobStoreDirectory = jobDir;
 		this.planDumpDirectory = planDir;
 
@@ -139,7 +139,7 @@ protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws Se
 			}
 
 			// parse the arguments
-			List<String> params = null;
+			List<String> params;
 			try {
 				params = tokenizeArguments(args);
 			} catch (IllegalArgumentException iaex) {
@@ -166,7 +166,7 @@ else if (params.get(0).equals("-p")) {
 			}
 
 			// create the plan
-			String[] options = params.isEmpty() ? new String[0] : (String[]) params.toArray(new String[params.size()]);
+			String[] options = params.isEmpty() ? new String[0] : params.toArray(new String[params.size()]);
 			PackagedProgram program;
 			FlinkPlan optPlan;
 			Client client;
@@ -178,7 +178,7 @@ else if (params.get(0).equals("-p")) {
 					program = new PackagedProgram(jarFile, assemblerClass, options);
 				}
 
-				client = new Client(nepheleConfig, program.getUserCodeClassLoader());
+				client = new Client(config, program.getUserCodeClassLoader());
 
 				optPlan = client.getOptimizedPlan(program, parallelism);
 
@@ -239,7 +239,7 @@ else if (params.get(0).equals("-p")) {
 				// we have a request to show the plan
 
 				// create a UID for the job
-				Long uid = null;
+				Long uid;
 				do {
 					uid = Math.abs(this.rand.nextLong());
 				} while (this.submittedJobs.containsKey(uid));
@@ -250,24 +250,33 @@ else if (params.get(0).equals("-p")) {
 
 				if (optPlan instanceof StreamingPlan) {
 					((StreamingPlan) optPlan).dumpStreamingPlanAsJSON(jsonFile);
-				} else {
+				}
+				else {
 					PlanJSONDumpGenerator jsonGen = new PlanJSONDumpGenerator();
 					jsonGen.setEncodeForHTML(true);
 					jsonGen.dumpOptimizerPlanAsJSON((OptimizedPlan) optPlan, jsonFile);
 				}
 
 				// submit the job only, if it should not be suspended
 				if (!suspend) {
-					try {
-						client.run(program,(OptimizedPlan) optPlan, false);
-					} catch (Throwable t) {
-						LOG.error("Error submitting job to the job-manager.", t);
-						showErrorPage(resp, t.getMessage());
-						return;
-					} finally {
-						program.deleteExtractedLibraries();
+					if (optPlan instanceof OptimizedPlan) {
+						try {
+							client.run(program, (OptimizedPlan) optPlan, false);
+						}
+						catch (Throwable t) {
+							LOG.error("Error submitting job to the job-manager.", t);
+							showErrorPage(resp, t.getMessage());
+							return;
+						}
+						finally {
+							program.deleteExtractedLibraries();
+						}
 					}
-				} else {
+					else {
+						throw new RuntimeException("Not implemented for Streaming Job plans");
+					}
+				}
+				else {
 					try {
 						this.submittedJobs.put(uid, client.getJobGraph(program, optPlan));
 					}
@@ -285,23 +294,27 @@ else if (params.get(0).equals("-p")) {
 
 				// redirect to the plan display page
 				resp.sendRedirect("showPlan?id=" + uid + "&suspended=" + (suspend ? "true" : "false"));
-			} else {
+			}
+			else {
 				// don't show any plan. directly submit the job and redirect to the
-				// nephele runtime monitor
+				// runtime monitor
 				try {
 					client.run(program, parallelism, false);
-				} catch (Exception ex) {
+				}
+				catch (Exception ex) {
 					LOG.error("Error submitting job to the job-manager.", ex);
 					// HACK: Is necessary because Message contains whole stack trace
 					String errorMessage = ex.getMessage().split("\n")[0];
 					showErrorPage(resp, errorMessage);
 					return;
-				} finally {
+				}
+				finally {
 					program.deleteExtractedLibraries();
 				}
 				resp.sendRedirect(START_PAGE_URL);
 			}
-		} else if (action.equals(ACTION_RUN_SUBMITTED_VALUE)) {
+		}
+		else if (action.equals(ACTION_RUN_SUBMITTED_VALUE)) {
 			// --------------- run a job that has been submitted earlier, but was -------------------
 			// --------------- not executed because of a plan display -------------------
 
@@ -328,9 +341,10 @@ else if (params.get(0).equals("-p")) {
 
 			// submit the job
 			try {
-				Client client = new Client(nepheleConfig, getClass().getClassLoader());
+				Client client = new Client(config, getClass().getClassLoader());
 				client.run(job, false);
-			} catch (Exception ex) {
+			}
+			catch (Exception ex) {
 				LOG.error("Error submitting job to the job-manager.", ex);
 				resp.setStatus(HttpServletResponse.SC_BAD_REQUEST);
 				// HACK: Is necessary because Message contains whole stack trace