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
[email protected] <[email protected]> 于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)
>
>
>
>
>
>
>
