kafka max.poll.interval.ms配置太短

max.poll.interval.ms這個應該是消費者每次去kafka拉取數據最大間隔，如果超過這個間隔，服務端會認為消費者已離線。觸發rebalance

demo

    public ConsumerDemo(String topicName) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", GROUPID);
        props.put("enable.auto.commit", "false");
        props.put("max.poll.interval.ms", "1000");
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        this.consumer = new KafkaConsumer<String, String>(props);
        this.topic = topicName;
        this.consumer.subscribe(Arrays.asList(topic));
    }

5行配置自動提交為false，手動提交。6行配置 max.poll.interval.ms為1秒

    public void receiveMsg() {
        int messageNo = 1;
        System.out.println("---------開始消費---------");
        try {
            for (;;) {
                msgList = consumer.poll(1000);
                System.out.println("start sleep" + System.currentTimeMillis() / 1000);
                Thread.sleep(10000);
                System.out.println("end sleep" + System.currentTimeMillis() / 1000);
                if(null!=msgList&&msgList.count()>0){
                    for (ConsumerRecord<String, String> record : msgList) {
                        System.out.println(messageNo+"=======receive: key = " + record.key() + ", value = " + record.value()+" offset==="+record.offset());
                    }
                }else{
                    Thread.sleep(1000);
                }
                consumer.commitSync();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }

8行slepp 10秒，模擬處理消息耗時。提交消息的時候報錯

Exception in thread "main" 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:722)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:600)
    at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1211)
    at com.gxf.kafka.ConsumerDemo.receiveMsg(ConsumerDemo.java:49)
    at com.gxf.kafka.ConsumerDemo.main(ConsumerDemo.java:59)

max.poll.interval.ms 可以配置稍微大點，或者減少處理時間，每次少拉取數據，異步處理等