2021-02-28 22:03:05 131 pool-7-thread-3 ERROR [] - 
commit failed 
org.apache.kafka.clients.consumer.CommitFailedException: 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 the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.sendOffsetCommitRequest(ConsumerCoordinator.java:713) ~[MsgAgent-jar-with-dependencies.jar:na]
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:596) ~[MsgAgent-jar-with-dependencies.jar:na]
        at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1218) ~[MsgAgent-jar-with-dependencies.jar:na]
        at com.today.eventbus.common.MsgConsumer.run(MsgConsumer.java:121) ~[MsgAgent-jar-with-dependencies.jar:na]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_161]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_161]
        at java.lang.Thread.run(Thread.java:748) [na:1.8.0_161]

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 the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.

既然Kafka觸發(fā)了Rebalance機(jī)制，那我就來(lái)說(shuō)說(shuō)Kafka觸發(fā)Rebalance的時(shí)機(jī)。

什么是Rebalance

觸發(fā)Rebalance的時(shí)機(jī)

• 組內(nèi)成員的個(gè)數(shù)發(fā)生了變化，比如有新的消費(fèi)者加入消費(fèi)組，或者離開(kāi)消費(fèi)組。組成員離開(kāi)消費(fèi)組包含組成員崩潰或者主動(dòng)離開(kāi)消費(fèi)組。
• 訂閱的主題個(gè)數(shù)發(fā)生了變化。
• 訂閱的主題分區(qū)數(shù)發(fā)生了變化。

• 設(shè)置 session.timeout.ms = 6s。
• 設(shè)置 heartbeat.interval.ms = 2s。
• 要保證 Consumer 實(shí)例在被判定為 “dead” 之前，能夠發(fā)送至少 3 輪的心跳請(qǐng)求，即 session.timeout.ms >= 3 * heartbeat.interval.ms。

拉取偏移量與提交偏移量

異常日志提示的方案

@KafkaListener(topicPartitions = {@TopicPartition(topic = KafkaConstants.TOPIC_LOGS, partitions = { "0" }) }, groupId = "kafka-consumer", containerFactory = "kafkaListenerContainerFactory")
public void consumerReceive (ConsumerRecord<?, ?> record, Acknowledgment ack){
    logger.info("topic is {}, offset is {}, value is {} n", record.topic(), record.offset(), record.value());
    try {
        Object value = record.value();
        logger.info(value.toString());
        ack.acknowledge();
    } catch (Exception e) {
        logger.error("日志消費(fèi)端異常: {}", e);
    }
}

嘗試解決

spring:
  kafka:
    consumer:
    properties:
     max.poll.interval.ms: 3600000
     max.poll.records: 50
     session.timeout.ms: 60000
     heartbeat.interval.ms: 3000

最終解決

@KafkaListener(topicPartitions = {@TopicPartition(topic = KafkaConstants.TOPIC_LOGS, partitions = { "0" }) }, groupId = "kafka-consumer", containerFactory = "kafkaListenerContainerFactory")
public void consumerReceive (ConsumerRecord<?, ?> record, Acknowledgment ack){

@KafkaListener(topicPartitions = {@TopicPartition(topic = KafkaConstants.TOPIC_LOGS, partitions = { "0" }) }, groupId = "kafka-consumer-logs", containerFactory = "kafkaListenerContainerFactory")
public void consumerReceive (ConsumerRecord<?, ?> record, Acknowledgment ack){