kafka使用常见问题

连接不上kafka,报下边的错

org.apache.kafka.common.KafkaException: Producer is closed forcefully.at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortBatches(RecordAccumulator.java:760) [kafka-clients-3.0.2.jar:na]at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortIncompleteBatches(RecordAccumulator.java:747) [kafka-clients-3.0.2.jar:na]at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:283) ~[kafka-clients-3.0.2.jar:na]at java.lang.Thread.run(Thread.java:750) ~[na:1.8.0_351]2024-12-31 21:04:59.505 ERROR 35092 --- [ad | producer-1] o.s.k.support.LoggingProducerListener    : Exception thrown when sending a message with key='null' and payload='{"id":null,"payerName":"payer756","payerAcc":"payer_acc756","payeeName":"payee756","payeeAcc":"payee...' to topic Fraud_acc:org.apache.kafka.common.KafkaException: Producer is closed forcefully.at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortBatches(RecordAccumulator.java:760) [kafka-clients-3.0.2.jar:na]at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortIncompleteBatches(RecordAccumulator.java:747) [kafka-clients-3.0.2.jar:na]at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:283) ~[kafka-clients-3.0.2.jar:na]at java.lang.Thread.run(Thread.java:750) ~[na:1.8.0_351]2024-12-31 21:04:59.505 ERROR 35092 --- [ad | producer-1] o.s.k.support.LoggingProducerListener    : Exception thrown when sending a message with key='null' and payload='{"id":null,"payerName":"payer757","payerAcc":"payer_acc757","payeeName":"payee757","payeeAcc":"payee...' to topic Fraud_acc:org.apache.kafka.common.KafkaException: Producer is closed forcefully.at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortBatches(RecordAccumulator.java:760) [kafka-clients-3.0.2.jar:na]at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortIncompleteBatches(RecordAccumulator.java:747) [kafka-clients-3.0.2.jar:na]at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:283) ~[kafka-clients-3.0.2.jar:na]at java.lang.Thread.run(Thread.java:750) ~[na:1.8.0_351]2024-12-31 21:04:59.505 ERROR 35092 --- [ad | producer-1] o.s.k.support.LoggingProducerListener    : Exception thrown when sending a message with key='null' and payload='{"id":null,"payerName":"payer758","payerAcc":"payer_acc758","payeeName":"payee758","payeeAcc":"payee...' to topic Fraud_acc:org.apache.kafka.common.KafkaException: Producer is closed forcefully.at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortBatches(RecordAccumulator.java:760) [kafka-clients-3.0.2.jar:na]at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortIncompleteBatches(RecordAccumulator.java:747) [kafka-clients-3.0.2.jar:na]at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:283) ~[kafka-clients-3.0.2.jar:na]at java.lang.Thread.run(Thread.java:750) ~[na:1.8.0_351]2024-12-31 21:04:59.505 ERROR 35092 --- [ad | producer-1] o.s.k.support.LoggingProducerListener    : Exception thrown when sending a message with key='null' and payload='{"id":null,"payerName":"payer759","payerAcc":"payer_acc759","payeeName":"payee759","payeeAcc":"payee...' to topic Fraud_acc:org.apache.kafka.common.KafkaException: Producer is closed forcefully.at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortBatches(RecordAccumulator.java:760) [kafka-clients-3.0.2.jar:na]at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortIncompleteBatches(RecordAccumulator.java:747) [kafka-clients-3.0.2.jar:na]at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:283) ~[kafka-clients-3.0.2.jar:na]at java.lang.Thread.run(Thread.java:750) ~[na:1.8.0_351]2024-12-31 21:04:59.505 ERROR 35092 --- [ad | producer-1] o.s.k.support.LoggingProducerListener    : Exception thrown when sending a message with key='null' and payload='{"id":null,"payerName":"payer760","payerAcc":"payer_acc760","payeeName":"payee760","payeeAcc":"payee...' to topic Fraud_acc:org.apache.kafka.common.KafkaException: Producer is closed forcefully.at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortBatches(RecordAccumulator.java:760) [kafka-clients-3.0.2.jar:na]at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortIncompleteBatches(RecordAccumulator.java:747) [kafka-clients-3.0.2.jar:na]at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:283) ~[kafka-clients-3.0.2.jar:na]at java.lang.Thread.run(Thread.java:750) ~[na:1.8.0_351]2024-12-31 21:04:59.505 ERROR 35092 --- [ad | producer-1] o.s.k.support.LoggingProducerListener    : Exception thrown when sending a message with key='null' and payload='{"id":null,"payerName":"payer761","payerAcc":"payer_acc761","payeeName":"payee761","payeeAcc":"payee...' to topic Fraud_acc:org.apache.kafka.common.KafkaException: Producer is closed forcefully.at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortBatches(RecordAccumulator.java:760) [kafka-clients-3.0.2.jar:na]at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortIncompleteBatches(RecordAccumulator.java:747) [kafka-clients-3.0.2.jar:na]at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:283) ~[kafka-clients-3.0.2.jar:na]at java.lang.Thread.run(Thread.java:750) ~[na:1.8.0_351]
解决方法

