[Bug Report]: ConsumerBatch do store offsets when exception next middleware

[vkuzugashev]

Prerequisites

• I have searched issues to ensure it has not already been reported

Description

ConsumerBatch do store offsets when exception next middleware.
I,m thing this bag. In my case messages consumed into batch. If batch stored to database success on next middleware, then messages offset must stored to kafka. If batch store on next middleware with exception, then all message offset must not store to kafka.

Now, when batch stored on next middleware with exception, then all messages offset store to kafka:

BatchConsumeMiddleware.cs line 106.
foreach (var messageContext in localBatch) { messageContext.ConsumerContext.StoreOffset(); }

Steps to reproduce

1. Add ConsumeBatch
2. Add next middleware with throw exception

Expected behavior

If get exception on next middleware, then all message offsets, not be stored to kafka

Actual behavior

get exception on next middleware and all messages offsets will stored to kafka

KafkaFlow version

no

[JoaoRodriguesGithub]

Hi @vkuzugashev,

When using BatchConsume, it is expected that KafkaFlow framework will commit all messages, including the ones with exceptions that were not treated when processed.

The framework client should be the one to decide how those exceptions should be treated, and to implement the resilience mechanisms (retry policies, DLQ, etc.)

The reasoning for this is that even if the offsets are not committed, there are no guaranty that the message will be processed again, unless a rebalance occurs and no other offsets have been committed.

Do you have any specific use case where this is not compatible?

[JoaoRodriguesGithub]

Hi @vkuzugashev,

If you're still experiencing this issue or have any additional information to share, please feel free to let us know.

However, if we don't receive any updates or feedback from you within the next 5 business days, we may need to consider closing this issue. Please understand that this is not a final decision, and you can always reopen the issue or create a new one in the future.

We appreciate your contribution to our project and look forward to hearing from you soon. If you have any questions or need further assistance, don't hesitate to reach out.

[JoaoRodriguesGithub]

Hi @vkuzugashev ,

We will be closing this issue since the question has been answered. Please feel free to respond to this message or reopen the issue if you'd like to continue the discussion or if you've encountered any new developments related to it.

Thank you for your contribution.