卡夫卡不断重新平衡消费者

我们在一个组中有 10 个消费者在监听一个主题。经常发生的事情是看到消费者经常被重新平衡(这完全停止了消费者过程一段时间(。

# ./kafka-consumer-groups.sh --describe --bootstrap-server localhost:9092  --describe  --group ParserKafkaPipeline | grep -e ParserBody | sort
ParserBodyToParse 0          99              99              0               consumer-1-f29b7eb7-b871-477c-af52-446fbf4b0496  /10.12.18.58    consumer-1
ParserBodyToParse 1          97              97              0               consumer-10-6639ee02-8e68-40e6-aca1-eabd89bf828e /10.12.18.58    consumer-10
ParserBodyToParse 2          97              97              0               consumer-11-c712db8b-0396-4388-9e3a-e8e342355547 /10.12.18.58    consumer-11
ParserBodyToParse 3          97              98              1               consumer-12-0cc6fe12-d640-4344-91c0-f15e63c20cca /10.12.18.58    consumer-12
ParserBodyToParse 4          97              98              1               consumer-13-b904a958-141d-412e-83ea-950cd51e25e0 /10.12.18.58    consumer-13
ParserBodyToParse 5          97              98              1               consumer-14-7c70ba88-8b8c-4fad-b15b-cf7692a4b9ce /10.12.18.58    consumer-14
ParserBodyToParse 6          98              98              0               consumer-15-f0983c3d-8704-4127-808d-ec8b6b847008 /10.12.18.58    consumer-15
ParserBodyToParse 7          97              97              0               consumer-18-de5d20dd-217c-4db2-9b39-e2fdbca386e9 /10.12.18.58    consumer-18
ParserBodyToParse 8          98              98              0               consumer-5-bdeaf30a-d2bf-4aec-86ea-9c35a7acfe21  /10.12.18.58    consumer-5
ParserBodyToParse 9          98              98              0               consumer-9-4de1bf17-9474-4bd4-ae61-4ab254f52863  /10.12.18.58    consumer-9
# ./kafka-consumer-groups.sh --describe --bootstrap-server localhost:9092  --describe  --group ParserKafkaPipeline | grep -e ParserBody | sort
Warning: Consumer group 'ParserKafkaPipeline' is rebalancing.
ParserBodyToParse 0          99              99              0               -               -               -
ParserBodyToParse 1          99              99              0               -               -               -
ParserBodyToParse 2          99              99              0               -               -               -
ParserBodyToParse 3          99              100             1               -               -               -
ParserBodyToParse 4          99              100             1               -               -               -
ParserBodyToParse 5          99              100             1               -               -               -
ParserBodyToParse 6          100             100             0               -               -               -
ParserBodyToParse 7          99              99              0               -               -               -
ParserBodyToParse 8          100             100             0               -               -               -
ParserBodyToParse 9          100             100             0               -               -               -

请注意上面第二个调用中的警告。

使用这些消息可能需要很长时间，但不应超过两分钟。我检查了consumer.poll的限制是 5 分钟，这应该不是问题。是否有一些日志来检查到底发生了什么？

更新：

我们使用Kafka 2.2.1和Java consumer。我们没有更改默认值 max.session 和 max.heartbeat 。消费者基本上正在等待来自其他服务的 IO，因此它没有使用任何 CPU——这就是为什么我希望心跳应该正常工作。

我们的消费者准则如下：

    inline fun <reified T : Any> consume(
            topic: KafkaTopic,
            groupId: String,
            batchSize: Int = 50,
            crossinline consume: (key: String?, value: T) -> (Unit)
    ) = thread {
        val consumerProperties = Properties()
        consumerProperties.putAll(properties)
        consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId)
        consumerProperties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, batchSize)
        val consumer = KafkaConsumer<String?, ByteArray>(consumerProperties)
        consumer.subscribe(listOf(topic.toString()))
        while (true) try {
            val records = consumer.poll(Duration.ofMinutes(pollDurationMinutes))
            log.debug("Topic $topic consumed by group $groupId: ${records.count()} records.")
            records.forEach { record -> consumeRecord(record, topic, consume) }
        } catch (e: Exception) {
            log.fatal("Couldn't consume records: ${e.message}.", e)
            // sleep to prevent logging hell when connection failure
            Thread.sleep(1000)
        }
    }

频繁的重新平衡通常是因为使用者处理批处理的时间太长。发生这种情况是因为消费者长时间处理批处理(并且未发送检测信号(，因此代理认为消费者丢失了，他们开始重新平衡。

我建议通过减少max.partition.fetch.bytes的值来创建较小的批次，或者通过增加heartbeat.interval.ms的值来延长/增加心跳间隔。

我认为Giorgos答案的第一部分是正确的，直到".....长时间处理批处理"，但配置建议是针对其他问题的。

重新平衡有两个原因，轮询之间太长或检测信号之间太长。日志应该告诉您哪个导致了重新平衡，但通常是前者。

如果问题是心跳，那么建议的配置更改可能会有所帮助，和/或session.timeout.ms。检测信号在单独的线程中运行，并允许组快速确定使用者应用程序是否已死亡。

如果问题在轮询之间太

长，并且您无法加快处理速度，则需要增加调用轮询之间允许的间隔，或减少每次轮询处理的记录数。相关属性为max.poll.interval(默认 5 分钟(或max.poll.records(默认 500(

对于任何遇到这种情况的人来说，即使他们觉得处理记录不是瓶颈：

我们最近在 Kafka Connect 运行时中遇到了一个令人讨厌的错误，它将保留心跳线程并启动更多具有相同线程名称的 Kafka Connect 任务(本质上，不会杀死较旧的任务线程和心跳线程(

在 2.3.1 版和 JIRA 中提到的其他版本中遇到了以下错误。

https://issues.apache.org/jira/browse/KAFKA-9841

https://issues.apache.org/jira/browse/KAFKA-10574

https://issues.apache.org/jira/browse/KAFKA-9184

在 Confluent Platform 版本 5.3.1 中也发生了，所以如果可能的话，请将您的 kafka connect 运行时和 connect-api 升级到最新版本。

相关内容

最新更新

热门标签：