報(bào)錯(cuò)信息

Caused by: org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
	at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:138)
	at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:82)
	at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:207)
	at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:197)
	at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:188)
	at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:677)
	at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:79)
	at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:435)
	at sun.reflect.GeneratedMethodAccessor15.invoke(Unknown Source)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:497)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:305)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:212)
	at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:77)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:158)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor$$Lambda$152/1531429858.apply(Unknown Source)
	at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)
	at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)
	at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)
	at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
	at akka.actor.Actor$class.aroundReceive(Actor.scala:517)
	at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)
	at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)
	at akka.actor.ActorCell.invoke(ActorCell.scala:561)
	at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
	at akka.dispatch.Mailbox.run(Mailbox.scala:225)
	at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
	... 4 more
Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata

程序配置

        //2、從Kafka讀取數(shù)據(jù)
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "xx.xxx.xxx.xx:9092");
        properties.setProperty("group.id", "consumer-group");

嘗試

1、在網(wǎng)上搜了半天，大多數(shù)都是說需要改kafka的server.properties配置，指明0.0.0.0，外網(wǎng)才能訪問（其實(shí)是不對(duì)的，壓根就不需要改，kafka安裝好里面參數(shù)是啥就是啥）。

listeners=PLAINTEXT://0.0.0.0:9092

2、還有說程序中引入的scala依賴需要跟Linux上運(yùn)行的kafka內(nèi)嵌的scala版本一致（這個(gè)確實(shí)需要對(duì)應(yīng)），但是改完都沒什么用。
3、阿里云服務(wù)器的用戶組端口也開放了。
4、后面就又搜到了一篇文章，試了試，原來真的是阿里云服務(wù)器有別名配置的原因。

最終解決

1、windows使用telnet工具排查下，發(fā)現(xiàn)能夠正常通信，說明原因不在kafka配置這一塊。

telnet xx.xxx.xxx.xx 9092

2、此時(shí)打開程序debug級(jí)別日志（開始默認(rèn)是error級(jí)別，少了很多重要日志）查看，發(fā)現(xiàn)日志不停在報(bào)不知道該主機(jī)信息的錯(cuò)誤，此時(shí)才發(fā)現(xiàn)這個(gè)主機(jī)名竟然是自己服務(wù)器的別名，明明自己配的是ip地址。

5464 [Kafka Fetcher for Source: Custom Source -> Sink: Print to Std. Out (6/16)#0] WARN  org.apache.kafka.clients.NetworkClient  - [Consumer clientId=consumer-consumer-group-27, groupId=consumer-group] Error connecting to node gZervbrgme7rrth0rthjrthu545:9092 (id: 0 rack: null)
java.net.UnknownHostException: gZervbrgme7rrth0rthjrthu545
	at java.net.InetAddress.getAllByName0(InetAddress.java:1259)
	at java.net.InetAddress.getAllByName(InetAddress.java:1171)
	at java.net.InetAddress.getAllByName(InetAddress.java:1105)
	at org.apache.kafka.clients.ClientUtils.resolve(ClientUtils.java:104)
	at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.currentAddress(ClusterConnectionStates.java:403)
	at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.access$200(ClusterConnectionStates.java:363)
	at org.apache.kafka.clients.ClusterConnectionStates.currentAddress(ClusterConnectionStates.java:151)
	at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:949)
	at org.apache.kafka.clients.NetworkClient.access$500(NetworkClient.java:71)
	at org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate(NetworkClient.java:1122)
	at org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate(NetworkClient.java:1010)
	at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:545)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:262)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)

3、于是便將Windows機(jī)器上的hosts文件改了下映射，最后發(fā)現(xiàn)可以正常使用了。

xx.xxx.xxx.xx gZervbrgme7rrth0rthjrthu545

文章參考：https://blog.csdn.net/maoyuanming0806/article/details/80553632文章來源地址http://www.zghlxwxcb.cn/news/detail-524590.html

到了這里，關(guān)于flink連接kafka報(bào)：org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！