Bug: Event ack'd after calling `KafkaMessage.nack` with `auto_commit=False`

[bradydean]

Describe the bug

Events are ack'd after calling nack().

How to reproduce

from faststream import FastStream, Logger
from faststream.kafka import KafkaMessage

broker = KafkaBroker()
app = FastStream(broker)

@broker.subscriber("test", group_id="async_subscriber", auto_commit=False)
async def async_subscriber(message: str, logger: Logger, msg: KafkaMessage):
    logger.info(message)
    await msg.nack()

Expected behavior

After publishing an event to the test topic, the message is logged and consumer does not ack the message.

Observed behavior
The message is logged and it appears the consumer acks the message. In kafka-ui I see the consumer group is 0 messages behind.

Environment

Running FastStream 0.2.15 with CPython 3.11.6 on Linux

[kumaranvpl]

Hello @bradydean,

Apologies for replying late.
Thanks for letting us know about the issue. we were working on releasing redis support so couldn't work on this issue. If you are interested in that feature, please take a look https://github.com/airtai/faststream/releases/tag/0.3.0rc0.
We will look at this issue and will fix it soon.

[Lancetnik]

Interesting... FastStream doen't call consumer.commit() in this case and passes enable_auto_commit=False to the original AiokafkaConsumer correctly. But, the message still acked...

[Lancetnik]

Seems like with aiokafka >= 0.9 all works fine. All tests at 0.3.5 versions are working correctly, so we can close it, I suppose

[lorenzobenvenuti]

Hi, I'm still experiencing this issue with Faststream 0.47.0.

from typing import Any, Dict
from faststream import FastStream
from faststream.kafka import KafkaBroker
from faststream.kafka.annotations import (
    KafkaMessage,
    Logger
)

broker = KafkaBroker("localhost:9093")
app = FastStream(broker)

@broker.subscriber("source", group_id="foo", auto_commit=False)
async def async_subscriber(message: Dict[str,Any], logger: Logger, msg: KafkaMessage):
    logger.info(message)
    await msg.nack() # calling this or NOT calling this doesn't change the behavior

If I send some messages with

for i in $(seq 1 20); do echo "${i}:{\"index\":\"${i}\"}" | kafka-console-producer.sh --bootstrap-server localhost:9093  --topic source --property parse.key=true --property key.separator=:; done

I see

2024-04-12 18:08:56,994 INFO     - source | foo | 110-171293 - Received
2024-04-12 18:08:56,995 INFO     - source | foo | 110-171293 - {'index': '1'}
2024-04-12 18:08:56,996 INFO     - source | foo | 110-171293 - Processed
2024-04-12 18:08:58,556 INFO     - source | foo | 111-171293 - Received
2024-04-12 18:08:58,557 INFO     - source | foo | 111-171293 - {'index': '2'}
2024-04-12 18:08:58,557 INFO     - source | foo | 111-171293 - Processed
2024-04-12 18:09:00,131 INFO     - source | foo | 112-171293 - Received
2024-04-12 18:09:00,132 INFO     - source | foo | 112-171293 - {'index': '3'}
2024-04-12 18:09:00,132 INFO     - source | foo | 112-171293 - Processed
2024-04-12 18:09:01,679 INFO     - source | foo | 113-171293 - Received
2024-04-12 18:09:01,680 INFO     - source | foo | 113-171293 - {'index': '4'}
2024-04-12 18:09:01,680 INFO     - source | foo | 113-171293 - Processed
2024-04-12 18:09:03,355 INFO     - source | foo | 114-171293 - Received
2024-04-12 18:09:03,356 INFO     - source | foo | 114-171293 - {'index': '5'}
2024-04-12 18:09:03,356 INFO     - source | foo | 114-171293 - Processed

I was expecting to consume the same message again and again.

Thanks,

lorenzo

(edit: same behavior if I raise a NackMessage exception)

[Lancetnik]

I am not sure, what is it: we creates AIOKafkaConsumer(..., enable_auto_commit=False) in such case and does not call consumer.commit() method at all.

I have no ideas why it commits the offset anyway.
@kumaranvpl have you any ideas?

[kumaranvpl]

@Lancetnik @lorenzobenvenuti
Right now I have no idea why this is happening. Let me try to replicate the same in local.
And then I will work on the fix.

[lorenzobenvenuti]

Thanks @Lancetnik @kumaranvpl . You can reproduce the issue with this application.