Receiving Events

This section details how to receive events from Azure Event Hubs. Understanding the different consumer groups and checkpointing mechanisms is crucial for reliable event processing.

Consumer Groups

Event Hubs use consumer groups to allow multiple applications to independently read from the same event stream. Each consumer group maintains its own position in the event stream.

Default Consumer Group: Every Event Hub has a default consumer group ($Default) that applications can use.
Custom Consumer Groups: You can create your own consumer groups to isolate the reading of events for different applications or microservices. This is recommended to avoid conflicts.

Receiving with Azure SDKs

The Azure SDKs provide powerful abstractions for receiving events. The EventProcessorClient is a key component for managing the complexities of reading from Event Hubs, including load balancing and checkpointing.

Example: C# EventProcessorClient

Here's a basic example of how to set up an EventProcessorClient in C#.


using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Processor;
using Azure.Storage.Blobs;
using System;
using System.Text;
using System.Threading.Tasks;

// ...

string eventHubsConnectionString = "YOUR_EVENTHUBS_CONNECTION_STRING";
string consumerGroup = "$Default"; // Or your custom consumer group name
string blobStorageConnectionString = "YOUR_BLOB_STORAGE_CONNECTION_STRING";
string blobContainerName = "eventhub-checkpoints";

BlobServiceClient blobServiceClient = new BlobServiceClient(blobStorageConnectionString);
BlobContainerClient blobContainerClient = blobServiceClient.GetBlobContainerClient(blobContainerName);

EventProcessorClient processorClient = new EventProcessorClient(
    blobContainerClient,
    consumerGroup,
    eventHubsConnectionString);

processorClient.ProcessEventAsync += HandleEventAsync;
processorClient.ProcessErrorAsync += HandleErrorAsync;

await processorClient.StartProcessingAsync();

// Keep the application running to process events
Console.WriteLine("Starting to process events. Press Enter to stop.");
Console.ReadLine();

await processorClient.StopProcessingAsync();

async Task HandleEventAsync(ProcessEventArgs args)
{
    Console.WriteLine($"Received event: {Encoding.UTF8.GetString(args.Data.EventBody.ToArray())} from partition {args.Partition.Id}");

    // You can update checkpoints here if needed
    await args.UpdateCheckpointAsync();
}

async Task HandleErrorAsync(ProcessErrorEventArgs args)
{
    Console.WriteLine($"Error processing event: {args.Exception.Message} from partition {args.PartitionId}");
}

Checkpointing

Checkpointing is the process by which a consumer group tracks its progress through the Event Hub partition. When an application restarts, it can resume reading from the last recorded checkpoint, ensuring that no events are missed or reprocessed unnecessarily.

Important: For reliable event processing, it's highly recommended to use a persistent storage solution like Azure Blob Storage for checkpointing. The EventProcessorClient integrates seamlessly with Blob Storage.

Partition Management

Event Hubs distribute events across multiple partitions. The EventProcessorClient automatically handles load balancing across partitions, distributing the workload among active instances of your consumer. If an instance fails, its partitions are reassigned to other active instances.

Key Considerations for Receiving Events

Choose the appropriate consumer group for your application.
Implement robust error handling and logging.
Configure checkpointing to ensure fault tolerance and exactly-once processing semantics where applicable.
Monitor your consumer applications for performance and errors.

For more advanced scenarios, consider libraries like Azure Functions with Event Hubs Trigger or Azure Stream Analytics which offer managed event processing capabilities.

Next: Processing Model