Kafka分区滞后增加答案

【问题标题】：Kafka partition Lag increasingKafka分区滞后增加
【发布时间】：2018-04-28 13:22:19
【问题描述】：

我有一个使用 Kafka 1.0 作为队列的应用程序。 Kafka 主题有 80 个分区和 80 个消费者在运行。（Kafka-python 消费者）。

通过运行命令：

./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group mygroup  --describe

我看到其中一个分区卡在偏移处，并且随着新记录的添加，延迟不断增加。

上述命令的输出如下所示：

TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST

118 mytopic                       37         1924            2782            858        kafka-python-1.3.4-3da99d4d-63e8-4e72-967e-xxxxxxxxxxx/localhost
119 mytopic                       38         2741            2742            1          kafka-python-1.3.4-40b44482-39fc-42d0-8f55-xxxxxxxxxxx/localhost
120 mytopic                       39         2713            2713            0          kafka-python-1.3.4-4121d080-1d7c-4d6b-ac58-xxxxxxxxxxx/localhost
121 mytopic                       40         2687            2688            1          kafka-python-1.3.4-43441f6e-fd35-448e-b791-xxxxxxxxxxx/localhost

这是什么原因造成的？此外，使用 reset-offsets 命令重置偏移量也是不可取的，因为可能不会定期手动监控此服务器。

客户端在 Linux m/c 中作为并行进程在后台运行：

consumer = KafkaConsumer('mytopic', group_id='mygroup', bootstrap_servers='localhost:9092',
                     session_timeout_ms=120000, heartbeat_interval_ms=100000, max_poll_records=1,
                     auto_commit_interval_ms=100000, request_timeout_ms=350000, max_partition_fetch_bytes=3*1024*1024,
                     value_deserializer=lambda m: json.loads(m.decode('ascii')))

for message in consumer:
    msg = json.loads(message.value)
    process_message(msg)

【问题讨论】：

标签： python apache-kafka kafka-consumer-api kafka-python

【解决方案1】：

如果消费者偏移量在一段时间后没有移动，那么消费者是很可能已经停止。如果消费者偏移量正在移动，但消费者滞后（日志结尾和消费者偏移之间的差异）是增加，消费者比生产者慢。如果消费者很慢，典型的解决办法是提高并行度在消费者中。这可能需要增加分区的数量一个话题。

在 Kafka docs 阅读更多内容。

简单地说；你生产的比消耗的多。您需要提高消耗率以减少滞后。您需要添加更多消费者。如果您只是在测试，那么您的消费者很慢。

【讨论】：

消费者是后台运行的kafka-python客户端。它可能突然停止的任何原因。我检查了客户端实例的数量，看起来不错。重启消费者似乎也没有解决问题。

【解决方案2】：

我面临着类似的问题。我不是卡夫卡专家，因此需要您的意见。我有 20 个分区和 20 个工作 pod 正在运行（比例为 1:1）。在这 20 个分区中，有 2 个分区过载（不一定总是相同）。既然我还有 18 个分区，为什么不将作业发送给这些分区，并给它们过载的分区休息一下。

如果没有key，那么kafka是否总是做round-robin而不考虑分区上的当前负载？

例如：在上面的例子中，其中一个分区有 858 的延迟，几乎所有其他分区都有 0/1 的延迟。那为什么它不重新分配或开始向其他分区发送更多负载呢？

【讨论】：

如果您有新问题，请点击按钮提出问题。如果有助于提供上下文，请包含指向此问题的链接。