Mac电脑安装Zulu Open JDK 8 使用 spring-kafka 消费不到Kafka Partition中的消息

阿里云国内75折 回扣 微信号：monov8
阿里云国际，腾讯云国际，低至75折。AWS 93折 免费开户实名账号 代冲值 优惠多多 微信号：monov8 飞机：@monov6

请添加图片描述

一、现象描述

使用Mac电脑本地启动spring-kakfa消费不到Kafka的消息监控消费组的消息偏移量发现存在Lag的消息但是本地客户端就是拉取不到通过部署到公司k8s容器上消息却能正常消费

本地启动的服务消费组监控
kafka消费组监控图
公司k8s容器服务消费组监控
kafka消费组监控图

二、环境信息

Spring Kafka版本 2.1.13.RELEASE
Kafka Client版本 1.0.2
Local JDK版本 Zulu 8.60.0.21-CA-macos-aarch64
K8s JDK版本 Oracle 1.8.0_202-b08

三、排查过程

  • 猜测是JDK版本或者JDK 对 Apple Silicon芯片兼容问题

  • Debug跟踪了KafkaConsumer poll过程并没有发现任何异常轮询拉取的线程正常循环执行只是每次都拉取到 records 为0条。

  • 决定调整kafka 日志级别看下心跳是否正常居然发现了有异常抛出看到是snappy相关类NotClassFound

SLF4J: Failed toString() invocation on an object of type [org.apache.kafka.common.protocol.types.Struct]
Reported exception:
java.lang.NoClassDefFoundError: Could not initialize class org.xerial.snappy.Snappy
	at org.xerial.snappy.SnappyInputStream.hasNextChunk(SnappyInputStream.java:435)
	at org.xerial.snappy.SnappyInputStream.read(SnappyInputStream.java:466)
	at java.io.DataInputStream.readByte(DataInputStream.java:265)
	at org.apache.kafka.common.utils.ByteUtils.readVarint(ByteUtils.java:168)
	at org.apache.kafka.common.record.DefaultRecord.readFrom(DefaultRecord.java:292)
	at org.apache.kafka.common.record.DefaultRecordBatch$1.readNext(DefaultRecordBatch.java:264)
	at org.apache.kafka.common.record.DefaultRecordBatch$RecordIterator.next(DefaultRecordBatch.java:563)
	at org.apache.kafka.common.record.DefaultRecordBatch$RecordIterator.next(DefaultRecordBatch.java:532)
	at org.apache.kafka.common.record.MemoryRecords.toString(MemoryRecords.java:292)
	at java.lang.String.valueOf(String.java:2994)
	at java.lang.StringBuilder.append(StringBuilder.java:136)
	at org.apache.kafka.common.protocol.types.Struct.toString(Struct.java:390)
	at java.lang.String.valueOf(String.java:2994)
	at java.lang.StringBuilder.append(StringBuilder.java:136)
	at org.apache.kafka.common.protocol.types.Struct.toString(Struct.java:384)
	at java.lang.String.valueOf(String.java:2994)
	at java.lang.StringBuilder.append(StringBuilder.java:136)
	at org.apache.kafka.common.protocol.types.Struct.toString(Struct.java:384)
	at org.slf4j.helpers.MessageFormatter.safeObjectAppend(MessageFormatter.java:299)
	at org.slf4j.helpers.MessageFormatter.deeplyAppendParameter(MessageFormatter.java:271)
	at org.slf4j.helpers.MessageFormatter.arrayFormat(MessageFormatter.java:233)
	at org.slf4j.helpers.MessageFormatter.arrayFormat(MessageFormatter.java:173)
	at ch.qos.logback.classic.spi.LoggingEvent.getFormattedMessage(LoggingEvent.java:293)
	at ch.qos.logback.classic.spi.LoggingEvent.prepareForDeferredProcessing(LoggingEvent.java:206)
	at ch.qos.logback.core.OutputStreamAppender.subAppend(OutputStreamAppender.java:223)
	at ch.qos.logback.core.OutputStreamAppender.append(OutputStreamAppender.java:102)
	at ch.qos.logback.core.UnsynchronizedAppenderBase.doAppend(UnsynchronizedAppenderBase.java:84)
	at ch.qos.logback.core.spi.AppenderAttachableImpl.appendLoopOnAppenders(AppenderAttachableImpl.java:51)
	at ch.qos.logback.classic.Logger.appendLoopOnAppenders(Logger.java:270)
	at ch.qos.logback.classic.Logger.callAppenders(Logger.java:257)
	at ch.qos.logback.classic.Logger.buildLoggingEventAndAppend(Logger.java:421)
	at ch.qos.logback.classic.Logger.filterAndLog_0_Or3Plus(Logger.java:383)
	at ch.qos.logback.classic.Logger.trace(Logger.java:437)
	at org.apache.kafka.common.utils.LogContext$KafkaLogger.trace(LogContext.java:135)
	at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:689)
	at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:469)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:258)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.pollNoWakeup(ConsumerNetworkClient.java:297)
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$HeartbeatThread.run(AbstractCoordinator.java:948)
[2023-09-15 14:02:27.248]^^A[TID: N/A]^^A[kafka-coordinator-heartbeat-thread | ingest-consume-group-follow-test-4]^^ATRACE^^Aorg.apache.kafka.clients.NetworkClient^^A[Consumer clientId=consumer-1, groupId=ingest-consume-group-follow-test-4] Completed receive from node 1 for FETCH with correlation id 15, received [FAILED toString()]

  • 如果了解 snappy-java这个依赖包的话到这里就对拉取不到消息原因猜测的八九不离十了因为 Kafka 服务端使用 snappy对息做了压缩并序列化为二进制进行传输如果客户端在对消息的解压与反序列化过程中抛出异常那么自然就拉取不到消息。

  • 接着解决一下snappy-java包的兼容问题通过验证升级版本可以解决此问题。
    排除kafka-client包中 snappy-java v1.1.4版本依赖

