[FLINK-2776][cli] print job id when submitting a job

nltran · Oct 2, 2015 · fc7369e · fc7369e
1 parent 19fd5bd
commit fc7369e
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 21 deletions.
diff --git a/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java b/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java
@@ -647,20 +647,13 @@ protected int executeProgramDetached(PackagedProgram program, Client client, int
  program.deleteExtractedLibraries();
  }
 
- if (result != null) {
- // if the job has been submitted to a detached YARN cluster, there won't be any
- // exec results, but the object will be set (for the job id)
-  if (yarnCluster != null && yarnCluster.isDetached()) {
+ if (yarnCluster != null && yarnCluster.isDetached()) {
+ yarnCluster.stopAfterJob(result.getJobID());
+ yarnCluster.disconnect();
+ }
 
- yarnCluster.stopAfterJob(result.getJobID());
- yarnCluster.disconnect();
- if (!webFrontend) {
- System.out.println("The Job has been submitted with JobID " + result.getJobID());
- }
- return 0;
- } else {
- throw new RuntimeException("Error while starting job. No Job ID set.");
- }
+ if (!webFrontend) {
+ System.out.println("Job has been submitted with JobID " + result.getJobID());
  }
 
  return 0;
@@ -683,17 +676,14 @@ protected int executeProgramBlocking(PackagedProgram program, Client client, int
 
  LOG.info("Program execution finished");
 
- if (result != null) {
- if (!webFrontend) {
- System.out.println("Job Runtime: " + result.getNetRuntime() + " ms");
- }
+ if (!webFrontend) {
+ System.out.println("Job with JobID " + result.getJobID() + " has finished.");
+ System.out.println("Job Runtime: " + result.getNetRuntime() + " ms");
  Map<String, Object> accumulatorsResult = result.getAllAccumulatorResults();
- if (accumulatorsResult.size() > 0 && !webFrontend) {
+ if (accumulatorsResult.size() > 0) {
  System.out.println("Accumulator Results: ");
  System.out.println(AccumulatorHelper.getResultsFormated(accumulatorsResult));
  }
- } else {
- LOG.info("The Job did not return an execution result");
  }
 
  return 0;

diff --git a/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java b/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
@@ -487,11 +487,18 @@ private void testDetachedPerJobYarnClusterInternal(String job) {
  "-ytm", "1024",
  "-ys", "2", // test requesting slots from YARN.
  "--yarndetached", job, tmpInFile.getAbsoluteFile().toString() , tmpOutFolder.getAbsoluteFile().toString()},
- "The Job has been submitted with JobID",
+ "Job has been submitted with JobID",
  RunTypes.CLI_FRONTEND);
 
  // it should usually be 2, but on slow machines, the number varies
  Assert.assertTrue("There should be at most 2 containers running", getRunningContainers() <= 2);
+ // give the runner some time to detach
+ for (int attempt = 0; runner.isAlive() && attempt < 5; attempt++) {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ }
+ }
  Assert.assertFalse("The runner should detach.", runner.isAlive());
  LOG.info("CLI Frontend has returned, so the job is running");