Apache Flink 中文用户邮件列表

flink1.10.1在yarn上无法写入kafka的问题

classic Classic list List threaded Threaded
♦
♦
 8 messages Options
maqi@sinoiov.com
Reply | Threaded
Open this post in threaded view
♦
♦
|

flink1.10.1在yarn上无法写入kafka的问题

maqi@sinoiov.com

请教各位：
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)






zhisheng
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: flink1.10.1在yarn上无法写入kafka的问题

zhisheng
hi，maqi

有完整的日志吗？在这个异常之前还有其他的异常信息吗？如果有，可以提供一下！

Best，
zhisheng

[hidden email] <[hidden email]> 于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)
>
>
>
>
>
>
>
maqi@sinoiov.com
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Re: flink1.10.1在yarn上无法写入kafka的问题

maqi@sinoiov.com
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
 
[hidden email] <[hidden email]> 于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)
>
>
>
>
>
>
>
LakeShen
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Re: flink1.10.1在yarn上无法写入kafka的问题

LakeShen
Hi,

从日志看出，应该是你提交到 Yarn 的环境，这个环境和你的测试环境的 kafka 连接不上，获取不到元数据。

这里你检查一下你的 Yarn 环境，Flink kafka broker 地址是否是测试环境的 kafka broker 地址。

Best,
LakeShen

[hidden email] <[hidden email]> 于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
>
> [hidden email] <[hidden email]> 于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)
> >
> >
> >
> >
> >
> >
> >
>
LakeShen
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Re: flink1.10.1在yarn上无法写入kafka的问题

LakeShen
你的 Yarn 环境，Flink 任务使用的 Kafka 地址，应该是 Yarn 环境的 kafka broker 地址。

LakeShen <[hidden email]> 于2020年7月10日周五 上午10:08写道：

> Hi,
>
> 从日志看出，应该是你提交到 Yarn 的环境，这个环境和你的测试环境的 kafka 连接不上，获取不到元数据。
>
> 这里你检查一下你的 Yarn 环境，Flink kafka broker 地址是否是测试环境的 kafka broker 地址。
>
> Best,
> LakeShen
>
> [hidden email] <[hidden email]> 于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
>>
>> [hidden email] <[hidden email]> 于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)
>> >
>> >
>> >
>> >
>> >
>> >
>> >
>>
>
maqi@sinoiov.com
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Re: flink1.10.1在yarn上无法写入kafka的问题

maqi@sinoiov.com
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. [hidden email]
地址：北京市海淀区东北旺西路8号中关村软件园27号院千方科技大厦A座（100085）



 
发件人： [hidden email]
发送时间： 2020-07-10 10:10
收件人： [hidden email]
主题： Re: Re: flink1.10.1在yarn上无法写入kafka的问题
你的 Yarn 环境，Flink 任务使用的 Kafka 地址，应该是 Yarn 环境的 kafka broker 地址。
 
LakeShen <[hidden email]> 于2020年7月10日周五 上午10:08写道：
 
> Hi,
>
> 从日志看出，应该是你提交到 Yarn 的环境，这个环境和你的测试环境的 kafka 连接不上，获取不到元数据。
>
> 这里你检查一下你的 Yarn 环境，Flink kafka broker 地址是否是测试环境的 kafka broker 地址。
>
> Best,
> LakeShen
>
> [hidden email] <[hidden email]> 于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
>>
>> [hidden email] <[hidden email]> 于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)
>> >
>> >
>> >
>> >
>> >
>> >
>> >
>>
>
Congxian Qiu
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Re: flink1.10.1在yarn上无法写入kafka的问题

Congxian Qiu
In reply to this post by LakeShen
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 <[hidden email]> 于2020年7月10日周五 上午10:10写道：

> 你的 Yarn 环境，Flink 任务使用的 Kafka 地址，应该是 Yarn 环境的 kafka broker 地址。
>
> LakeShen <[hidden email]> 于2020年7月10日周五 上午10:08写道：
>
> > Hi,
> >
> > 从日志看出，应该是你提交到 Yarn 的环境，这个环境和你的测试环境的 kafka 连接不上，获取不到元数据。
> >
> > 这里你检查一下你的 Yarn 环境，Flink kafka broker 地址是否是测试环境的 kafka broker 地址。
> >
> > Best,
> > LakeShen
> >
> > [hidden email] <[hidden email]> 于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
> >>
> >> [hidden email] <[hidden email]> 于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)
> >> >
> >> >
> >> >
> >> >
> >> >
> >> >
> >> >
> >>
> >
>
maqi@sinoiov.com
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Re: flink1.10.1在yarn上无法写入kafka的问题

maqi@sinoiov.com
hi：Congxian qiu：

topic没问题，用kafka指令创建，其他应用也能写入，我换一个kafka集群也不行



马琪
研发中心
北京中交兴路车联网科技有限公司


T. 010-50822710          M. 13701177502
F. 010-50822899          E. [hidden email]
地址：北京市海淀区东北旺西路8号中关村软件园27号院千方科技大厦A座（100085）



 
发件人： [hidden email]
发送时间： 2020-07-10 10:20
收件人： [hidden email]
主题： 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 <[hidden email]> 于2020年7月10日周五 上午10:10写道：
 
> 你的 Yarn 环境，Flink 任务使用的 Kafka 地址，应该是 Yarn 环境的 kafka broker 地址。
>
> LakeShen <[hidden email]> 于2020年7月10日周五 上午10:08写道：
>
> > Hi,
> >
> > 从日志看出，应该是你提交到 Yarn 的环境，这个环境和你的测试环境的 kafka 连接不上，获取不到元数据。
> >
> > 这里你检查一下你的 Yarn 环境，Flink kafka broker 地址是否是测试环境的 kafka broker 地址。
> >
> > Best,
> > LakeShen
> >
> > [hidden email] <[hidden email]> 于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
> >>
> >> [hidden email] <[hidden email]> 于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)
> >> >
> >> >
> >> >
> >> >
> >> >
> >> >
> >> >
> >>
> >
>
Free forum by Nabble Edit this page