Access Kafka consumer

[lorenzobenvenuti]

Hi,

is there any way we can access the Kafka consumer backing an handler? I see KafkaMessage has a consumer fields but it's a ConsumerProtocol object that only exposes a commit method. My use case is the following: a consumer needs to process a message and then interact with some external systems; in case one of the dependencies is down I want to:

• Suspend the consumer (using AIOKafkaConsumer.pause, because the consumer still needs to invoke poll or the broker will kick it out of the consumer group and trigger a rebalance; if I raise a StopConsumer exception I understand the consumer will stop completely)

• Wait for a while

• Resume the consumer and try to process the message again

Something like

@broker.subscriber("input-topic", auto_commit=False)
def handle(body : Foo, msg : KafkaMessage):
    try:
        send_to_external_system(body)
        msg.ack()
    except ExternalSystemNotAvailableError:
        # message can be retried
        msg.nack()
        get_consumer().pause()
        wait_for_five_minutes_and_resume() # or ping the external dependency until it's up again and then resume
    except Exception as err:
        # message failed for other reasons
        handle_error(err) # do something else: log, send to DLQ, ..
        msg.ack()

(or use a decorator)

Right now (*) the only thing we can do is retrying the message without pausing the consumer but this means that either we reprocess a message continuosly or, if we add a sleep in the handler, we need to increase max.poll.interval.ms.

Thanks,

lorenzo

(*) Actually, with version 0.47.0 nack-ing a message still results in the offset being committed, see my comment here

[Lancetnik]

Indeed, KafkaMessage requires ConsumerProtocol, but in fact it is a AIOKafkaConsumer object, so you can use it the way you want

[lorenzobenvenuti]

Hi @Lancetnik , thanks for the reply! Yeah that will work, although it looks a bit hacky: if the underlying implementation changes (for example, I see that the Confluent implementation doesn't return confluent_kafka.Consumer directly but it wraps it with a AsyncConfluentConsumer object) the code could break. I'm wondering exposing pause and resume in ConsumerProtocol could be useful for other people as well; I did a quick and dirty POC here.
If you think it makes sense I could try to put a PR together, otherwise I'll stick with casting msg.consumer to AIOKafkaConsumer.

Thanks,

lorenzo

[Lancetnik]

I don't like an idea to expand this protocol cuz it has the only one method required by the message. Anyway, I can guarantee you, that message will always has AioKafkaConsumer object in this place. Also I prefer to fix current behavior instead of legalize such hack and I can promise to you t make it.

[lorenzobenvenuti]

Ack, tyvm!