Can I clear the Producer output queue when flush() times out?

[peteryin21]

Description

Currently, if a producer times out (i.e when a broker goes down), messages are added to an internal queue to be published. The number of remaining messages is returned via the flush() command.

In my use case, I would like failed messages to be cleared from the outgoing queue. However, with current functionality, the next time I produce a message, both it and the other outgoing queued messages will be sent. Looking through the documentation, I didn't see any straightforward way to do this.

How to reproduce

producer = AvroProducer(kafka_producer_config)
producer.produce(
        key=key,
        value=value,
        topic=topic,
        key_schema=key_schema,
        value_schema=value_schema,
        on_delivery=delivery_callback,
    )
remaining_messages = self.client.flush(timeout)
# Here I would like to clear the outgoing queue 
if remaining_messages > 0:
    raise KafkaProducerException(
            f"Kafka producer timed out after {self.flush_timeout} seconds"
            f" with {remaining_messages} messages remaining to be published."
        )

Create any producer and give it a config with invalid port numbers in the bootstrap.servers section to simulate a broker it can't connect to. Attempt to produce several messages, and each time the remaining_messages goes up. I would like a way to clear the output queue, to ensure that the next time a successful message comes through, none of the previously failed messages are sent.

Checklist

Please provide the following information:

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

('0.11.6', 722687)
>>> confluent_kafka.version()
('0.11.6', 722432)

• Apache Kafka broker version: confluentinc/cp-kafka:5.0.0
• Client configuration: {...}
• Operating system:
• Provide client logs (with 'debug': '..' as necessary)
• Provide broker log excerpts
• Critical issue

[rnpridgeon]

librdkafka does actually expose a purge api but it is not currently exposed in Python.

https://github.com/edenhill/librdkafka/blob/master/src/rdkafka.h#L3455-L3485

I suspect what you are after will be covered by the upcoming idempotent producer feature. This feature, unlike the purge api, will be made available via the Python binding.

https://github.com/edenhill/librdkafka/blob/d306f5f7ef2aea04cdb9e7146796efd4a28d49f1/INTRODUCTION.md#idempotent-producer

[peteryin21]

@rnpridgeon I did read about the idempotent feature. In my use case, a user can submit tasks (as in something that happens in the real world) to be executed via a task request over the MQ. The failure mode I'm trying to defend against is when the broker is down and the user tries sending additional tasks (each with different uuid, which is why idempotent producer won't suffice) after seeing each previous one failed. Currently, these will build up in the internal queue, and if there were X previous failures, send X+1 tasks, when the user only wanted to submit 1.

I do realize that this use case of Kafka is not traditional as we want synchronous, immediate feedback to the user if a message fails to send and we don't have to to worry about volume. The reason we are using Kafka is because it has already been set up for data produced by the "tasks" where volume is much higher and "at least once" delivery is fine.

I'll take a closer look at the purge method-- do you think exposing it would be a doable contribution for a new contributor? That said, it might be that Kafka publishing isn't the right tool for the downstream execution request, and I should opt for HTTP or another protocol.

[rnpridgeon]

Yeah it shouldn't be too bad. Take look over some of the existing producer methods and how we bind to librdkafka and then feel free to ask any questions that you have here!

[peteryin21]

@rnpridgeon I've made a PR #548 to get the ball rolling on this. Since it's my first contribution in this library and first python-C binding I expect a decent amount of guidance- thank you in advance!