kafka连接错误

来源：4-8 Source API编程之对接Kafka数据

慕容3565349

2022-01-15

14:55:22,491 WARN org.apache.flink.runtime.webmonitor.WebMonitorUtils [] - Log file environment variable ‘log.file’ is not set.
14:55:22,495 WARN org.apache.flink.runtime.webmonitor.WebMonitorUtils [] - JobManager log files are unavailable in the web dashboard. Log file location not found in environment variable ‘log.file’ or configuration key ‘web.log.path’.
14:55:53,698 WARN org.apache.kafka.clients.NetworkClient [] - [Consumer clientId=consumer-test-8, groupId=test] Bootstrap broker 192.168.128.10:9092 (id: -1 rack: null) disconnected

14:56:23,492 WARN org.apache.flink.runtime.taskmanager.Task [] - Source: Custom Source -> Sink: Print to Std. Out (2/8)#0 (80f6d820c451838a843add794b31f74c) switched from INITIALIZING to FAILED with failure cause: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata

Exception in thread “main” org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:144)
at org.apache.flink.runtime.rpc.akka.AkkaInvocationHandler.lambda$invokeRpc $0 (A k k a I n v o c a t i o n H a n d l e r . j a v a : 250) a t o r g . a p a c h e . f l i n k . r u n t i m e . c o n c u r r e n t . a k k a . C l a s s L o a d i n g U t i l s . l a m b d a$ null $1 (C l a s s L o a d i n g U t i l s . j a v a : 93) a t o r g . a p a c h e . f l i n k . r u n t i m e . c o n c u r r e n t . a k k a . C l a s s L o a d i n g U t i l s . r u n W i t h C o n t e x t C l a s s L o a d e r (C l a s s L o a d i n g U t i l s . j a v a : 68) a t o r g . a p a c h e . f l i n k . r u n t i m e . c o n c u r r e n t . a k k a . C l a s s L o a d i n g U t i l s . l a m b d a$ guardCompletionWithContextClassLoader $2 (C l a s s L o a d i n g U t i l s . j a v a : 92) a t j a v a . u t i l . c o n c u r r e n t . C o m p l e t a b l e F u t u r e . u n i W h e n C o m p l e t e (C o m p l e t a b l e F u t u r e . j a v a : 774) a t j a v a . u t i l . c o n c u r r e n t . C o m p l e t a b l e F u t u r e$ UniWhenComplete.tryFire(CompletableFuture.java:750)
at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)
at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975)
at org.apache.flink.runtime.concurrent.akka.AkkaFutureUtils $1 . o n C o m p l e t e (A k k a F u t u r e U t i l s . j a v a : 47) a t a k k a . d i s p a t c h . O n C o m p l e t e . i n t e r n a l (F u t u r e . s c a l a : 300) a t a k k a . d i s p a t c h . O n C o m p l e t e . i n t e r n a l (F u t u r e . s c a l a : 297) a t a k k a . d i s p a t c h . j a p i$ CallbackBridge.apply(Future.scala:224)
at akka.dispatch.japi $C a l l b a c k B r i d g e . a p p l y (F u t u r e . s c a l a : 221) a t s c a l a . c o n c u r r e n t . i m p l . C a l l b a c k R u n n a b l e . r u n (P r o m i s e . s c a l a : 60) a t a k k a . p a t t e r n . P i p e T o S u p p o r t$ PipeableFuture$ $a n o n f u n$ pipeTo $1 . a p p l y O r E l s e (P i p e T o S u p p o r t . s c a l a : 23) a t s c a l a . c o n c u r r e n t . F u t u r e .$ anonfun$andThen$1(Future.scala:532)
at scala.concurrent.impl.Promise.liftedTree1 $1 (P r o m i s e . s c a l a : 29) a t s c a l a . c o n c u r r e n t . i m p l . P r o m i s e .$ anonfun$transform $1 (P r o m i s e . s c a l a : 29) a t s c a l a . c o n c u r r e n t . i m p l . C a l l b a c k R u n n a b l e . r u n (P r o m i s e . s c a l a : 60) a t a k k a . d i s p a t c h . B a t c h i n g E x e c u t o r$ AbstractBatch.processBatch(BatchingExecutor.scala:63)
at akka.dispatch.BatchingExecutor $B l o c k a b l e B a t c h .$ anonfun$run $1 (B a t c h i n g E x e c u t o r . s c a l a : 100) a t s c a l a . r u n t i m e . j a v a 8 . J F u n c t i o n 0$ mcV $s p . a p p l y (J F u n c t i o n 0$ mcV $s p . j a v a : 12) a t s c a l a . c o n c u r r e n t . B l o c k C o n t e x t$ .withBlockContext(BlockContext.scala:81)
at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:100)
at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:49)
at
at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:209)
at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:679)
at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:79)

at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:24)
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:20)
at scala.PartialFunction.applyOrElse(PartialFunction.scala:123)
at scala.PartialFunction.applyOrElse$(PartialFunction.scala:122)
at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:20)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
at akka.actor.Actor.aroundReceive(Actor.scala:537)
at akka.actor.Actor.aroundReceive$(Actor.scala:535)
at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:220)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:580)
at akka.actor.ActorCell.invoke(ActorCell.scala:548)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270)
at akka.dispatch.Mailbox.run(Mailbox.scala:231)
at akka.dispatch.Mailbox.exec(Mailbox.scala:243)
... 4 more

Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata

Process finished with exit code 1
图片描述