Re: Re: flink1.10.1在yarn上无法写入kafka的问题
hi：Congxian qiu：
topic没问题，用kafka指令创建，其他应用也能写入，我换一个kafka集群也不行
马琪
研发中心
北京中交兴路车联网科技有限公司
T. 010-50822710 M. 13701177502
F. 010-50822899 E. m...@sinoiov.com
地址：北京市海淀区东北旺西路8号中关村软件园27号院千方科技大厦A座（100085）
发件人： Congxian Qiu
发送时间： 2020-07-10 10:20
收件人： user-zh
主题： Re: Re: flink1.10.1在yarn上无法写入kafka的问题
Hi
从 org.apache.kafka.common.errors.InvalidTopicException: 这个异常来看，是 topic
invalid 导致，具体的可以看一下 InvalidTopicException 的介绍[1], 这上面说的有可能是
名字太长，或者有非法字符等，这也可以查看一下
[1]
https://www.javadoc.io/doc/org.apache.kafka/kafka-clients/2.0.0/org/apache/kafka/common/errors/InvalidTopicException.html
Best,
Congxian
LakeShen 于2020年7月10日周五 上午10:10写道：
> 你的 Yarn 环境，Flink 任务使用的 Kafka 地址，应该是 Yarn 环境的 kafka broker 地址。
>
> LakeShen 于2020年7月10日周五 上午10:08写道：
>
> > Hi,
> >
> > 从日志看出，应该是你提交到 Yarn 的环境，这个环境和你的测试环境的 kafka 连接不上，获取不到元数据。
> >
> > 这里你检查一下你的 Yarn 环境，Flink kafka broker 地址是否是测试环境的 kafka broker 地址。
> >
> > Best,
> > LakeShen
> >
> > m...@sinoiov.com 于2020年7月9日周四 下午9:21写道：
> >
> >> hi：zhisheng：
> >>
> >> 这是TM日志，在这之前没有任何错误日志，
> >>
> >> 代码逻辑很简单：
> >> SingleOutputStreamOperator>
> >> 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.
Re: Re: flink1.10.1在yarn上无法写入kafka的问题
gt; >> 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 于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) > >> > > >> > > >> > > >> > > >> > > >> > > >> > > >> > > >
Re: Re: flink1.10.1在yarn上无法写入kafka的问题
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=30
马琪
研发中心
北京中交兴路车联网科技有限公司
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 于2020年7月10日周五 上午10:08写道：
> Hi,
>
> 从日志看出，应该是你提交到 Yarn 的环境，这个环境和你的测试环境的 kafka 连接不上，获取不到元数据。
>
> 这里你检查一下你的 Yarn 环境，Flink kafka broker 地址是否是测试环境的 kafka broker 地址。
>
> Best,
> LakeShen
>
> m...@sinoiov.com 于2020年7月9日周四 下午9:21写道：
>
>> hi：zhisheng：
>>
>> 这是TM日志，在这之前没有任何错误日志，
>>
>> 代码逻辑很简单：
>> SingleOutputStreamOperator>
>> 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.oper
Re: Re: flink1.10.1在yarn上无法写入kafka的问题
.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 于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) >> > >> > >> > >> > >> > >> > >> > >> >
Re: Re: flink1.10.1在yarn上无法写入kafka的问题
essor.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 于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) > > > > > > > > > > > > > > >
Re: Re: flink1.10.1在yarn上无法写入kafka的问题
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 于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) > > > > > > >
Re: flink1.10.1在yarn上无法写入kafka的问题
hi，maqi 有完整的日志吗？在这个异常之前还有其他的异常信息吗？如果有，可以提供一下！ Best， zhisheng 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) > > > > > > >
