Microsoft Azure Event Hubs Checkpoint Store for Blob Storage (Async)

Warnings

• breaking The `BlobPartitionManager` class was renamed to `BlobCheckpointStore` in version 1.0.0b6. The constructor signature also changed, now taking storage container details directly instead of a `ContainerClient` instance.
  Fix: Update class name to `BlobCheckpointStore` and adjust constructor parameters. Use `BlobCheckpointStore.from_connection_string` for connection string-based initialization or pass `blob_account_url`, `container_name`, and `credential`.
• breaking As of version 1.2.0, support for Python 2.7, 3.6, and 3.7 has been dropped. The library now requires Python 3.8 or later.
  Fix: Upgrade your Python environment to version 3.8 or newer.
• gotcha This is an asynchronous (aio) library. All operations and client instantiation require `async` and `await` keywords within an `asyncio` event loop.
  Fix: Ensure your application structure is asynchronous, utilizing `async def` functions and `await` calls for library methods.
• gotcha When using Azure Blob Storage for checkpointing, it's recommended to use a separate container for each consumer group, locate the storage account in the same region as the deployed application, and disable Hierarchical Namespace, Blob Soft Delete, and Versioning on the storage account for optimal performance and to prevent issues.
  Fix: Follow Azure's best practices for Blob Storage configuration when used as a checkpoint store, as detailed in the documentation.
• gotcha If deploying on Azure Stack Hub or environments using older Azure Storage Service APIs, you may need to explicitly specify the `api_version` in the `BlobCheckpointStore` constructor (e.g., `api_version='2017-11-09'`) to avoid `HttpResponseError` due to incorrect header values.
  Fix: Determine the supported Storage Service API version for your environment and pass it as the `api_version` keyword argument during `BlobCheckpointStore` instantiation.
• gotcha While connection strings can be used for authentication, for production applications, Azure recommends passwordless authentication using Azure Active Directory and the `azure-identity` library. Ensure the principal has 'Azure Event Hubs Data Owner' and 'Storage Blob Data Contributor' roles.
  Fix: Migrate to passwordless authentication using `DefaultAzureCredential` or other `azure-identity` credentials. Assign the necessary Azure RBAC roles to your application's identity.

Install

• pip install azure-eventhub-checkpointstoreblob-aio Install stable version

Imports

• BlobCheckpointStore

  from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore

  The class was renamed from `BlobPartitionManager` to `BlobCheckpointStore` and the internal module `blobstoragepmaio` should not be imported directly since version 1.0.0b6.

Quickstart

This quickstart demonstrates how to set up an `EventHubConsumerClient` to receive events from Azure Event Hubs, using `BlobCheckpointStore` for managing checkpoints and partition ownership. It uses `DefaultAzureCredential` from `azure-identity` for passwordless authentication, which is recommended for production scenarios. Ensure the necessary environment variables for your Event Hubs namespace, Event Hub name, Blob Storage account URL, and container name are set, and the consuming identity has appropriate 'Azure Event Hubs Data Owner' and 'Storage Blob Data Contributor' roles.

import asyncio
import os
from azure.eventhub.aio import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore
from azure.identity.aio import DefaultAzureCredential

# Environment variables for Event Hubs
FULLY_QUALIFIED_NAMESPACE = os.environ.get('EVENT_HUB_FULLY_QUALIFIED_NAMESPACE', 'your_namespace.servicebus.windows.net')
EVENT_HUB_NAME = os.environ.get('EVENT_HUB_NAME', 'your_event_hub_name')
CONSUMER_GROUP = os.environ.get('EVENT_HUB_CONSUMER_GROUP', '$Default')

# Environment variables for Blob Storage
BLOB_STORAGE_ACCOUNT_URL = os.environ.get('AZURE_STORAGE_BLOB_URL', 'https://yourstorageaccount.blob.core.windows.net')
BLOB_CONTAINER_NAME = os.environ.get('BLOB_CONTAINER_NAME', 'your_container_name')

async def on_event(partition_context, event):
    # Process the event here
    if event.body_as_str():
        print(f"Received event from partition ID: {partition_context.partition_id}, Body: {event.body_as_str()}")
    # Update the checkpoint to mark the event as processed
    await partition_context.update_checkpoint(event)

async def main():
    credential = None
    checkpoint_store = None
    client = None
    try:
        # Authenticate using DefaultAzureCredential (Managed Identity, Environment variables, etc.)
        credential = DefaultAzureCredential()

        # 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(
            fully_qualified_namespace=FULLY_QUALIFIED_NAMESPACE,
            eventhub_name=EVENT_HUB_NAME,
            consumer_group=CONSUMER_GROUP,
            checkpoint_store=checkpoint_store,
            credential=credential,
        )

        async with client:
            # Start receiving events
            print(f"Listening for events in consumer group: {CONSUMER_GROUP} from Event Hub: {EVENT_HUB_NAME}")
            await client.receive(on_event=on_event, starting_position="-1") # Start from beginning if no checkpoint

    except Exception as e:
        print(f"An error occurred: {e}")
    finally:
        if client:
            await client.close()
        if checkpoint_store:
            await checkpoint_store.close()
        if credential:
            await credential.close()

if __name__ == "__main__":
    asyncio.run(main())