consume(0) cannot trigger stats_cb?

[chengcheng-pei]

Description

I am using confluent-kafka-python 1.3. when the stats_cb enabled, but the consumer stops consuming messages. there may be memory leak.

Please refer the following section codes.

I fount that when one message causes handler_fn(batch) to raise exception, the offsets in the batch cannot be committed. The batch will store 500 messages but cannot commit the offsets. The consumer will be stuck here. It cannot consume and commit messages. And the buffer queue should be full soon. The consumer finally did not poll any messages because batch if full. And the memory usage should be bounded by partition number * 1 M, which is set by "queued.max.messages.kbytes". The memory usage should be bounded.

But in fact, I found the memory usage increased up to 100%.

I have no idea why. I guess because consumer stops consuming or polling new messages and the queue may be full. And stats is queued up and is not dispatched finally.

Any ideas welcomed. Thanks

How to reproduce

Config:

config = {
    "client.id": socket.gethostname()
    "bootstrap.servers": kafka_hosts,
    "group.id": group_id,
    "enable.auto.commit": False,
    "auto.offset.reset": "latest" if reset_to_latest else "earliest",
    "statistics.interval.ms": 15000,
   "stats_cb": simple_print_function,
    "queued.max.messages.kbytes": 1024,
}

        rdkafka_config={
            'queued.max.messages.kbytes': 1024,
            'debug': 'all',
            'queued.min.messages': 10,
            'queued.min.messages': 20,
            'fetch.message.max.bytes': 1024*512,
        },

Consumer codes:

def run(self):
    batch = []
    while not self.stopped:
        ...
        try:
            events = self.consumer.consume(
                num_messages = 500 - len(batch),
                timeout = max(0, ...),
            )
        except Exception:
            continue

        messages = [event for event in events if not event.error()]
        batch += messages
        if not batch:
            continue

        try:
            handler_fn(batch)
        except Exception:
            continue

        # Successful handler, clear batch and send a commit
        batch = []
        self.commit(...) 

Checklist

Please provide the following information:

• confluent-kafka-python and librdkafka version (confluent_kafka.version() and confluent_kafka.libversion()): 1.3
• Apache Kafka broker version:
• Client configuration: {...}
• Operating system: ubuntu
• Provide client logs (with 'debug': '..' as necessary)
• Provide broker log excerpts
• Critical issue

[josephglanville]

So this is currently a constraint of librdkafka.

When using Consumer.consume and specifying num_messages=0 no callbacks are processed as rd_kafka_q_serve_rkmessages doesn't actually dequeue any events for processing in this case. The reason this means no callbacks fire is because the callbacks themselves are inline with the fetched messages on the same queue.

Right now librdkafka doesn't provide an API that can service callbacks without also consuming messages. However the behaviour of rd_kafka_q_serve_rkmessages can be augmented to attempt to process any ops that don't require being returned to the application. I have done a PoC of that here: https://github.com/josephglanville/librdkafka/commits/callbacks-only

Unfortunately that isn't quite enough for stats_cb as the stats op is enqueued at the tail of the queue. This means it's status as high priority isn't very meaningful, perhaps it's intended to insert it at the head or use the sorted insert so it will be inserted after any flash priority events? Unsure at this point in time. In the branch above I updated it to insert it at the head of the queue for now to demonstrate this strategy works.

An alternate approach would be to provide an API that has the specific purpose of only servicing callbacks but extending the API surface is probably a high bar to clear.