Github user aljoscha commented on a diff in the pull request: https://github.com/apache/flink/pull/5901#discussion_r185449426 --- Diff: flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java --- @@ -85,54 +88,75 @@ public void checkForProhibitedLogContents() { public void testDetachedMode() throws InterruptedException, IOException { LOG.info("Starting testDetachedMode()"); addTestAppender(FlinkYarnSessionCli.class, Level.INFO); - Runner runner = - startWithArgs(new String[]{"-j", flinkUberjar.getAbsolutePath(), - "-t", flinkLibFolder.getAbsolutePath(), - "-n", "1", - "-jm", "768", - "-tm", "1024", - "--name", "MyCustomName", // test setting a custom name - "--detached"}, - "Flink JobManager is now running on", RunTypes.YARN_SESSION); - + File exampleJarLocation = getTestJarPath("StreamingWordCount.jar"); + // get temporary file for reading input data for wordcount example + File tmpInFile; + try { + tmpInFile = tmp.newFile(); + FileUtils.writeStringToFile(tmpInFile, WordCountData.TEXT); + } + catch (IOException e) { + throw new RuntimeException(e); + } + Runner runner = isNewMode ? + startWithArgs(new String[]{"run", "-m", "yarn-cluster", + "-yj", flinkUberjar.getAbsolutePath(), + "-yt", flinkLibFolder.getAbsolutePath(), + "-yn", "1", + "-yjm", "768", + "-ytm", "1024", + "-ynm", "MyCustomName", // test setting a custom name + "--detached", exampleJarLocation.getAbsolutePath(), + "--input", tmpInFile.getAbsoluteFile().toString()}, + "Job has been submitted with JobID", RunTypes.CLI_FRONTEND) : + startWithArgs(new String[]{"-j", flinkUberjar.getAbsolutePath(), + "-t", flinkLibFolder.getAbsolutePath(), + "-n", "1", + "-jm", "768", + "-tm", "1024", + "--name", "MyCustomName", // test setting a custom name + "--detached"}, + "Flink JobManager is now running on", RunTypes.YARN_SESSION); // before checking any strings outputted by the CLI, first give it time to return runner.join(); - checkForLogString("The Flink YARN client has been started in detached mode"); - if (!isNewMode) { - LOG.info("Waiting until two containers are running"); - // wait until two containers are running - while (getRunningContainers() < 2) { - sleep(500); - } + LOG.info("Waiting until two containers are running"); + // wait until two containers are running + while (getRunningContainers() < 2) { + sleep(500); } // additional sleep for the JM/TM to start and establish connection long startTime = System.nanoTime(); while (System.nanoTime() - startTime < TimeUnit.NANOSECONDS.convert(10, TimeUnit.SECONDS) && !(verifyStringsInNamedLogFiles( - new String[]{"YARN Application Master started"}, "jobmanager.log") && + new String[]{isNewMode ? "JobManager successfully registered at ResourceManager" + : "YARN Application Master started"}, "jobmanager.log") && verifyStringsInNamedLogFiles( - new String[]{"Starting TaskManager actor"}, "taskmanager.log"))) { + new String[]{isNewMode ? "Successful registration at job manager" + : "Starting TaskManager actor"}, "taskmanager.log"))) { LOG.info("Still waiting for JM/TM to initialize..."); sleep(500); } LOG.info("Two containers are running. Killing the application"); + // Wait for the cluster to shutdown itself, so it wont get killed externally during shutdown. --- End diff -- A simple static sleep is inviting test instability. Can we have a more deterministic check? For example, wait for the cluster to be shutdown by polling the status or something.
---