Github user GJL commented on a diff in the pull request: https://github.com/apache/flink/pull/5215#discussion_r159224695 --- Diff: flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java --- @@ -743,6 +690,142 @@ private void logAndSysout(String message) { System.out.println(message); } + public static void main(final String[] args) throws Exception { + final FlinkYarnSessionCli cli = new FlinkYarnSessionCli("", ""); // no prefix for the YARN session + + final String configurationDirectory = CliFrontend.getConfigurationDirectoryFromEnv(); + + final Configuration flinkConfiguration = GlobalConfiguration.loadConfiguration(); + SecurityUtils.install(new SecurityConfiguration(flinkConfiguration)); + int retCode = SecurityUtils.getInstalledContext().runSecured(new Callable<Integer>() { + @Override + public Integer call() { + return cli.run(args, flinkConfiguration, configurationDirectory); + } + }); + System.exit(retCode); + } + + private static void runInteractiveCli( + YarnClusterClient clusterClient, + YarnApplicationStatusMonitor yarnApplicationStatusMonitor, + boolean readConsoleInput) { + try (BufferedReader in = new BufferedReader(new InputStreamReader(System.in))) { + boolean continueRepl = true; + int numTaskmanagers = 0; + long unknownStatusSince = System.currentTimeMillis(); + + while (continueRepl) { + + final ApplicationStatus applicationStatus = yarnApplicationStatusMonitor.getApplicationStatusNow(); + + switch (applicationStatus) { + case FAILED: + case CANCELED: + System.err.println("The Flink Yarn cluster has failed."); + continueRepl = false; + break; + case UNKNOWN: + if (unknownStatusSince < 0L) { + unknownStatusSince = System.currentTimeMillis(); + } + + if ((System.currentTimeMillis() - unknownStatusSince) > CLIENT_POLLING_INTERVAL_MS) { + System.err.println("The Flink Yarn cluster is in an unknown state. Please check the Yarn cluster."); + continueRepl = false; + } else { + continueRepl = repStep(in, readConsoleInput); + } + break; + case SUCCEEDED: + if (unknownStatusSince > 0L) { + unknownStatusSince = -1L; + } + + // ------------------ check if there are updates by the cluster ----------- + try { + final GetClusterStatusResponse status = clusterClient.getClusterStatus(); + + if (status != null && numTaskmanagers != status.numRegisteredTaskManagers()) { + System.err.println("Number of connected TaskManagers changed to " + + status.numRegisteredTaskManagers() + ". " + + "Slots available: " + status.totalNumberOfSlots()); + numTaskmanagers = status.numRegisteredTaskManagers(); + } + } catch (Exception e) { + LOG.warn("Could not retrieve the current cluster status. Skipping current retrieval attempt ...", e); + } + + printClusterMessages(clusterClient); + + continueRepl = repStep(in, readConsoleInput); + } + } + } catch (Exception e) { + LOG.warn("Exception while running the interactive command line interface.", e); + } + } + + private static void printClusterMessages(YarnClusterClient clusterClient) { + final List<String> messages = clusterClient.getNewMessages(); + if (messages != null && messages.size() > 0) { --- End diff -- nit: ```if (!messages.isEmpty())``` should suffice because `messages` is never null
---