scala中的Flink Kafka程序给出超时错误。阿帕奇。Kafka。常见的错误。TimeoutException:在60000毫秒后更新元数据失败

Question

我正在编写一个Flink-Kafka集成程序，如下所示，但Kafka出现超时错误：

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer010, 
FlinkKafkaProducer010}
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import java.util.Properties

object StreamKafkaProducer {

def main(args: Array[String]) {
val env = StreamExecutionEnvironment.getExecutionEnvironment
val properties = new Properties()
properties.setProperty("bootstrap.servers", "localhost:9092")
properties.setProperty("zookeeper.connect", "localhost:2181")
properties.setProperty("serializer.class", "kafka.serializer.StringEncoder")


val stream: DataStream[String] =env.fromElements(
  ("Adam"),
  ("Sarah"))

val kafkaProducer = new FlinkKafkaProducer010[String](
  "localhost:9092",
  "output",
  new SimpleStringSchema
)
// write data into Kafka
stream.addSink(kafkaProducer)

env.execute("Flink kafka integration  ")
}
}

从终端我可以看到Kafka和zookeeper正在运行，但当我从Intellij运行上面的程序时，它显示了这个错误：

C:\Users\amdass\workspace\flink-project-master>sbt run
Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=256m; 
support was removed in 8.0
[info] Loading project definition from C:\Users\amdass\workspace\flink-
project-master\project
[info] Set current project to Flink Project (in build 
file:/C:/Users/amdass/workspace/flink-project-master/)
[info] Compiling 1 Scala source to C:\Users\amdass\workspace\flink-project-
master\target\scala-2.11\classes...
[info] Running org.example.StreamKafkaProducer
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further 
details.
Connected to JobManager at Actor[akka://flink/user/jobmanager_1#-563113020] 
with leader session id 5a637740-5c73-4f69-a19e-c8ef7141efa1.
12/15/2017 14:41:49     Job execution switched to status RUNNING.
12/15/2017 14:41:49     Source: Collection Source(1/1) switched to SCHEDULED
12/15/2017 14:41:49     Sink: Unnamed(1/4) switched to SCHEDULED
12/15/2017 14:41:49     Sink: Unnamed(2/4) switched to SCHEDULED
12/15/2017 14:41:49     Sink: Unnamed(3/4) switched to SCHEDULED
12/15/2017 14:41:49     Sink: Unnamed(4/4) switched to SCHEDULED
12/15/2017 14:41:49     Source: Collection Source(1/1) switched to DEPLOYING
12/15/2017 14:41:49     Sink: Unnamed(1/4) switched to DEPLOYING
12/15/2017 14:41:49     Sink: Unnamed(2/4) switched to DEPLOYING
12/15/2017 14:41:49     Sink: Unnamed(3/4) switched to DEPLOYING
12/15/2017 14:41:49     Sink: Unnamed(4/4) switched to DEPLOYING
12/15/2017 14:41:50     Source: Collection Source(1/1) switched to RUNNING
12/15/2017 14:41:50     Sink: Unnamed(2/4) switched to RUNNING
12/15/2017 14:41:50     Sink: Unnamed(4/4) switched to RUNNING
12/15/2017 14:41:50     Sink: Unnamed(3/4) switched to RUNNING
12/15/2017 14:41:50     Sink: Unnamed(1/4) switched to RUNNING
12/15/2017 14:41:50     Source: Collection Source(1/1) switched to FINISHED
12/15/2017 14:41:50     Sink: Unnamed(3/4) switched to FINISHED
12/15/2017 14:41:50     Sink: Unnamed(4/4) switched to FINISHED
12/15/2017 14:42:50     Sink: Unnamed(1/4) switched to FAILED
<b>  org.apache.kafka.common.errors.TimeoutException: Failed to update metadata 
after 60000 ms. </b>

12/15/2017 14:42:50     Sink: Unnamed(2/4) switched to FAILED
org.apache.kafka.common.errors.TimeoutException: Failed to update metadata 
after 60000 ms.

12/15/2017 14:42:50     Job execution switched to status FAILING.

组织。阿帕奇。Kafka。常见的错误。TimeoutException:在60000毫秒后更新元数据失败。2017年12月15日14:42:50作业执行切换到失败状态。[错误]（run-main-0）组织。阿帕奇。Flink。运行时。客户JobExecutionException:作业执行失败。组织。阿帕奇。Flink。运行时。客户JobExecutionException:作业执行失败。在org。阿帕奇。Flink。运行时。工作经理。JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$6。在org上申请$mcV$sp（JobManager.scala:933）。阿帕奇。Flink。运行时。工作经理。JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$6。申请（JobManager.scala:876）组织。阿帕奇。Flink。运行时。工作经理。JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$6。在scala申请（JobManager.scala:876）。同时发生的impl。未来$PromiseCompletingRunnable。在scala，liftedTree1美元（未来，scala:24）。同时发生的impl。未来$PromiseCompletingRunnable。在akka跑步（Future.scala:24）。派遣任务调用。在akka运行（AbstractDispatcher.scala:40）。派遣ForkJoinExecutor配置程序$AkkaForkJoinTask。scala的执行官（AbstractDispatcher.scala:397）。同时发生的分叉连接。这是一项艰巨的任务。scala的doExec（ForkJoinTask.java:260）。同时发生的分叉连接。ForkJoinPool$WorkQueue。scala上的runTask（ForkJoinPool.java:1339）。同时发生的分叉连接。福克斯：泳池。runWorker（ForkJoinPool.java:1979）在

斯卡拉。同时发生的分叉连接。ForkJoinWorkerThread。运行（ForkJoinWorkerThread.java:107）的原因是：org。阿帕奇。Kafka。常见的错误。TimeoutException:在60000毫秒后更新元数据失败。[trace]堆栈跟踪被抑制：最后运行*：运行完整输出。JAVAlang.RuntimeException:scala的非零退出代码：1。系统。套餐美元。错误（package.scala:27）[trace]堆栈跟踪被抑制：运行上次编译：运行完整输出。[错误]（编译：运行）非零退出代码：1[错误]总时间：75秒，已完成2017年12月15日下午2:42:51

Answer 1

请检查并确保您的Kafka服务器正在运行。此错误通常发生在Flink程序无法连接到Kafka服务器时。Flink会在一定时间内自动尝试连接到Kafka服务器。一旦达到这个临界值，Flink仍然无法与Kafka建立联系，那么它就会抛出这个组织。阿帕奇。Kafka。常见的错误。超时异常

请检查您的Kafka服务器详细信息，Kafka主题，并验证您的Kafka服务器是否正在运行。