EventProcessorHost is check pointing after every batch

[gambrose]

Actual Behaviour

1. After a call to ProcessEventsAsync PartitionPump is calling PartitionContext.SetOffsetAndSequenceNumber(last)
2. I read off the messages and batch them then to send them on to a another system.
3. My app restarts and I have lost data because checkpoint has been set even though my code never called context.CheckpointAsync(eventData) to mark the batch as being processed.

Expected Behaviour

1. Check-pointing to only occur when the client calls CheckpointAsync on the PartitionContext

Versions

• OS platform and version: Windows 10
• .NET Version: net452
• NuGet package version or commit ID: Microsoft.Azure.EventHubs.Processor version="1.0.0"

[gambrose]

Creating a simple IEventProcessor that check points each message throws ArgumentOutOfRangeException "New offset 27984/2 is less than previous 147264/8" which seems to suggest that the messages are being check-pointed before have even been consumed?

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Microsoft.Azure.EventHubs;
using Microsoft.Azure.EventHubs.Processor;

class CheckpointAllTheThings : IEventProcessor
{
    Task IEventProcessor.CloseAsync(PartitionContext context, CloseReason reason)
    {
        Console.WriteLine("Processor Shutting Down. Partition '{0}', Reason: '{1}'.", context.PartitionId, reason);
        return Task.FromResult<object>(null);
    }

    Task IEventProcessor.OpenAsync(PartitionContext context)
    {
        Console.WriteLine("SimpleEventProcessor initialized.  Partition: '{0}'", context.PartitionId);
        return Task.FromResult<object>(null);
    }

    async Task IEventProcessor.ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
    {
        foreach (EventData eventData in messages)
        {
            await context.CheckpointAsync(eventData);
        }
    }

    Task IEventProcessor.ProcessErrorAsync(PartitionContext context, Exception error)
    {
        Console.WriteLine($"Error: {error}");

        return Task.FromResult<object>(null);
    }
}

[serkantkaraca]

PartitionContext.SetOffsetAndSequenceNumber(last) isn't checkpointing. You need to call context.CheckpointAsync(eventData) while processing the events. I don't really recommend checkpointing every message though. This will slowdown your pipe really bad.

[gambrose]

@serkantkaraca The checkpoint every message was an attempt at a simplified example that context.CheckpointAsync(eventData) is broken. As far as I understand this scenario should work, else why have the method?

My actual code only calls checkpoint when it has batched enough messages from ProcessEventsAsync but at the moment the checkpoint is being called more frequently, at least every call to ProcessEventsAsync from what I can see in PartitionPump although I do not understand what is causing the second checkpoint every message to break.

[serkantkaraca]

How many hosts are you running with? Can you try with a single host? Let it process the messages while checkpointing. Then restart the host and it should resume from the last checkpoint. Btw, we have unit tests covering these type of scenarios. You can also review them and compare with your code. The unit tests are all passing.

[gambrose]

I am running on a single host. I will have a look at the unit tests.

[gambrose]

@serkantkaraca I've done some more testing and looked at the unit tests.

My original assertion, that I was loosing data, was wrong. The app restarts from the checkpoint that was saved to storage. I was confused by the error message when calling checkpoint that said "New offset 27984/2 is less than previous 147264/8" I wrongly assumed that the previous value reported was the checkpoint saved to storage and so assumed that I was loosing data.

Testing using my CheckpointAllTheThings event processor repo against a hub will events in. I found that the first checkpoint call always worked. Then the second always failed. The first call had only one event passed to ProcessEventsAsync where the second has multiple events. If I set MaxBatchSize = 1 on EventProcessorOptions then It can run though.
So the issue seems to be related to calling CheckpointAsync with messages that are not the last in the batch.

Looking at your tests I do not think this scenario is covered. You only ever checkpoint the last message and also only ever send a max of 1 message.
I was finding you test fixture really hard to use. The pattern of using events with Tubules for args makes the code very hard to understand. So I have given up trying to create a new test that covers the scenario using you test fixture.

I will probably revert to the .net ServiceBus library as I need to move forward. Thanks for your time and I hope you will be able to reproduce this issue in your tests.

[serkantkaraca]

Ok, I can now understand what you are doing different. You are check-pointing the first message in the batch. Frankly, I have never thought someone would do this intentionally. I will check the code and see if there is something we can improve there. In the meantime, can you consider check-pointing the last message? Will this hold with your app's requirements?

[serkantkaraca]

Thx for reporting this. While reviewing the code for "check-point first message of the batch" approach, I have discovered a bug. Again, honestly I didn't think of check-pointing first message approach and never tested that. I will send a PR for this fix shortly.
I sill recommend to check-point the last message you processed.

[gambrose]

Thanks @serkantkaraca.
To be clear I am not taking a checkpoint the first message approach. I am taking an approach where the downstream system to event hub needs to dictate the batch size.
I.e. Checkpoint point every 100 messages for example. As I don't control the batch size sent to process events the 100th message may be the first, last or anywhere in between. The point is I need to be able to checkpoint any message in the event hub batch.

I hope this sheads some light on my scenario.

[serkantkaraca]

No worries I understand that. Any eventdata other than the last one will fail while check-pointing. I will fix the issue.

[serkantkaraca]

Fix merged into dev branch