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 <shenleifight...@gmail.com> 于2020年7月10日周五 上午10:10写道： > 你的 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) > >> > > >> > > >> > > >> > > >> > > >> > > >> > > >> > > >
