Add blocking mode for topic subscription

[spenczar]

Description

I'm working on short-lived command line scripts which interact with a Kafka broker. For example, one script reads a number of messages out of Kafka, changes a few of the serialized fields, and inserts them back into a new topic.

This is hard to do with confluent_kafka because it uses an asynchronous style, designed for long-term consistency. In particular, I'd like to have a script that looks like this:

client = confluent_kafka.Consumer(config)
client.assign(confluent_kafka.TopicPartion(topic=t, partition=0, offset=confluent_kafka.OFFSET_BEGINNING)
while True:
    messages = client.consume(100, timeout=1)
    if len(messages) == 0: break # We have reached the end of the topic partition.
    handle(messages)

But this doesn't work because client.assign returns immediately - before we have actually managed to assign, and so client.consume returns an empty list. This took me a few hours to understand. At a minimum, I wish the documentation were clearer about this.

There is maybe a way to do this with callbacks, but it gets hairy. A far nicer API would be to have client.assign(..., blocking=True). Analogous stuff for the subscribe method would be great. This mode would block until assignment either succeeds or fails, and if it fails, it should raise an error.

I think this would be helpful when constructing the consumer or producer too. Right now, I have trouble knowing whether I have put in the right broker URL because the consumer or producer will retry connecting indefinitely without a loud error message.

Checklist

Please provide the following information:

• confluent-kafka-python and librdkafka version (confluent_kafka.version() and confluent_kafka.libversion()):

>>> import confluent_kafka
>>> confluent_kafka.libversion()
('1.4.0', 17039615)
>>> confluent_kafka.version()
('1.4.1', 17039616)

[edenhill]

What if you assign() to two partitions, where the leader for the two partitions are on different brokers, and only one of those brokers can be connected to?
That one partition will be consumable, the second will not.
Should this block the assign()?

[edenhill]

whether I have put in the right broker URL because the consumer or producer will retry connecting indefinitely without a loud error message.

The v1.5.0 release will improve on the default error reporting. Currently some error logs will be silenced unless an error_cb is configured.

[spenczar]

What if you assign() to two partitions, where the leader for the two partitions are on different brokers, and only one of those brokers can be connected to?
That one partition will be consumable, the second will not.
Should this block the assign()?

Yes, I would want to only return when we've succeeded. If we fail (because we timed out trying to connect), then the function should raise an exception.

When does confluent_kafka first attempt to connect? If it's when I instantiate a new confluent_kafka.Consumer, then I'd really want that to block.

The v1.5.0 release will improve on the default error reporting. Currently some error logs will be silenced unless an error_cb is configured.

That sounds great. What will the mechanism be? Will it be by raising exceptions?

[edenhill]

Some background on the error philosophy:
https://github.com/edenhill/librdkafka/wiki/FAQ#i-want-to-get-an-event-when-brokers-come-up-and-down

[spenczar]

I see. I don't understand how to use this library well in a command-line application, then.

Consumer: if less than P% of the expected data rate has been received in the last X minutes, something is wrong. Producer: if more than P% of messages failed producing during the last X minutes, something is wrong.

This is easily measured in the application logic.

In my case, I am working on a command line client for a new telescope's data pipeline. We don't have an expected data rate yet. We want to stay flexible in that regard, since it will probably change hugely over the course of a night or months as instruments are changed. Meanwhile, we want to rule out network problems, which are much more dangerous (and likely!) since we're running this client across heterogeneous networks, sometimes over the internet. Disconnection needs to be a noisy event, and certainty about whether data has been received is really important to me.

The FAQ you linked basically says "this is easy to do in the application," as I understand it. I don't agree. Maybe I'm boneheaded but I've had lots of scientists complain that our tools sometimes hang and do nothing and they can't figure out why; it virtually always has to do with a network problem connecting to any brokers.

The error philosophy laid out in that document is making it hard for me to structure my application's code. It makes some sense for a very long-lived service in a stable deployment, but I don't see how to apply it to my case, in which Kafka is a remote API, connection problems are par for the course, and we don't benefit from background workers that might eventually reconnect.

I hope that makes sense. Do you think a blocking mode is worth adding to some of these methods?

[mjuric]

@edenhill Thanks for the pointers -- I've been interested in the same!

Maybe a different way to express @spenczar's concern is to ask is there a recommended way to have the library notify the application code that a message would not have been received over the past X seconds (for any issue, including disconnects), irrespective of whether any messages were actually sent over the past X seconds? I.e., to always know the topic is silent because there were no messages, and not because a partition may be disconnected. Or is the only way to do this by injecting heartbeats that the client would monitor for (which in our case would add a significant bandwidth overhead).

Our situation is one of topics being quiet for a long time (days), but when a message is produced it's a) extremely time-sensitive (seconds matter) and b) has a huge value (the messages are notifications of gravitational wave detections that get sent out to telescopes for follow-up). The situation we're trying to avoid is someone's telescope getting silently disconnected w/o them noticing (and missing a candidate GW event because of it).