hadoop-yarn - 从 YARN 上的另一个程序启动的 Flink 作业失败，错误为 "JobClientActor seems to have died"

Question

我是新的 flink 用户，我有以下问题。
我在 YARN 集群上使用 flink 将从 RDBMS 提取的相关数据传输到 HBase。
我在 java 上编写了具有多个 ExecutionEnvironments 的 flink 批处理应用程序(每个 RDB 表一个以并行传输表行)以按顺序传输表(因为 env.execute() 的调用被阻塞)。

我像这样开始 YARN session

export YARN_CONF_DIR=/etc/hadoop/conf
export FLINK_HOME=/opt/flink-1.3.1
export FLINK_CONF_DIR=$FLINK_HOME/conf
$FLINK_HOME/bin/yarn-session.sh -n 1 -s 4 -d -jm 2048 -tm 8096

然后我在通过 shell 脚本 transfer.sh 启动的 YARN session 上运行我的应用程序。它的内容在这里

#!/bin/bash

export YARN_CONF_DIR=/etc/hadoop/conf
export FLINK_HOME=/opt/flink-1.3.1
export FLINK_CONF_DIR=$FLINK_HOME/conf
$FLINK_HOME/bin/flink run -p 4 transfer.jar

当我从命令行手动启动这个脚本时，它工作正常 - 作业被一一提交到 YARN session ，没有错误。

现在我应该可以从另一个 Java 程序运行这个脚本了。
为此，我使用

Runtime.exec("transfer.sh");

(也许有更好的方法来做到这一点？我在 REST API 上看到过，但有一些困难，因为作业管理器是由 YARN 代理的)。
一开始像往常一样工作 - 首先将几个作业提交给 session 并成功完成。但是以下作业不会提交到 YARN session 。
在/opt/flink-1.3.1/log/flink-tsvetkoff-client-hadoop-dev1.log 我看到错误(在调试级别没有发现其他错误)

The program execution failed: JobClientActor seems to have died before the JobExecutionResult could be retrieved.

我自己尝试分析过这个问题，发现在向JobClientActor(即YARN集群)发送超时的ping请求时，在JobClient类中出现了这个错误。
我试图增加多个心跳和超时选项，如 akka.*.timeout、akka.watch.heartbeat.* 和 yarn.heartbeat-delay 选项，但它没有解决问题 - 新作业不会从 CliFrontend 提交到 YARN session 。

这两种情况(手动调用和从另一个程序调用)的环境是相同的。当我打电话

$ ps axu | grep transfer

它会给我输出

/usr/lib/jvm/java-8-oracle/bin/java -Dlog.file=/opt/flink-1.3.1/log/flink-tsvetkoff-client-hadoop-dev1.log -Dlog4j.configuration=file:/opt/flink-1.3.1/conf/log4j-cli.properties -Dlogback.configurationFile=file:/opt/flink-1.3.1/conf/logback.xml -classpath /opt/flink-1.3.1/lib/flink-metrics-graphite-1.3.1.jar:/opt/flink-1.3.1/lib/flink-python_2.11-1.3.1.jar:/opt/flink-1.3.1/lib/flink-shaded-hadoop2-uber-1.3.1.jar:/opt/flink-1.3.1/lib/log4j-1.2.17.jar:/opt/flink-1.3.1/lib/slf4j-log4j12-1.7.7.jar:/opt/flink-1.3.1/lib/flink-dist_2.11-1.3.1.jar:::/etc/hadoop/conf org.apache.flink.client.CliFrontend run -p 4 transfer.jar

我还尝试将 flink 更新到 1.4.0 版本或更改作业的并行性(甚至到 -p 1)，但仍然发生错误。

我不知道有什么不同？有什么解决方法吗？

感谢您的任何帮助。

Answer 1

最后我找到了解决该错误的方法
只需更换 Runtime.exec(...)与 new ProcessBuilder(...).inheritIO().start() .

真不知道为什么要打inheritIO在这种情况下有帮助，因为据我所知，它只是将 IO 流从子进程重定向到父进程。
但是我已经检查过，如果我注释掉这行代码，程序就会再次开始下降。