Manual Store Offsets & Process

[sruthidanda5a9]

Hello,

If I need to set the AutoStoreOffsets to false so I can store and process the offsets myself (on message handlers), whats the right way\place to implement it? I see the IOffsetManager and its implementation have been marked as internal. Does that mean the package consumers need to write their own offsetmanagers and use them directly on Messagehandlers? Does Kafka-flow provide any ability to consumer message handlers to process the offsets directly?

Thank you!

[filipeesch]

Hi @sruthidanda5a9 ,

You can use the MessageContext to store the offsets using the context.Consumer.StoreOffset() method.
Bare in mind that only in particular cases you need to manage the offset store manually, by default the framework stores the offsets when the message processing ends.

[ktb-dev]

If manual fixation of the offset is set in the settings, then a memory leak occurs!
My scenario requires manual fixation of the offset.

@filipeesch,
could you give recommendations about this, please?

public static IServiceCollection AddKafka(this IServiceCollection services, Action config)
{
return services.AddKafka(configuration =>
{
//...
configuration.AddCluster(cluster =>
{
//...
cluster.AddConsumer(consumer =>
{
//...
consumer.WithManualStoreOffsets();
consumer.AddMiddlewares(middlewares => middlewares
//...
.AddTypedHandlers(handlers => handlers
.WithHandlerLifetime(InstanceLifetime.Singleton)
.AddHandler()
)
);
});
});
});
}

public class MyHandler : IMessageHandler
{
int i = 0;
public async Task Handle(IMessageContext context, MyClass tag)
{
try
{
i++;
if (i > 5000)
{
context.ConsumerContext.StoreOffset();
i = 0;
}

        var tm = GC.GetTotalMemory(false) / 1024 / 1024;
        _logger.LogDebug("{0}mb, offset: - {1}", tm, context.ConsumerContext.Offset);
    }
    catch (Exception ex)
    {
    }
}

}

[dosper7]

Can you share more details that lead you to reach thar conclusion? Do you have a trace that tou can share?