[Bug Report]: Manually committing offsets not working as expected

[kevin-mcmanus]

Prerequisites

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

Description

I'm using version 4.0 and would like to control when offsets are written back to the broker.

As a simplified example I would like to process a batch of 1000 messages, and commit the offset of the last message in the batch.

These docs, describe how to turn off the automatic mode, but don't provide guidance on how to define which offset to commit.

I've experimented with various combinations of context.ConsumerContext.Complete() and context.ConsumerContext.ShouldStoreOffset = true but haven't found a combination that works. Please can you advise what I might be missing.

Steps to reproduce

Consumer config:

                    .AddConsumer(consumer => consumer
                        .Topic("sample-topic")
                        .WithName("input")
                        .WithGroupId("service1")
                        .WithAutoOffsetReset(KafkaFlow.AutoOffsetReset.Earliest)
                        .WithBufferSize(100)
                        .WithWorkersCount(1)
                        .WithManualMessageCompletion()
                        //.WithoutStoringOffsets()
                        //.WithAutoCommitIntervalMs(0)
                        .AddMiddlewares(middleware =>
                        {
                            middleware.Add<CommitMiddleware>();
                        })
                    )

Middleware

    internal class CommitMiddleware : IMessageMiddleware
    {
        public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
        {
            Console.WriteLine($"Processing offset {context.ConsumerContext.Offset}");

            await next(context);

            context.ConsumerContext.ShouldStoreOffset = false;

            if (context.ConsumerContext.Offset % 1000 == 0)
            {
                context.ConsumerContext.ShouldStoreOffset = true;

                context.ConsumerContext.Complete();
                Console.WriteLine($"Committing offset {context.ConsumerContext.Offset}");
            }

            //context.ConsumerContext.Complete();
        }
    }

Expected behavior

Every all to Complete() will update the offset number stored on the broker for the consumer group.

Actual behavior

A single offset is registered on the broker after the first call to Complete(). The consumer group continues to report the same offset number no matter how many times the middleware calls Complete() again.

KafkaFlow version

4.0.0

[vitorgoncalvesff]

Hi @kevin-mcmanus!

I believe the behaviour you’re seeing isn’t a bug, it comes from how KafkaFlow sequences message completion and offset storage.

Short Answer

You must still call Complete() on every message (even the ones you don’t want to commit) so the internal offset sequencing can advance. Use ShouldStoreOffset = true only on the messages whose offsets you want committed (e.g., every 1000th). Calling Complete() only on the 1000th message leaves the earlier ones “in flight,” so the commit logic never progresses past the first stored offset.

What Was Happening

Internally KafkaFlow does this per partition:

1. Every received message is enqueued.
2. When you call Complete():
   If ShouldStoreOffset == true, that offset becomes a candidate to commit.
   Whether or not it stores, the message is marked as processed and dequeued in order.
3. Only once all lower offsets are marked completed can a higher offset be stored/committed (this prevents gaps).

In your middleware you were:

• Setting ShouldStoreOffset = false for offsets 1..999
• Only calling Complete() for offset 1000
• Result: offsets 1..999 stayed "pending", offset 1000 could not move the high-water mark and only the very first offset ever appeared committed.

Your Two Options

Option A (Recommended for your case): Keep auto completion
Remove .WithManualMessageCompletion(), and only control ShouldStoreOffset:

internal class CommitMiddleware : IMessageMiddleware
{
    public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
    {
        await next(context);

        // Only persist the offset every 1000th message
        context.ConsumerContext.ShouldStoreOffset = context.ConsumerContext.Offset % 1000 == 0;

        if (context.ConsumerContext.ShouldStoreOffset)
        {
            Console.WriteLine($"Committing offset {context.ConsumerContext.Offset}");
        }
    }
}

KafkaFlow will still call Complete() automatically after your middleware finishes.

Option B: Keep manual completion
If you really need manual timing (e.g., multi-message transactional grouping):

internal class CommitMiddleware : IMessageMiddleware
{
    public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
    {
        // You're in manual mode, so KafkaFlow won't auto-complete
        await next(context);

        context.ConsumerContext.ShouldStoreOffset = context.ConsumerContext.Offset % 1000 == 0;

        if (context.ConsumerContext.ShouldStoreOffset)
        {
            Console.WriteLine($"Committing offset {context.ConsumerContext.Offset}");
        }

        // MUST always call this so later offsets can advance
        context.ConsumerContext.Complete();
    }
}

TL;DR

Not a bug. The fix is to always call Complete() for every message and toggle only ShouldStoreOffset.
Or let KafkaFlow handle completion automatically and just set ShouldStoreOffset on the commit intervals you want.

Suggested Documentation Clarification

We should add a brief paragraph to docs clarifying:

"Manual completion does not mean ‘optional completion’. You must still call Complete() for every message to advance internal offset sequencing, even if ShouldStoreOffset = false."

This would directly address the confusion the user hit.