apache-kafka - Kafka消费者长时间闲置不消费

Question

您好，我有 kafka 消费者在消费数据。下面的命令给我消费者组命令超时。

kafka-consumer-groups.sh --bootstrap-server b1:9092,b2:9092,b3:9092,b4:9092,b5:9092,b6:9092,b7:9092,b8:9092,b9:9092,b10:9092,b11:9092,b12:9092,b13:9092 --describe --group testgroup

错误:执行消费者组命令失败，原因是消费者组命令在等待组初始化时超时:

所有消费者消费数据的时间都超过 26 小时。有超过 6 小时的差距，因为生产者在这 6 小时内不再生产数据。

我怀疑有一些空闲时间可能断开了消费者群体与消费者之间的联系。所有消费者都以 100 毫秒的轮询间隔进行消费 poll(100)。

这种情况已经观察了 3 次以上。感谢 Kafka 专家的任何帮助。谢谢。

代码:

@Service
public class DedupeConsumerService {

    final Logger logger = LoggerFactory.getLogger(DedupeConsumerService.class);

    @Autowired
    private TaskExecutor taskExecutor;

    @Autowired
    private PropertyConfig config;

    @Autowired
    private ApplicationContext applicationContext;

    public void consume() {

        String topic = config.getDedupServiceConsumerTopic();
        String consGroup = config.getDedupServiceConsGroup();

        Properties props = new Properties();
        props.put("enable.auto.commit", "false");
        props.put("session.timeout.ms", "20000");
        props.put("max.poll.records", "10000");

        KafkaConsumer<String, AvroSyslogMessage> consumer = new GenericConsumer<String, AvroSyslogMessage>().initialize(topic, consGroup, STREAMSERDE.STRINGDESER, STREAMSERDE.AVRODESER, props);

        logger.info("Dedupe Kafka Consumer Initialized......");

        try {
            while (true) {
                ConsumerRecords<String, AvroSyslogMessage> records = consumer.poll(100);
                if (records.count() > 0) {

                    }


                    logger.info("Number of Records:: " + records.count() + " Time took to process poll :: " + durationInMilliSec);

                }
            }

        } catch (Throwable e) {
            logger.error("Error occured while processing message", e);
            e.printStackTrace();
        } finally {
            logger.debug("dedupe kafka consume is closing");
            consumer.close();
        }

    }

}

我尝试将轮询时间设置为 Integer.MAX_VALUE 这没有帮助。

Answer 1

考虑为您的消费者配置心跳设置:https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#consumerconfigs_heartbeat.interval.ms

Heartbeat 将确保连接不会自动关闭(很像保持事件状态)。