EventProcessorClient and maintaining checkpoints in the Storage Account

[kobelev-dev]

Dear sirs,

Azure SDK contains abstract class Azure.Messaging.EventHubs.Primitives.CheckpointStore. This implies that users may want to create their own implementations. Great.

At the same time, from what I see, all constructors of the EventProcessorClient class use Azure.Storage.Blobs.BlobContainerClient as a parameter. This limits use of EventProcessorClient class only to the BlobCheckpointStore and not any other. I see that packages for other languages offer more, like Redis. So, my questions:

1. Am I missing something? Can it be that there is some trick that allows using EventProcessorClient with the custom check point sore? Note that I know about issue with multiple instances of EventProcessorClient and their need to have access to the same checkpoint store. My application for numerous reasons needs its own store. Is that possible?
2. Is there any other class that allows that? It may happen that I am looking at a wrong class. I am interested in .Net/C# solution.

[SimardeepSingh-zsh]

It sounds like you're looking to use a custom checkpoint store with the EventProcessorClient in .NET/C#. You're correct that the EventProcessorClient typically uses BlobCheckpointStore for checkpointing. However, you can indeed create your own implementation of the CheckpointStore class to use with EventProcessorClient.

Here's a high-level overview of how you can achieve this:

Implement the CheckpointStore Interface: Create a class that implements the CheckpointStore interface from the Azure.Messaging.EventHubs.Primitives namespace. This class will handle checkpointing and ownership data2.

Use PluggableCheckpointStoreEventProcessor: Instead of using the standard EventProcessorClient, use the PluggableCheckpointStoreEventProcessor. This class allows you to plug in your custom checkpoint store3.

Custom Checkpoint Store Logic: Implement the necessary methods (ClaimOwnershipAsync, GetCheckpointAsync, ListOwnershipAsync, UpdateCheckpointAsync) in your custom checkpoint store class.

Here's a simplified example: [csharp]

using Azure.Messaging.EventHubs.Primitives;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

public class CustomCheckpointStore : CheckpointStore
{
public override async Task GetCheckpointAsync(
string namespaceName,
string eventHubName,
string consumerGroup,
string partitionId,
CancellationToken cancellationToken)
{
// Implement your logic to retrieve checkpoint data
}

public override async Task<IEnumerable<EventProcessorPartitionOwnership>> ListOwnershipAsync(
    string namespaceName,
    string eventHubName,
    string consumerGroup,
    CancellationToken cancellationToken)
{
    // Implement your logic to list ownership data
}

public override async Task<bool> UpdateCheckpointAsync(
    string namespaceName,
    string eventHubName,
    string consumerGroup,
    string partitionId,
    long offset,
    long sequenceNumber,
    CancellationToken cancellationToken)
{
    // Implement your logic to update checkpoint data
}

public override async Task<bool> ClaimOwnershipAsync(
    IEnumerable<EventProcessorPartitionOwnership> partitionOwnership,
    CancellationToken cancellationToken)
{
    // Implement your logic to claim ownership
}

}

// Usage with PluggableCheckpointStoreEventProcessor
var checkpointStore = new CustomCheckpointStore();
var eventProcessor = new PluggableCheckpointStoreEventProcessor(checkpointStore);
await eventProcessor.StartProcessingAsync("namespaceName", "eventHubName", "consumerGroup");

This should give you the flexibility to use your own checkpoint store with the EventProcessorClient