hi ,LakeShen 对,测试环境包括yarn集群和kafka集群,他们想联通的
配置的是测试环境的kafka broker的地址 road.kafka.brokers=172.17.47.134:9092,172.17.47.135:9092,172.17.47.136:9092 road.kafka.topic=road-map road.kafka.group.id=ins-001 road.kafka.transaction.timeout.ms=300000 马琪 研发中心 北京中交兴路车联网科技有限公司 T. 010-50822710 M. 13701177502 F. 010-50822899 E. m...@sinoiov.com 地址:北京市海淀区东北旺西路8号中关村软件园27号院千方科技大厦A座(100085) 发件人: LakeShen 发送时间: 2020-07-10 10:10 收件人: user-zh 主题: Re: Re: flink1.10.1在yarn上无法写入kafka的问题 你的 Yarn 环境,Flink 任务使用的 Kafka 地址,应该是 Yarn 环境的 kafka broker 地址。 LakeShen <shenleifight...@gmail.com> 于2020年7月10日周五 上午10:08写道: > Hi, > > 从日志看出,应该是你提交到 Yarn 的环境,这个环境和你的测试环境的 kafka 连接不上,获取不到元数据。 > > 这里你检查一下你的 Yarn 环境,Flink kafka broker 地址是否是测试环境的 kafka broker 地址。 > > Best, > LakeShen > > m...@sinoiov.com <m...@sinoiov.com> 于2020年7月9日周四 下午9:21写道: > >> hi:zhisheng: >> >> 这是TM日志,在这之前没有任何错误日志, >> >> 代码逻辑很简单: >> SingleOutputStreamOperator<ConcurrentLinkedQueue<ProtocolEvent>> >> sourceStream = env.addSource(source) >> .setParallelism(2) >> .uid("DataProcessSource") >> .flatMap(new DataConvertFunction()) >> .setParallelism(2) >> .uid("DataProcessDataCovert") >> .keyBy(new KeySelectorFunction()) >> .process(new DataCleanFunction()) >> .setParallelism(2) >> .uid("DataProcessDataProcess"); >> >> AsyncDataStream.orderedWait( >> sourceStream, >> new AsyncDataCleanFunction(), >> EnvUtil.TOOL.getLong(Constant.ASYNC_TOMEOUT), >> TimeUnit.MILLISECONDS, >> EnvUtil.TOOL.getInt(Constant.ASYNC_CAPACITY) >> ).uid("DataProcessAsync") >> .setParallelism(2) >> .addSink(sink) >> .uid("DataProcessSinkKafka") >> .setParallelism(2); >> >> 2020-07-09 19:33:37,291 WARN >> org.apache.kafka.clients.consumer.ConsumerConfig - The configuration >> 'gps.kafka.sasl' was supplied but isn't a known config. >> 2020-07-09 19:33:37,291 WARN >> org.apache.kafka.clients.consumer.ConsumerConfig - The configuration >> 'java.ext.dirs' was supplied but isn't a known config. >> 2020-07-09 19:33:37,291 WARN >> org.apache.kafka.clients.consumer.ConsumerConfig - The configuration >> 'java.class.version' was supplied but isn't a known config. >> 2020-07-09 19:33:37,291 INFO org.apache.kafka.common.utils.AppInfoParser >> - Kafka version: 2.2.0 >> 2020-07-09 19:33:37,291 INFO org.apache.kafka.common.utils.AppInfoParser >> - Kafka commitId: 05fcfde8f69b0349 >> 2020-07-09 19:33:38,482 INFO com.sinoi.rt.common.protocol.HttpPoolUtil - >> http pool init,maxTotal:18,maxPerRoute:6 >> 2020-07-09 19:33:38,971 WARN org.apache.kafka.clients.NetworkClient - >> [Producer clientId=producer-1] Error while fetching metadata with >> correlation id 8 : {=INVALID_TOPIC_EXCEPTION} >> 2020-07-09 19:33:38,974 INFO >> org.apache.kafka.clients.producer.KafkaProducer - [Producer >> clientId=producer-1] Closing the Kafka producer with timeoutMillis = >> 9223372036854775807 ms. >> 2020-07-09 19:33:41,612 INFO org.apache.flink.runtime.taskmanager.Task - >> async wait operator -> Sink: Unnamed (1/2) >> (cdbe008dcdb76813f88c4a48b9907d77) switched from RUNNING to FAILED. >> org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Failed >> to send data to Kafka: >> at >> org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.checkErroneous(FlinkKafkaProducer.java:1225) >> at >> org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.invoke(FlinkKafkaProducer.java:767) >> at >> org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.invoke(FlinkKafkaProducer.java:99) >> at >> org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.invoke(TwoPhaseCommitSinkFunction.java:235) >> at >> org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:56) >> at >> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:641) >> at >> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:616) >> at >> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:596) >> at >> org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:53) >> at >> org.apache.flink.streaming.api.operators.async.queue.StreamRecordQueueEntry.emitResult(StreamRecordQueueEntry.java:65) >> at >> org.apache.flink.streaming.api.operators.async.queue.OrderedStreamElementQueue.emitCompletedElement(OrderedStreamElementQueue.java:71) >> at >> org.apache.flink.streaming.api.operators.async.AsyncWaitOperator.outputCompletedElement(AsyncWaitOperator.java:279) >> at >> org.apache.flink.streaming.api.operators.async.AsyncWaitOperator.access$100(AsyncWaitOperator.java:76) >> at >> org.apache.flink.streaming.api.operators.async.AsyncWaitOperator$ResultHandler.processResults(AsyncWaitOperator.java:351) >> at >> org.apache.flink.streaming.api.operators.async.AsyncWaitOperator$ResultHandler.lambda$processInMailbox$0(AsyncWaitOperator.java:336) >> at >> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.run(StreamTaskActionExecutor.java:87) >> at >> org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:78) >> at >> org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:255) >> at >> org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:186) >> at >> org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:485) >> at >> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:469) >> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:708) >> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:533) >> at java.lang.Thread.run(Thread.java:748) >> Caused by: org.apache.kafka.common.errors.InvalidTopicException: >> 2020-07-09 19:33:41,615 INFO org.apache.flink.runtime.taskmanager.Task - >> Freeing task resources for async wait operator -> Sink: Unnamed (1/2) >> (cdbe008dcdb76813f88c4a48b9907d77). >> 2020-07-09 19:33:41,615 INFO org.apache.flink.runtime.taskmanager.Task - >> Ensuring all FileSystem streams are closed for task async wait operator -> >> Sink: Unnamed (1/2) >> >> >> >> >> >> 发件人: zhisheng >> 发送时间: 2020-07-09 21:06 >> 收件人: user-zh >> 主题: Re: flink1.10.1在yarn上无法写入kafka的问题 >> hi,maqi >> >> 有完整的日志吗?在这个异常之前还有其他的异常信息吗?如果有,可以提供一下! >> >> Best, >> zhisheng >> >> m...@sinoiov.com <m...@sinoiov.com> 于2020年7月9日周四 下午7:57写道: >> >> > >> > 请教各位: >> > flink任务在本机写入测试环境kafka集群没问题, >> > >> > 但是上传到yarn环境,就是写不进去,其他job运行在yarn可以写入测试环境的kafka >> > >> > 异常信息如下: >> > >> > 2020-07-09 19:17:33,126 INFO >> > org.apache.flink.runtime.executiongraph.ExecutionGraph - >> > KeyedProcess (1/2) (9449b1e3b758a40fb5e1e60cf84fd844) switched from >> > DEPLOYING to RUNNING. >> > 2020-07-09 19:17:33,164 INFO >> > org.apache.flink.runtime.executiongraph.ExecutionGraph - >> > KeyedProcess (2/2) (bc6eefd911cf44412121939d0afa6a81) switched from >> > DEPLOYING to RUNNING. >> > 2020-07-09 19:17:39,049 INFO >> > org.apache.flink.runtime.executiongraph.ExecutionGraph - async >> wait >> > operator -> Sink: Unnamed (1/2) (cfc31005099a8ad7e44a94dc617dd45f) >> switched >> > from RUNNING to FAILE >> > D. >> > org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Failed >> to >> > send data to Kafka: >> > at >> > >> org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.checkErroneous(FlinkKafkaProducer.java:1225) >> > at >> > >> org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.invoke(FlinkKafkaProducer.java:767) >> > at >> > >> org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.invoke(FlinkKafkaProducer.java:99) >> > at >> > >> org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.invoke(TwoPhaseCommitSinkFunction.java:235) >> > at >> > >> org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:56) >> > at >> > >> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:641) >> > at >> > >> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:616) >> > at >> > >> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:596) >> > at >> > >> org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:53) >> > at >> > >> org.apache.flink.streaming.api.operators.async.queue.StreamRecordQueueEntry.emitResult(StreamRecordQueueEntry.java:65) >> > at >> > >> org.apache.flink.streaming.api.operators.async.queue.OrderedStreamElementQueue.emitCompletedElement(OrderedStreamElementQueue.java:71) >> > >> > >> > >> > >> > >> > >> > >> >