开发者社区 > 大数据与机器学习 > 实时计算 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 909 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版
  • 相关电子书

    更多
    移动互联网测试到质量的转变 立即下载
    给ITer的技术实战进阶课-阿里CIO学院独家教材(四) 立即下载
    F2etest — 多浏览器兼容性测试整体解决方案 立即下载