cat server.properties 文件中配置一下下边的参数, 当初是localhost所以访问丢失
在这里插入图片描述

2025-01-01 09:18:04.991  WARN 6240 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-my-group-1, groupId=my-group] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2025-01-01 09:18:05.736  INFO 6240 --- [nio-8080-exec-1] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring DispatcherServlet 'dispatcherServlet'
2025-01-01 09:18:05.736  INFO 6240 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Initializing Servlet 'dispatcherServlet'
2025-01-01 09:18:05.737  INFO 6240 --- [nio-8080-exec-1] o.s.web.servlet.DispatcherServlet        : Completed initialization in 1 ms
2025-01-01 09:18:05.843  INFO 6240 --- [nio-8080-exec-1] o.a.k.clients.producer.ProducerConfig    : ProducerConfig values: acks = -1batch.size = 16384bootstrap.servers = [192.168.1.112:9092]buffer.memory = 33554432client.dns.lookup = use_all_dns_ipsclient.id = producer-1compression.type = noneconnections.max.idle.ms = 540000delivery.timeout.ms = 120000enable.idempotence = trueinterceptor.classes = []key.serializer = class org.apache.kafka.common.serialization.StringSerializerlinger.ms = 0max.block.ms = 60000max.in.flight.requests.per.connection = 5max.request.size = 1048576metadata.max.age.ms = 300000metadata.max.idle.ms = 300000metric.reporters = []metrics.num.samples = 2metrics.recording.level = INFOmetrics.sample.window.ms = 30000partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitionerreceive.buffer.bytes = 32768reconnect.backoff.max.ms = 1000reconnect.backoff.ms = 50request.timeout.ms = 30000retries = 2147483647retry.backoff.ms = 100sasl.client.callback.handler.class = nullsasl.jaas.config = nullsasl.kerberos.kinit.cmd = /usr/bin/kinitsasl.kerberos.min.time.before.relogin = 60000sasl.kerberos.service.name = nullsasl.kerberos.ticket.renew.jitter = 0.05sasl.kerberos.ticket.renew.window.factor = 0.8sasl.login.callback.handler.class = nullsasl.login.class = nullsasl.login.refresh.buffer.seconds = 300sasl.login.refresh.min.period.seconds = 60sasl.login.refresh.window.factor = 0.8sasl.login.refresh.window.jitter = 0.05sasl.mechanism = GSSAPIsecurity.protocol = PLAINTEXTsecurity.providers = nullsend.buffer.bytes = 131072socket.connection.setup.timeout.max.ms = 30000socket.connection.setup.timeout.ms = 10000ssl.cipher.suites = nullssl.enabled.protocols = [TLSv1.2]ssl.endpoint.identification.algorithm = httpsssl.engine.factory.class = nullssl.key.password = nullssl.keymanager.algorithm = SunX509ssl.keystore.certificate.chain = nullssl.keystore.key = nullssl.keystore.location = nullssl.keystore.password = nullssl.keystore.type = JKSssl.protocol = TLSv1.2ssl.provider = nullssl.secure.random.implementation = nullssl.trustmanager.algorithm = PKIXssl.truststore.certificates = nullssl.truststore.location = nullssl.truststore.password = nullssl.truststore.type = JKStransaction.timeout.ms = 60000transactional.id = nullvalue.serializer = class org.apache.kafka.common.serialization.StringSerializer2025-01-01 09:18:05.862  INFO 6240 --- [nio-8080-exec-1] o.a.k.clients.producer.KafkaProducer     : [Producer clientId=producer-1] Instantiated an idempotent producer.
2025-01-01 09:18:05.882  INFO 6240 --- [nio-8080-exec-1] o.a.kafka.common.utils.AppInfoParser     : Kafka version: 3.0.2
2025-01-01 09:18:05.882  INFO 6240 --- [nio-8080-exec-1] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId: 25b1aea02e37da14
2025-01-01 09:18:05.882  INFO 6240 --- [nio-8080-exec-1] o.a.kafka.common.utils.AppInfoParser     : Kafka startTimeMs: 1735694285881
2025-01-01 09:18:06.127  INFO 6240 --- [ad | producer-1] org.apache.kafka.clients.Metadata        : [Producer clientId=producer-1] Resetting the last seen epoch of partition my-topic-0 to 0 since the associated topicId changed from null to ZiipuoTKS22oBX6HbBpMbQ
2025-01-01 09:18:06.128  INFO 6240 --- [ad | producer-1] org.apache.kafka.clients.Metadata        : [Producer clientId=producer-1] Cluster ID: DioEcCfQQNi6Ea50_-07Ag
2025-01-01 09:18:06.160  INFO 6240 --- [ad | producer-1] o.a.k.c.p.internals.TransactionManager   : [Producer clientId=producer-1] ProducerId set to 16 with epoch 0
2025-01-01 09:18:08.127  WARN 6240 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-my-group-1, groupId=my-group] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.

