开发者社区 > 大数据与机器学习 > 实时计算 Flink > 正文

Flink目前测试环境。正在做数据处理后写入到Kafka的功能,出现报错,为什么?

Flink目前测试环境。正在做数据处理后写入到Kafka的功能。 使用的是基于Docker搭建的单节点Kafka。 然后在运行一段时间后,出现 Caused by: org.apache.flink.util.FlinkRuntimeException: Failed to send data to Kafka metrics-0@-1 with FlinkKafkaInternalProducer{transactionalId='null', inTransaction=false, closed=false} at org.apache.flink.connector.kafka.sink.KafkaWriter$WriterCallback.throwException(KafkaWriter.java:436) at org.apache.flink.connector.kafka.sink.KafkaWriter$WriterCallback.lambda$onCompletion$0(KafkaWriter.java:417) at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50) at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMail(MailboxProcessor.java:398) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:367) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:352) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:229) at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:831) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:780) at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:935) at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:914) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:728) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550) at java.base/java.lang.Thread.run(Thread.java:834) Caused by: org.apache.kafka.common.errors.NotLeaderOrFollowerException: For requests intended only for the leader, this error indicates that the broker is not the current leader. For requests intended for any replica, this error indicates that the broker is not a replica of the topic partition. 尝试过将Kafka的 retries 配置为 3 ,但是仍然报错。

展开
收起
wenti 2023-01-16 19:28:06 932 0
2 条回答
写回答
取消 提交回答
  • 也有可能是kafka压力过大,可以试试批量发送(Batching),设置合适的批量发送参数,将多条记录打包成一个批次进行发送,减少I/O操作次数,提高写入性能。

    2023-04-22 16:31:58
    赞同 展开评论 打赏
  • 你的topic可能坏了 分区的Leader 没有了 你查看topic详情看看——该回答整理自钉群“【③群】Apache Flink China社区“

    2023-01-16 21:11:37
    赞同 展开评论 打赏

实时计算Flink版是阿里云提供的全托管Serverless Flink云服务,基于 Apache Flink 构建的企业级、高性能实时大数据处理系统。提供全托管版 Flink 集群和引擎,提高作业开发运维效率。

相关产品

  • 实时计算 Flink版
  • 相关电子书

    更多
    ICA安全标准组测试认证分享 立即下载
    MaxCompute基于BigBench标准的最新测试进展 立即下载
    用AI 高效测试移动应用 立即下载