BatchConsumeMiddleware error handling

[sedjura]

Hi
Talking about https://github.com/Farfetch/kafkaflow/blob/master/src/KafkaFlow.BatchConsume/BatchConsumeMiddleware.cs

After batch processing we store all offsets from context. But we are storing these offsets even in case of failures in
await next(batchContext).ConfigureAwait(false); which can cause message loss.
Maybe it's a good idea to store offsets only in case of successful processing

try
{
    var batchContext = new BatchConsumeMessageContext(context.ConsumerContext, localBatch);
    await next(batchContext).ConfigureAwait(false);
}
catch (OperationCanceledException) when (context.ConsumerContext.WorkerStopped.IsCancellationRequested)
{
    return;
}
catch (Exception ex)
{
    this.logHandler.Error(
        "Error executing a message batch",
        ex,
        new
        {
            context.ConsumerContext.Topic,
            context.ConsumerContext.GroupId,
            context.ConsumerContext.WorkerId
        });
}
finally
{
    this.batch.Clear();
    this.dispatchSemaphore.Release();
}

foreach (var messageContext in localBatch) 
    messageContext.ConsumerContext.StoreOffset();

[gsferreira]

@filipeesch ☝️

[kjsulls]

@filipeesch any further thought if this is doable? We are also needing to not store/commit offsets if there were failures or a certain threshold is not successful.