[QUESTION] Retrying getmany batch if it fails before manual committing

[marcelolima]

Hey,

I'm trying to find the best way to retry a batch of messages when they fail to be processed (fail before commit() is called). The idea is to have something like:

while True:
  try:
      data = await consumer.getmany()
  
      for tp, messages in data.items():
          for message in messages:
              await handle_message(message)  # raises exception
  
      await consumer.commit()
  except Exception:
       print("failed, retrying...")
       await asyncio.sleep(0.5)

But this doesn't seem to work as expected, as the next call to getmany will continue to the next batch of messages instead of returning the failed one again. From what I could find in the codebase, this is due to the fetcher position counter that is updated on getmany calls, no matter if we later failed to commit the offset or not, so it's always increasing.

The best solution I could come up with so far is to call await consumer.seek_to_committed() when there is an error, so we take position back to the failing offset and the next call to getmany return the same batch (from the current stored offset). Would there be a better way to do that?

Thanks!

[ods]

If you need to retry just handling errors, the most simple choice is to do this inside your handle_message functions.

If not, then you may may track offset of successfully handled messages for each partition and commit them explicitly in case of error. Then either reset to committed of even recreate consumer completely.