Azure Event Hubs Checkpoint Store Blob

Warnings

• breaking Version 1.0.0b6 introduced breaking changes: `BlobPartitionManager` was renamed to `BlobCheckpointStore`, and its constructor was updated to accept storage container details directly instead of a `ContainerClient` instance. The `blobstoragepm` module became internal.
  Fix: Update imports to `from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore` and adjust constructor calls.
• gotcha For optimal performance and reliability, it is strongly recommended to use a dedicated Blob Storage container for each consumer group. The storage account should be in the same region as the Event Hub consumer application, and it should not be used for other workloads. Additionally, disable 'Hierarchical namespace', 'Blob soft delete', and 'Versioning' for the container.
• gotcha In environments like Azure Stack Hub, which may support older Azure Storage Service API versions, you might need to explicitly specify `api_version` (e.g., `api_version='2017-11-09'`) when creating `BlobCheckpointStore` to avoid `HttpResponseError` exceptions. The default API version is '2019-07-07'.
• deprecated Support for Python 2.7 has officially ended on January 1, 2022. All versions of this library from 1.2.0 onwards require Python 3.8 or later. Earlier versions had varying Python 3.x support.
  Fix: Upgrade to Python 3.8 or later and use a compatible version of the library.
• gotcha Azure Event Hubs client libraries are transitioning towards exclusively using sequence number-based checkpoints. Relying on offsets for external systems like KEDA scalers can be problematic because offsets do not change predictably. While the library contract might remain stable, the underlying behavior for external monitoring or scaling based on checkpoint data could change.
  Fix: When building external systems that rely on checkpoint data, favor sequence numbers over offsets for reliable scaling and monitoring.

Install

• pip install azure-eventhub-checkpointstoreblob Install stable version

Imports

• BlobCheckpointStore

  from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore

  The `BlobPartitionManager` class and `blobstoragepm` module were renamed to `BlobCheckpointStore` in `azure.eventhub.extensions.checkpointstoreblob` in version 1.0.0b6. The old module is now internal.

Quickstart

This quickstart demonstrates how to consume events from an Azure Event Hub using `EventHubConsumerClient` and persist checkpoints in Azure Blob Storage using `BlobCheckpointStore`. It authenticates using `DefaultAzureCredential` and processes events in a simple `on_event` callback, updating the checkpoint after each event. Ensure environment variables for Event Hubs connection details and Blob Storage account URL/container name are set.

import os
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
from azure.identity import DefaultAzureCredential

# Environment variables for Event Hubs
EVENTHUB_CONNECTION_STR = os.environ.get('EVENTHUB_CONNECTION_STR', '')
EVENTHUB_NAME = os.environ.get('EVENTHUB_NAME', '')
CONSUMER_GROUP = os.environ.get('CONSUMER_GROUP', '$Default')

# Environment variables for Azure Blob Storage Checkpoint Store
BLOB_STORAGE_ACCOUNT_URL = os.environ.get('BLOB_STORAGE_ACCOUNT_URL', '') # e.g., 'https://<your_storage_account_name>.blob.core.windows.net/'
BLOB_CONTAINER_NAME = os.environ.get('BLOB_CONTAINER_NAME', '')

# DefaultAzureCredential will attempt to authenticate using environment variables,
# managed identity, Azure CLI, etc.
credential = DefaultAzureCredential()

def on_event(partition_context, event):
    print(f"Received event from partition {partition_context.partition_id}, sequence_number: {event.sequence_number}")
    # Your event processing logic here

    # Update the checkpoint so that the program doesn't read the events
    # that it has already read when you run it next time.
    partition_context.update_checkpoint(event)

def main():
    if not all([EVENTHUB_CONNECTION_STR, EVENTHUB_NAME, BLOB_STORAGE_ACCOUNT_URL, BLOB_CONTAINER_NAME]):
        print("Please set EVENTHUB_CONNECTION_STR, EVENTHUB_NAME, BLOB_STORAGE_ACCOUNT_URL, and BLOB_CONTAINER_NAME environment variables.")
        return

    # Create an Azure blob checkpoint store to store the checkpoints.
    checkpoint_store = BlobCheckpointStore(
        blob_account_url=BLOB_STORAGE_ACCOUNT_URL,
        container_name=BLOB_CONTAINER_NAME,
        credential=credential,
    )

    # Create a consumer client for the event hub.
    client = EventHubConsumerClient.from_connection_string(
        conn_str=EVENTHUB_CONNECTION_STR,
        consumer_group=CONSUMER_GROUP,
        eventhub_name=EVENTHUB_NAME,
        checkpoint_store=checkpoint_store,
    )

    with client:
        print(f"Listening for events in consumer group '{CONSUMER_GROUP}' on Event Hub '{EVENTHUB_NAME}'...")
        # Call the receive method. Read from the beginning of the partition
        client.receive(on_event=on_event, starting_position="-1")

    credential.close()

if __name__ == "__main__":
    main()