Python Kafka使用者缺少轮询某些消息 [英] Python Kafka consumer missing to poll some messages

问题描述

我的Kafka用户的代码如下:

The code for my Kafka consumer looks like this

def read_messages_from_kafka():
    topic = 'my-topic'
    consumer = KafkaConsumer(
        bootstrap_servers=['my-host1', 'my-host2'],
        client_id='my-client',
        group_id='my-group',
        auto_offset_reset='earliest',
        enable_auto_commit=False,
        api_version=(0, 8, 2)
    )
    consumer.assign([TopicPartition(topic, 0), TopicPartition(topic, 1)])

    messages = consumer.poll(timeout_ms=kafka_config.poll_timeout_ms, max_records=kafka_config.poll_max_records)

    for partition in messages.values():
        for message in partition:
            log.info("read {}".format(message))

    if messages:
        consumer.commit()

    next_offset0, next_offset1 = consumer.position(TopicPartition(topic, 0)), consumer.position(TopicPartition(topic, 1))
    log.info("next offset0={} and offset1={}".format(next_offset0, next_offset1))

while True:
    read_messages_from_kafka()
    sleep(kafka_config.poll_sleep_ms / 1000.0)

我已经意识到，这种消费者设置无法读取所有消息.我无法重现此问题，因为它是断断续续的问题.

I have realised that this setup of consumer is not able to read all the messages. And I am not able to reproduce this as it's intermittent issue.

当我将使用 kafka-cat 的最后100条消息与该使用者进行比较时，我发现我的使用者间歇性地错过了几条消息.我的消费者怎么了?

When I compare last 100 messages using kafka-cat to this consumer, I found that my consumer intermittently misses few messages randomly. What's wrong with my consumer?

kafkacat -C -b my-host1 -X broker.version.fallback=0.8.2.1 -t my-topic -o -100

在python中使用消息的方式太多.应该只有一种，最好只有一种明显的方式.

There are just too many ways to consume messages in python. There should be one and preferably only one obvious way to do it.

推荐答案

您的Kafka客户端中缺少消息.我在此处:

There is a problem of missing messages in your Kafka client. I found solution here:

while True:
    raw_messages = consumer.poll(timeout_ms=1000, max_records=5000)
    for topic_partition, messages in raw_messages.items():
        application_message = json.loads(message.value.decode())

还存在另一个Kafka客户端: confluent_kafka.它没有这样的问题.

Also there is another Kafka client exists: confluent_kafka. It has no such problem.

这篇关于Python Kafka使用者缺少轮询某些消息的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！