<!-- spring-kafka -->
 <dependency>
     <groupId>org.springframework.kafka</groupId>
     <artifactId>spring-kafka</artifactId>
     <exclusions>
     	 <!-- 排除 snappy-java 1.1.4 版本 -->
         <exclusion>
             <groupId>org.xerial.snappy</groupId>
             <artifactId>snappy-java</artifactId>
         </exclusion>
     </exclusions>
 </dependency>
  • 再引入高版本v1.1.8.4的依赖包
<dependency>
    <groupId>org.xerial.snappy</groupId>
    <artifactId>snappy-java</artifactId>
    <version>1.1.8.4</version>
    <scope>compile</scope>
</dependency>
  • 重新编译启动spring kafka客户端程序消费问题解决~
    在这里插入图片描述

四、疑问解答

  1. 为什么Kafka Consumer poll消息过程没有异常抛出且可以正常运行
    答待补充
  2. 为什么调整日志级别为Trace才看到异常日志抛出
    答待补充
阿里云国内75折 回扣 微信号：monov8
阿里云国际，腾讯云国际，低至75折。AWS 93折 免费开户实名账号 代冲值 优惠多多 微信号：monov8 飞机：@monov6
标签: Spring
返回列表

上一篇：Linux基本指令

下一篇：Hadoop-Hbase

“Mac电脑安装Zulu Open JDK 8 使用 spring-kafka 消费不到Kafka Partition中的消息” 的相关文章

快递驿站取件管理系统|基于SpringBoot的快递栈系统设计与实现1年前 (2023-02-02)
spring动态数据源，多数据源_spring动态多数据源1年前 (2023-02-02)
Spring Data Elasticsearch 与 Elasticsearch 的关系1年前 (2023-02-02)
Spring获取配置文件常量1年前 (2023-02-02)
SpringMVC（十一）：SpringMVC文件上传中要解决的问题1年前 (2023-02-02)
Spring Boot---(13)Spring Boot 使用JPA访问数据库1年前 (2023-02-02)
java反射在spring ioc和aop中的应用1年前 (2023-02-02)
Java：SpringBoot项目中MongoTemplate的新增、删除、更新、查询操作1年前 (2023-02-02)
Spring的生命周期1年前 (2023-02-02)
Spring-Boot 操作 ElasticSearch 6.x 详解1年前 (2023-02-02)

阿里云国际版