kafka提交偏移量失败导致重复消费的解决

问题详情

org.springframework.kafka.KafkaException: Seek to current after exception; nested exception is org.apache.kafka.clients.consumer.CommitFailedException: Offset commit cannot be completed since the consumer is not part of an active group for auto partition assignment; it is likely that the consumer was kicked out of the group.
    at org.springframework.kafka.listener.SeekToCurrentBatchErrorHandler.handle(SeekToCurrentBatchErrorHandler.java:92)
    at org.springframework.kafka.listener.RecoveringBatchErrorHandler.handle(RecoveringBatchErrorHandler.java:124)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.handleConsumerException(KafkaMessageListenerContainer.java:1365)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1063)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.kafka.clients.consumer.CommitFailedException: Offset commit cannot be completed since the consumer is not part of an active group for auto partition assignment; it is likely that the consumer was kicked out of the group.
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.sendOffsetCommitRequest(ConsumerCoordinator.java:1116)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:983)
    at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1510)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doCommitSync(KafkaMessageListenerContainer.java:2311)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.commitSync(KafkaMessageListenerContainer.java:2306)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.commitIfNecessary(KafkaMessageListenerContainer.java:2292)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.processCommits(KafkaMessageListenerContainer.java:2106)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1097)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1031)
    ... 3 common frames omitted

解决思路

kafka的好多配置，在spring-kafka中没有明确的配置对应，但是预留了一个properties属性，可以设置所有的kafka配置

spring.kafka.properties.session.timeout.ms=10000  // 单位：毫秒

spring.kafka.properties.max.poll.interval.ms=300000 // 单位：毫秒

kafka会有一个心跳线程来同步服务端，告诉服务端自己是正常可用的，默认是3秒发送一次心跳，超过session.timeout.ms（默认10秒）服务端没有收到心跳就会认为当前消费者失效。max.poll.interval.ms决定了获取消息后提交偏移量的最大时间，超过设定的时间（默认5分钟），服务端也会认为该消费者失效。

Kafka配置max.poll.interval.ms参数

max.poll.interval.ms默认值是5分钟，如果需要加大时长就需要给这个参数重新赋值

这里解释下自己为什么要修改这个参数：因为第一次接收kafka数据，需要加载一堆基础数据，大概执行时间要8分钟，而5分钟后，kafka认为我没消费，又重新发送，导致我这边收到许多重复数据，所以我需要调大这个值，避免接收重复数据

大部分文章都是如下配置：

public static KafkaConsumer<String, String> createConsumer() {
    Properties properties = new Properties();
    properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
    properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 10000);
    return new KafkaConsumer<>(properties);
}

或是：

max.poll.interval.ms = 300000

如果需要在yml文件中配置，应该怎么写呢？

spring:
  kafka:
    consumer:
      max-poll-records: 500
  	properties:
      max.poll.interval.ms: 600000

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。

最新评论