Give the client a chance to save offset on graceful shutdown

[ppou-novibet]

I'm using a stream with IsSingleActiveConsumer=true and I am tracking the offset as messages are being consumed, periodically saving the offset by calling consumer.StoreOffset as described in the documentation.

However I haven't found a clean way of saving the last consumed offset just before shutting down my application.

One nice way of doing that is if I could provide a callback to Consumer.Create that would be called just before the consumer is closed. The callback would receive the RawConsumer and it would be able to call consumer.StoreOffset to save last offset to stream.

[Gsantomaggio]

Hi @ppou-novibet,
Thank you for using the client. I will investigate what the best way to do that would be.

[jonnepmyra]

Hi @ppou-novibet

We have the very same requirment and resolved it by registering a StoreOffset delegate in

IHostApplicationLifetime.ApplicationStopping.Register(() =>{
   consumer.StoreOffset(); //and continue closing our consumer
});

[Gsantomaggio]

@ppou-novibet have you tried StatusChanged event? https://github.com/rabbitmq/rabbitmq-stream-dotnet-client/blob/main/docs/ReliableClient/BestPracticesClient.cs#L265

[ppou-novibet]

@Gsantomaggio sorry just noticed this comment. Currently offsets are stored by calling RawConsumer.StoreOffset inside the MessageHandler with the reference to RawConsumer provided by the library as a parameter to the handler. StatusChanged handler does not provide a reference to RawConsumer and in the BestPracticesClient source there's a comment in bold capital letters to avoid blocking inside the handler! I also imagine that you would have to do this before the underlying _client is closed, so a fire-and-forget strategy seems risky.

So I'm not sure how I would go about doing this inside the StatusChanged handler.

[Gsantomaggio]

@ppou-novibet something like:

 public static async Task Start()
    {
        Console.WriteLine("Start");
        var streamSystem = await StreamSystem.Create(new StreamSystemConfig());
        const string StreamName = "my_stream";
        const string Reference = "my_reference";

        await streamSystem.CreateStream(new StreamSpec(StreamName));
        IOffsetType offsetRestart = new OffsetTypeFirst();
        ulong consumerOffset = 0;
        var offFromServer = await streamSystem.TryQueryOffset(Reference, StreamName)
            .ConfigureAwait(false);

        if (offFromServer != null)
        {
            Console.WriteLine($"Offset found in server: {offFromServer.Value}. Restarting from that offset");
            offsetRestart = new OffsetTypeOffset(offFromServer.Value);
        }

        var consumerConfiguration = new ConsumerConfig(streamSystem, StreamName)
        {
            OffsetSpec = offsetRestart,
            Reference = Reference,
            MessageHandler = async (stream, consumerSource, context, message) =>
            {
                consumerOffset = context.Offset;
                Console.WriteLine($"offset: {context.Offset}");
                // store the offset in the server based on your logic
                // await consumerSource.StoreOffset(context.Offset);

                await Task.CompletedTask;
            }
        };

        consumerConfiguration.StatusChanged += status =>
        {
            if (status.To != ReliableEntityStatus.Closed) return;
            Console.WriteLine($"Consumer closed, storing offset to {consumerOffset}");
            streamSystem.StoreOffset(Reference, StreamName, consumerOffset)
                .ConfigureAwait(false); // the consumer is closedand stores the offset.
        };

        var consumer = await Consumer.Create(consumerConfiguration);

        Console.WriteLine($"Consumer started {consumer.Info}");

        // press any key to stop the consumer
        Console.ReadKey();
        await consumer.Close();
    }

[ppou-novibet]

Thanks @Gsantomaggio I missed streamSystem.StoreOffset(). It's still a fire-and-forget though so the StoreOffset could execute after the StatusChanged handler returns (and possibly after the stream has been closed). Do you think that could be an issue at runtime?

[Gsantomaggio]

@ppou-novibet, the stream is never closed; maybe the consumer is closed.
The streamSystem.StoreOffset() uses another connection.

In general, by the way, if you need to track each single message, you should use local storage to store the offset.
It is best practice to implement the system with idempotency so if you skip some store offset it won't be a problem.