Retry the failed message (retry consuming)

[MaximMukhametov]

I'm not sure that the question is at the right place, but maybe there are ready-made tools in the library to solve this case.
The question is - how to make a retry mechanism if the logic that is launched by the consumer falls, and we need to try to restart this logic until it's processed successfully?

[MaximMukhametov]

One more question - is there way to commit offset manually after successful processing? If so, at what moment will the consumer be forced to read the message again?

[sternakt]

Not at the moment, no
The offset is committed when the event is read from the queue.
We do have plans to add transactional logic, so that the offset is committed at the end of your processing function (#15)

[sternakt]

I'm not sure that the question is at the right place, but maybe there are ready-made tools in the library to solve this case. The question is - how to make a retry mechanism if the logic that is launched by the consumer falls, and we need to try to restart this logic until it's processed successfully?

When you consume a message from a topic your decorated consume function will be called with the consumed message.
The processing logic you write inside your consuming function is entirely inside your control, so you can write the retry mechanism for the logic as you would in any other python function.

app = FastKafka(...)

@app.consumes()
async def on_messages(msg: MyMsg):
    "your processing logic with retries"

The only caveat of this, and why we plan to implement transactional logic, is that if your app crashes or is shut down while retrying, the message offset will be commited already and the message will count as "consumed"

Sorry, for the late answer and please let us know if this was helpful