我应该在使用 Apache Flink 的节点上的防火墙中打开哪些端口？

Question

当我尝试运行我在 Apache Flink 独立集群上的流程时，我看到以下异常：

java.lang.IllegalStateException: Update task on instance aaa0859f6af25decf1f5fc1821ffa55d @ app-2 - 4 slots - URL: akka.tcp://flink@192.168.38.98:46369/user/taskmanager failed due to:
    at org.apache.flink.runtime.executiongraph.Execution.onFailure(Execution.java:954)
    at akka.dispatch.OnFailure.internal(Future.scala:228)
    at akka.dispatch.OnFailure.internal(Future.scala:227)
    at akka.dispatch.japi$CallbackBridge.apply(Future.scala:174)
    at akka.dispatch.japi$CallbackBridge.apply(Future.scala:171)
    at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)
    at scala.runtime.AbstractPartialFunction.applyOrElse(AbstractPartialFunction.scala:28)
    at scala.concurrent.Future$$anonfun$onFailure.apply(Future.scala:136)
    at scala.concurrent.Future$$anonfun$onFailure.apply(Future.scala:134)
    at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
    at scala.concurrent.impl.ExecutionContextImpl$AdaptedForkJoinTask.exec(ExecutionContextImpl.scala:121)
    at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
Caused by: akka.pattern.AskTimeoutException: Ask timed out on [Actor[akka.tcp://flink@192.168.38.98:46369/user/taskmanager#1804590378]] after [10000 ms]
    at akka.pattern.PromiseActorRef$$anonfun.apply$mcV$sp(AskSupport.scala:333)
    at akka.actor.Scheduler$$anon.run(Scheduler.scala:117)
    at scala.concurrent.Future$InternalCallbackExecutor$.unbatchedExecute(Future.scala:599)
    at scala.concurrent.BatchingExecutor$class.execute(BatchingExecutor.scala:109)
    at scala.concurrent.Future$InternalCallbackExecutor$.execute(Future.scala:597)
    at akka.actor.LightArrayRevolverScheduler$TaskHolder.executeTask(Scheduler.scala:467)
    at akka.actor.LightArrayRevolverScheduler$$anon.executeBucket(Scheduler.scala:419)
    at akka.actor.LightArrayRevolverScheduler$$anon.nextTick(Scheduler.scala:423)
    at akka.actor.LightArrayRevolverScheduler$$anon.run(Scheduler.scala:375)
    at java.lang.Thread.run(Thread.java:745)

似乎端口 46369 被防火墙阻止了。这是真的，因为我阅读 configuration section 并只打开这些端口：

6121:
  comment: Apache Flink TaskManager (Data Exchange)
6122:
  comment: Apache Flink TaskManager (IPC)
6123:
  comment: Apache Flink JobManager
6130:
  comment: Apache Flink JobManager (BLOB Server)
8081:
  comment: Apache Flink JobManager (Web UI)

flink-conf.yaml中描述的相同端口：

jobmanager.rpc.address: app-1.stag.local
jobmanager.rpc.port: 6123
jobmanager.heap.mb: 1024

taskmanager.heap.mb: 2048
taskmanager.numberOfTaskSlots: 4
taskmanager.memory.preallocate: false

blob.server.port: 6130

parallelism.default: 4

jobmanager.web.port: 8081

state.backend: jobmanager

restart-strategy: none
restart-strategy.fixed-delay.attempts: 2
restart-strategy.fixed-delay.delay: 60s

所以，我有两个问题：

此异常与被阻止的端口有关。对吧？
我应该在防火墙上为独立的 Apache Flink 集群打开哪些端口？

更新 1

我在 masters 和 slaves 文件中发现了配置问题（我跳过了这些文件中描述的主机之间的换行符）。我修复了它，现在我看到了其他异常：

我有 2 个节点：

app-1.stag.local（带有运行ning 作业和任务管理器）
app-2.stag.local（使用运行ning 任务管理器）

正如您从这些日志中看到的，app-1.stag.local 任务管理器无法连接到其他任务管理器：

java.io.IOException: Connecting the channel failed: Connecting to remote task manager + 'app-2.stag.local/192.168.38.98:35806' has failed. This might indicate that the remote task manager has been lost.

但是 app-2.stag.local 有开放端口：

2016-03-18 16:24:14,347 INFO org.apache.flink.runtime.io.network.netty.NettyServer - Successful initialization (took 39 ms). Listening on SocketAddress /192.168.38.98:35806

所以，我认为问题与防火墙有关，但我不明白在 Apache Flink 中可以在哪里配置此端口（或端口范围）。

Answer 1

我发现了一个问题：taskmanager.data.port 参数默认设置为 0（但是 documentation 说明应该设置什么 6121）。

所以，我在 flink-conf.yaml 中设置了这个端口，现在一切正常。

我应该在使用 Apache Flink 的节点上的防火墙中打开哪些端口？

Which ports should I open in firewall on nodes with Apach Flink?

apache-flink