kafka消费者报错

2025-01-01 17:04:38.425 ERROR 38716 --- [. Out (20/24)#0] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-my-group-49, groupId=my-group] Offset commit failed on partition FraudAcc-0 at offset 25: The coordinator is not aware of this member.
2025-01-01 17:04:38.425  INFO 38716 --- [. Out (20/24)#0] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-my-group-49, groupId=my-group] OffsetCommit failed with Generation{generationId=-1, memberId='', protocol='null'}: The coordinator is not aware of this member.
2025-01-01 17:04:38.425  INFO 38716 --- [. Out (20/24)#0] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-my-group-49, groupId=my-group] Resetting generation due to: encountered UNKNOWN_MEMBER_ID from OFFSET_COMMIT response
2025-01-01 17:04:38.425  INFO 38716 --- [. Out (20/24)#0] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-my-group-49, groupId=my-group] Request joining group due to: encountered UNKNOWN_MEMBER_ID from OFFSET_COMMIT response
2025-01-01 17:04:38.425  WARN 38716 --- [. Out (20/24)#0] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-my-group-49, groupId=my-group] Asynchronous auto-commit of offsets {FraudAcc-0=OffsetAndMetadata{offset=25, leaderEpoch=0, metadata=''}} failed: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.

解决方法

根据你提供的日志信息,错误的根本原因在于Kafka消费者在尝试提交偏移量(offset commit)时遇到了协调者(coordinator)不知道的成员(member)。这通常发生在以下几种情况下:

  1. 消费者组重新平衡:当消费者组中的一个或多个消费者加入或离开时,Kafka会触发一次重新平衡操作。在此期间,所有消费者都会暂时失去对分区的所有权,并且需要重新加入组以获得新的分配。如果你的消费者在重新平衡后没有正确地重新加入,它将无法提交偏移量。

  2. max.poll.interval.ms超时:这个配置项定义了消费者可以花费多长时间处理来自poll()调用的消息而不返回到Kafka进行心跳检测。如果消费者花费的时间超过了这个值,Kafka认为该消费者已经死亡并触发重新平衡。这可能是由于消息处理时间过长或者消费者卡住未能及时调用poll()导致的。

  3. session.timeout.ms超时:这是另一个与消费者健康检查有关的参数,它决定了Kafka等待消费者发送心跳的最大时间。如果超过这个时间没有收到心跳,Kafka也会认为该消费者已经死亡并触发重新平衡。

  4. max.poll.records设置过高:如果你的max.poll.records配置得太高,那么每次poll()调用可能会返回大量的记录,从而增加了处理这些记录所需的时间,可能导致max.poll.interval.ms超时。

解决方案

你可以通过调整以下几个配置来解决这个问题:

  • 增加max.poll.interval.ms:如果知道你的消费者可能需要更多时间来处理一批消息,可以适当增加这个值。默认是5分钟(300,000毫秒),你可以根据实际情况调整为更长的时间。
    spring.kafka.consumer.max-poll-interval=600000  # 例如设置为10分钟
    
  • 减少max.poll.records:降低每次poll()调用返回的最大记录数,可以减少单次处理的负担,使消费者更快地完成处理并返回心跳。
    spring.kafka.consumer.max-poll-records=100  # 根据具体情况调整
    
  • 优化消息处理逻辑:确保你的消息处理逻辑尽可能高效,避免长时间阻塞的操作,如数据库查询、网络请求等。考虑使用异步处理或其他方式来加速处理过程。
  • 检查消费者的健壮性:确保消费者不会因为异常情况而卡住或崩溃。添加适当的异常处理和监控机制可以帮助快速发现和解决问题。
  • 确保消费者组ID唯一:确认每个消费者使用的消费者组ID是唯一的,除非它们确实属于同一个消费者组并且共享相同的订阅主题。

日志分析

从日志中可以看到,消费者consumer-my-group-49试图提交偏移量,但协调者并不知道这个成员的存在,因此提交失败。然后消费者重置了它的生成代(generation),并请求加入组。紧接着的日志警告指出了问题的核心——消费者在两次poll()之间的间隔超过了max.poll.interval.ms,这通常意味着消息处理时间过长。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/bicheng/65805.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Jenkins(持续集成与自动化部署)

Jenkins 是一个开源软件项目,是基于Java开发的一种持续集成工具。 官网:https://www.jenkins.io/ GitLab安装使用 安装前提:内存至少需要4G 官方网站:https://about.gitlab.com/ 安装文档:https://docs.gitlab.c…

MyBatis执行一条sql语句的流程(源码解析)

提示:文章写完后,目录可以自动生成,如何生成可参考右边的帮助文档 MyBatis执行一条sql语句的流程(源码解析) MyBatis执行sql语句的流程加载配置文件加载配置文件的流程 创建sqlsessionFactory对象解析Mapper创建sqlses…

【C++】探索C++11中的范围for循环与auto关键字

博客主页: [小ᶻ☡꙳ᵃⁱᵍᶜ꙳] 本文专栏: C 文章目录 💯前言💯1. 范围for循环的基础用法1.1 什么是范围for循环?1.2 示例代码代码解析 1.3 范围for的特点传统for循环与范围for循环对比 1.4 使用引用修改元素 💯2…

mac下载Homebrew安装nvm

通过Homebrew安装 - 国内下载地址 /bin/zsh -c "$(curl -fsSL https://gitee.com/cunkai/HomebrewCN/raw/master/Homebrew.sh)"安装nvm brew install nvm 配置nvm环境变量 export NVM_DIR“$HOME/.nvm” [ -s “/usr/local/opt/nvm/nvm.sh” ] && . “/usr/…

解決當前IP地址僅適用於本地網路

想要解決“當前IP地址僅適用於本地網路”其實並不困難。本篇文章將介紹其發生的原因以及如何解決。 “僅限本地網路”是什麼意思? 當IP地址為“僅限本地網路”時,意味著設備正在使用私人網路內部IP地址,但無法連接到互聯網。如果將本地IP視…

HANA数据库 XS 503 Service not available

http_url配置正确 在将单个数据库系统转换为多租户数据库容器系统之后,无法通过URL访问新租户数据库的独立XS引擎服务。调用相应的URL会产生这样的错误:503服务不可用 Solution (启用租户数据库引擎的嵌套模式(HDD不用&#xff…

【大模型实战篇】Mac本地部署RAGFlow的踩坑史

1. 题外话 最近一篇文章还是在11月30日写的,好长时间没有打卡了。最近工作上的事情特别多,主要聚焦在大模型的预训练、微调和RAG两个方面。主要用到的框架是Megatron-DeepSpeed,后续会带来一些分享。今天的文章主要聚焦在RAG。 近期调研了一系…

推荐5款局域网IP扫描工具,支持电脑+Android!

在日常网络管理中,快速扫描局域网中的设备和IP地址是一项基本但非常重要的任务。无论是排查网络问题还是进行设备管理,一款好用的 IP 扫描工具都能让你事半功倍。 如何选择适合自己需求的局域网 IP 扫描工具?有哪些功能强大又易于上手的工具…

python修改ppt中的文字部分及插入图片

批量修改ppt中的某个模块,或者批量制作奖状等场景会用到; import os import pandas as pd from pptx import Presentation from pptx.util import Inchesfilepath/Users/kangyongqing/Documents/kangyq/202303/分析模版/批量制作/file1时段预警_副本.pp…

计算机网络—————考研复试

第一章、计算机网络体系结构 1. OSI参考模型和TCP/IP模型: OSI与TCP/IP的记忆方法:只需把OSI的七层记住,将应用层、表示层、会话层一起记,到TCP/IP变成应用层。物理层和数据链路层换成网络接口层。把网络层换个字变成网际层。 而…

springboot534售楼管理系统(论文+源码)_kaic

摘 要 现代经济快节奏发展以及不断完善升级的信息化技术,让传统数据信息的管理升级为软件存储,归纳,集中处理数据信息的管理方式。本售楼管理系统就是在这样的大环境下诞生,其可以帮助管理者在短时间内处理完毕庞大的数据信息&am…

计算机体系结构期末复习3:GPU架构及控制流问题

目录 一、GPU设计思路 1.简化流水线、增加核数 2.单指令多线程(SIMT) 3.同时驻留大量线程 4.总思路:多线程单指令多线程 二、GPU的控制流问题 1.什么是控制流问题 2.怎么应对分支分歧 一、GPU设计思路 1.简化流水线、增加核数 2.单指…

[实用指南]如何将视频从iPhone传输到iPad

概括 将视频从 iPhone 传输到 iPad 时遇到问题?您可能知道一种方法,但不知道如何操作。此外,您要传输的视频越大,完成任务就越困难。那么如何将视频从 iPhone 传输到 iPad,特别是当您需要发送大视频文件时&#xff1f…

httpslocalhostindex 配置的nginx,一刷新就报404了

当你的Nginx配置导致页面刷新时报404错误时,通常是由于以下几个原因造成的: 静态文件路径配置错误:Nginx没有正确地指向静态文件的目录。前端路由问题:如果是SPA(单页应用),刷新页面时Nginx没有…

药片(药丸)和胶囊识别数据集,使用yolo,pasical voc xml, coco json格式标注,可识别药片和胶囊两种标签,2445张原始图片

药片(药丸)和胶囊识别数据集,使用yolo,pasical voc xml, coco json格式标注,可识别药片和胶囊两种标签,2445张原始图片 数据集分割 训练组80% 1967图片 有效集13% 317图片 测试集7% 161图片 预处…

解决winform中comboBox下拉不够长的问题

默认情况下的combobox 可以看到comboBox不够长 导致信息显示不完成 解决方法 修改属性:DropDownWidth为308 修改前修改后

HTML——43.创建表格

<!DOCTYPE html> <html><head><meta charset"UTF-8"><title>表格</title></head><body><!--table标签用来定义表格&#xff0c;border属性表示边框--><!--tr标签表示表格中的一行--><!--th标签表示表…

在 Ubuntu 上安装和使用 `opus-tools` 的详细教程

目录 1. **安装 opus-tools**2. **opus-tools 常用命令** 2.1 **opusenc - 编码音频文件为 Opus 格式**示例 1&#xff1a;将 WAV 文件编码为 Opus 格式示例 2&#xff1a;设置编码比特率示例 3&#xff1a;使用变比特率&#xff08;VBR&#xff09;示例 4&#xff1a;设置声道…

[硬件] DELL BIOS 相关注意事项

前言 前段时间重装系统. DELL BIOS属实资料少, 又难用. 这里给出相关的注意事项, 并且配上图片. BIOS相关注意事项 进入BIOS ESC/F2/ F12. 都可以进入BIOS, 当进U盘的入Win PE系统时, 使用F12 效果更佳. 关闭安全模式 切换到Boot Configuration选项,将Secure Boot选项off选…

ip归属地怎么判定?如何查看自己ip属地

在当今数字化时代&#xff0c;IP地址作为互联网通信的基础&#xff0c;扮演着至关重要的角色。而IP归属地的判定与查看&#xff0c;不仅关乎网络安全、隐私保护&#xff0c;还直接影响到社交平台的信任机制与信息传播的真实性。本文将深入探讨IP归属地的判定原理以及如何查看自…