From 193004ea55ce55a796bb4b7a00296319e349dc2f Mon Sep 17 00:00:00 2001 From: Swathi Pillalamarri Date: Wed, 30 Dec 2020 12:02:03 -0500 Subject: [PATCH 1/7] updated event hubs migration guide to follow template --- .../azure-eventhub/migration_guide.md | 366 +++++++----------- 1 file changed, 139 insertions(+), 227 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/migration_guide.md b/sdk/eventhub/azure-eventhub/migration_guide.md index a4fb99d8a07f..276ecbb9a463 100644 --- a/sdk/eventhub/azure-eventhub/migration_guide.md +++ b/sdk/eventhub/azure-eventhub/migration_guide.md @@ -1,266 +1,178 @@ -# Guide to migrate from azure-eventhub v1 to v5 +# Guide for migrating azure-eventhub to v5 from v1 -This document is intended for users that are familiar with V1 of the Python SDK for Event Hubs library (`azure-eventhub 1.x.x`) and wish -to migrate their application to V5 of the same library. +This guide is intended to assist in the migration to `azure-eventhub` v5 from v1. It will focus on side-by-side comparisons for similar operations between the two packages. -For users new to the Python SDK for Event Hubs, please see the [readme file for the azure-eventhub](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhub/README.md). +Familiarity with the `azure-eventhub` v1 package is assumed. For those new to the Event Hubs client library for Python, please refer to the [README for `azure-eventhub`](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhub/README.md) rather than this guide. -## General changes -Version 5 of the azure-eventhub package is the result of our efforts to create a client library that is user-friendly and idiomatic to the Python ecosystem. -Alongside an API redesign driven by the new [Azure SDK Design Guidelines for Python](https://azure.github.io/azure-sdk/python_introduction.html#design-principles), -the latest version improves on several areas from V1. +## Table of contents -### Specific clients for sending and receiving -In V5 we've simplified the API surface, making two distinct clients, rather than having a single `EventHubClient`: -* `EventHubProducerClient` for sending messages. [Sync API](https://azuresdkdocs.blob.core.windows.net/$web/python/azure-eventhub/latest/azure.eventhub.html#azure.eventhub.EventHubProducerClient) -and [Async API](https://azuresdkdocs.blob.core.windows.net/$web/python/azure-eventhub/latest/azure.eventhub.aio.html#azure.eventhub.aio.EventHubProducerClient) -* `EventHubConsumerClient` for receiving messages. [Sync API](https://azuresdkdocs.blob.core.windows.net/$web/python/azure-eventhub/latest/azure.eventhub.html#azure.eventhub.EventHubConsumerClient) -and [Async API](https://azuresdkdocs.blob.core.windows.net/$web/python/azure-eventhub/latest/azure.eventhub.aio.html#azure.eventhub.aio.EventHubConsumerClient) +* [Migration benefits](#migration-benefits) + - [Cross Service SDK improvements](#cross-service-sdk-improvements) +* [Important changes](#important-changes) + - [Client hierarchy](#client-hierarchy) + - [Client constructors](#client-constructors) + - [Sending](#sending-events) + - [Receiving](#receiving-events) + - [Receiving with checkpoints](#receiving-with-checkpoints) +* [Additional samples](#additional-samples) -We've also merged the functionality from `EventProcessorHost` into -`EventHubConsumerClient`, allowing `EventHubConsumerClient` to be the single -point of entry for receiving of any type (from single partition, all partitions, or with load balancing and checkpointing features) within Event Hubs. +## Migration benefits -V5 has both sync and async APIs. Sync API is under package `azure.eventhub` whereas async API is under package `azure.eventhub.aio`. -They have the same class names under the two packages. For instance, class `EventHubConsumerClient` with sync API under package `azure.eventhub` has its -async counterpart under package `auzre.eventhub.aio`. -The code samples in this migration guide use async APIs. +A natural question to ask when considering whether or not to adopt a new version or library is what the benefits of doing so would be. As Azure has matured and been embraced by a more diverse group of developers, we have been focused on learning the patterns and practices to best support developer productivity and to understand the gaps that the Python client libraries have. -### Client constructors +There were several areas of consistent feedback expressed across the Azure client library ecosystem. One of the most important is that the client libraries for different Azure services have not had a consistent approach to organization, naming, and API structure. Additionally, many developers have felt that the learning curve was difficult, and the APIs did not offer a good, approachable, and consistent onboarding story for those learning Azure or exploring a specific Azure service. -| In v1 | Equivalent in v5 | Sample | -|---|---|---| -| `EventHubClientAsync()` | `EventHubProducerClient()` or `EventHubConsumerClient()` | [using credential](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhub/samples/async_samples/client_identity_authentication_async.py) | -| `EventHubClientAsync.from_connection_string()` | `EventHubProducerClient.from_connection_string` or `EventHubConsumerClient.from_connection_string` |[client creation](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhub/samples/async_samples/client_creation_async.py) | -| `EventProcessorHost()`| `EventHubConsumerClient(..., checkpoint_store)`| [receive events using checkpoint store](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhub/samples/async_samples/recv_with_checkpoint_store_async.py) | +To try and improve the development experience across Azure services, a set of uniform [design guidelines](https://azure.github.io/azure-sdk/general_introduction.html) was created for all languages to drive a consistent experience with established API patterns for all services. A set of [Python-specific guidelines](https://azure.github.io/azure-sdk/python_introduction.html) was also introduced to ensure that Python clients have a natural and idiomatic feel with respect to the Python ecosystem. Further details are available in the guidelines for those interested. -In V5, the SDK provides `BlobCheckpointStore` in extension packages azure-eventhub-checkpointstoreblob (for sync) and azure-eventhub-checkpointstoreblob-aio (for async). -You can define your own `CheckpointStore` class to persist checkpoint data. +### Cross Service SDK improvements -### Receiving events +The modern Event Hubs client library also provides the ability to share in some of the cross-service improvements made to the Azure development experience, such as +- using the new [`azure-identity`](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/identity/azure-identity/README.md) library to share a single authentication approach between clients +- a unified logging and diagnostics pipeline offering a common view of the activities across each of the client libraries -| In v1 | Equivalent in v5 | Sample | -|---|---|---| -| `EventHubClientAsync.add_async_receiver()` and `AsyncReceiver.receive()`| `EventHubConsumerClient.receive()`| [receive events](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhub/samples/async_samples/recv_async.py) | +## Important changes -### Sending events -The process of building event batches is more transparent with `send_batch` of V5. +### Client hierarchy +In the interest of simplifying the API surface, we've made two distinct clients, rather than having a single `EventHubClient`: +* `EventHubProducerClient` for sending events. +* `EventHubConsumerClient` for receiving events. -| In v1 | Equivalent in v5 | Sample | -|---|---|---| -| `EventHubClientAsync.add_async_sender()` and `AsyncSender.send()`| `EventHubProducerClient.send_batch()`| [send events](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhub/samples/async_samples/send_async.py) | +We've also merged the functionality from `EventProcessorHost` into `EventHubConsumerClient`. -## Migration samples +#### Approachability +By having a single entry point for sending, the `EventHubProducerClient` helps with the discoverability of the API +as you can explore all available features for sending events through methods from a single client, as opposed to searching +through documentation or exploring namespace for the types that you can instantiate. -* [Receiving events](#migrating-code-from-eventhubclient-and-asyncreceiver-to-eventhubconsumerclient-for-receiving-events) -* [Sending events](#migrating-code-from-eventhubclient-and-asyncsender-to-eventhubproducerclient-for-sending-events) -* [Receiving events with checkpointing](#migrating-code-from-eventprocessorhost-to-eventhubconsumerclient-for-receiving-events) +Similarly, by having a single entry point for receiving of any type (from single partition, all partitions, or with load balancing and checkpointing features) within Event Hubs, the `EventHubConsumerClient` helps with the discoverability of the API as you can explore all available features for receiving events through methods from a single client, as opposed to searching +through documentation or exploring namespace for the types that you can instantiate. -### Migrating code from `EventHubClient` and `AsyncReceiver` to `EventHubConsumerClient` for receiving events +#### Consistency +We now have methods with similar names, signature and location for sending and receiving. +This provides consistency and predictability on the various features of the library. -In V1, `AsyncReceiver.receive()` returns a list of EventData. +### Client constructors -In V5, EventHubConsumerClient.receive() calls user callback on_event to process events. +- While we continue to support connection strings when constructing a client, the main difference is when using Azure Active Directory. +We now use the new [`azure-identity`](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/identity/azure-identity/README.md) library +to share a single authentication solution between clients of different Azure services. -For example, this code which keeps receiving from a partition in V1: +In v1: +```python + # Authenticate with address + eventhub_client = EventHubClient(address) + # Authenticate with connection string + eventhub_client = EventHubClient.from_connection_string(conn_str) +``` +In v5: ```python -client = EventHubClientAsync.from_connection_string(connection_str, eventhub=EVENTHUB_NAME) -receiver = client.add_async_receiver(consumer_group="$Default", partition="0", offset=Offset('@latest')) -try: - await client.run_async() - logger = logging.getLogger("azure.eventhub") - while True: - received = await receiver.receive(timeout=5) - for event_data in received: - logger.info("Message received:{}".format(event_data.body_as_str())) -finally: - await client.stop_async() + # Authenticate with connection string + producer_client = EventHubProducerClient.from_connection_string(conn_str) + consumer_client = EventHubConsumerClient.from_connection_string(conn_str) + + # Authenticate with Active Directory + from azure.identity import EnvironmentCredential + producer_client = EventHubProducerClient(fully_qualified_namespace, eventhub_name, credential=EnvironmentCredential()) + consumer_client = EventHubConsumerClient(fully_qualified_namespace, eventhub_name, consumer_group='$Default', credential=EnvironmentCredential()) + + # Authenticate consumer with connection string and checkpoint + from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore + checkpoint_store = BlobCheckpointStore.from_connection_string(storage_conn_str, container_name) + consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group='$Default', checkpoint_store=checkpoint_store) ``` +### Sending events +* `add_sender`, `run`, and `stop` methods are replaced by `from_connection_string` method on `EventGridProducerClient` to more approachably open a connection ready for sending. +* `send` method is replaced by `send_batch` method on `EventGridProducerClient` to clarify that, instead of single `EventData`, either an `EventDataBatch` or list of `EventData` are sent in one call not exceeding the event hub frame size limit. +* `EventDataBatch` is created using the `create_batch` method and `EventData` messages are added to the batch using the `add` method, until the size limit is reached. -Becomes this in V5: - +In v1: ```python -logger = logging.getLogger("azure.eventhub") -async def on_event(partition_context, event): - logger.info("Message received:{}".format(event.body_as_str())) - await partition_context.update_checkpoint(event) - -client = EventHubConsumerClient.from_connection_string( - conn_str=CONNECTION_STR, consumer_group="$Default", eventhub_name=EVENTHUB_NAME -) -async with client: - await client.receive(on_event=on_event, partition_id="0", starting_position="@latest") + client = EventHubClient(address) + sender = client.add_sender() + client.run() + sender.send(EventData('Single message')) + client.stop() ``` -### Migrating code from `EventHubClient` and `AsyncSender` to `EventHubProducerClient` for sending events +In v5: +```python + producer_client = EventHubProducerClient.from_connection_string(conn_str, eventhub_name) -In V1, you create an `EventHubClient`, then create a `AsyncSender`, and call `AsyncSender.send` to send an event that may have -a list/generator of messages. + # Send EventDataBatch + event_data_batch = producer.create_batch() + event_data_batch.add(EventData('Single message')) + producer.send_batch(event_data_batch) -In V5, this has been consolidated into a one method - `EventHubProducerClient.send_batch`. -Batching merges information from multiple events into a single send, reducing -the amount of network communication needed vs sending events one at a time. -This method deterministically tells you whether the batch of events are sent to the event hub. + # Send list of EventData + event_data_batch = [EventData('Single message')] + producer.send_batch(event_data_batch) +``` -So in V1: +### Receiving events +* `add_receiver`, `run`, and `stop` methods are replaced by `from_connection_string` method on `EventGridConsumerClient` to more approachably open a connection ready for receiving. +* `receive` method is renamed `receive_batch` on `EventGridConsumerClient` to be more consistent in the usage of `batch` suffix in other methods on the producer and consumer when receiving or sending batches. +* `receive` method on `EventGridConsumerClient` now receives only a single event as opposed to previously receiving a batch of events to more clearly reflect the naming, in which `batch` is not used as a suffix. + +In v1: ```python -client = EventHubClientAsync.from_connection_string(connection_str, eventhub=EVENTHUB_NAME) -sender = client.add_async_sender(partition="0") -try: - await client.run_async() - event_data = EventData(b"A single event") - await sender.send(event_data) -finally: - await client.stop_async() + client = EventHubClient(address) + receiver = client.add_receiver(consumer_group, partition) + client.run() + batch = receiver.receive() + client.stop() ``` -In V5: +In v5: ```python -producer = EventHubProducerClient.from_connection_string(conn_str=EVENT_HUB_CONNECTION_STR, eventhub_name=EVENTHUB_NAME) -async with producer: - event_data_batch = await producer.create_batch(partition_id="0") - event_data_batch.add(EventData(b"A single event")) - await producer.send_batch(event_data_batch) + # Receive + def on_event(partition_context, event): + print("Received event from partition: {}.".format(partition_context.partition_id)) + + consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group, eventhub_name=eh_name) + with consumer_client: + consumer_client.receive(on_event=on_event) + + # Receive batch + def on_event_batch(partition_context, event_batch): + print("Partition {}, Received count: {}".format(partition_context.partition_id, len(event_batch))) + + consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group, eventhub_name=eh_name) + with consumer_client: + consumer_client.receive_batch(on_event_batch=on_event_batch) ``` -### Migrating code from `EventProcessorHost` to `EventHubConsumerClient` for receiving events - -In V1, `EventProcessorHost` allowed you to balance the load between multiple instances of -your program when receiving events. - -In V5, `EventHubConsumerClient` allows you to do the same with the `receive()` method if you -pass a `CheckpointStore` to the constructor. - -> **Note:** V1 checkpoints are not compatible with V5 checkpoints. -If pointed at the same blob, consumption will begin at the first message. -V1 checkpoint json in the respective blobs can be manually converted (per-partition) if needed. -In V1 checkpoints (sequence_number and offset) are stored in the format of json along with ownership information -as the content of the blob, while in V5, checkpoints are kept in the metadata of a blob and the metadata is composed of name-value pairs. -Please check [update_checkpoint](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhub-checkpointstoreblob/azure/eventhub/extensions/checkpointstoreblob/_blobstoragecs.py#L231-L250) in V5 for implementation detail. +### Receiving with checkpoints +Consuming events and saving checkpoints using a checkpoint store was not available in v1. -So in V1: +In v5: ```python -import logging -import asyncio -import os - -from azure.eventprocessorhost import ( - AbstractEventProcessor, - AzureStorageCheckpointLeaseManager, - EventHubConfig, - EventProcessorHost, - EPHOptions) - -logger = logging.getLogger("azure.eventhub") - - -class EventProcessor(AbstractEventProcessor): - def __init__(self, params=None): - super().__init__(params) - self._msg_counter = 0 - - async def open_async(self, context): - logger.info("Connection established {}".format(context.partition_id)) - - async def close_async(self, context, reason): - logger.info("Connection closed (reason {}, id {})".format( - reason, - context.partition_id)) - - async def process_events_async(self, context, messages): - self._msg_counter += len(messages) - logger.info("Partition id {}, Events processed {}".format(context.partition_id, self._msg_counter)) - await context.checkpoint_async() - - async def process_error_async(self, context, error): - logger.error("Event Processor Error {!r}".format(error)) - -# Storage Account Credentials -STORAGE_ACCOUNT_NAME = os.environ.get('AZURE_STORAGE_ACCOUNT') -STORAGE_KEY = os.environ.get('AZURE_STORAGE_ACCESS_KEY') -LEASE_CONTAINER_NAME = "leases" - -NAMESPACE = os.environ.get('EVENT_HUB_NAMESPACE') -EVENTHUB = os.environ.get('EVENT_HUB_NAME') -USER = os.environ.get('EVENT_HUB_SAS_POLICY') -KEY = os.environ.get('EVENT_HUB_SAS_KEY') - -# Eventhub config and storage manager -eh_config = EventHubConfig(NAMESPACE, EVENTHUB, USER, KEY, consumer_group="$Default") -eh_options = EPHOptions() -eh_options.debug_trace = False -storage_manager = AzureStorageCheckpointLeaseManager( - STORAGE_ACCOUNT_NAME, STORAGE_KEY, LEASE_CONTAINER_NAME) - -# Event loop and host -loop = asyncio.get_event_loop() -host = EventProcessorHost( - EventProcessor, - eh_config, - storage_manager, - ep_params=["param1","param2"], - eph_options=eh_options, - loop=loop) -try: - loop.run_until_complete(host.open_async()) -finally: - await host.close_async() - loop.stop() - + # Receive with checkpoint + from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore + + def on_event(partition_context, event): + print("Received event from partition: {}.".format(partition_context.partition_id)) + partition_context.update_checkpoint(event) + + checkpoint_store = BlobCheckpointStore.from_connection_string(storage_conn_str, container_name) + consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group, checkpoint_store=checkpoint_store) + with consumer_client: + consumer_client.receive(on_event=on_event) + + # Receive batch with checkpoint + from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore + + def on_event_batch(partition_context, event_batch): + print("Partition {}, Received count: {}".format(partition_context.partition_id, len(event_batch))) + # TODO: find out whether anything should be passed in, and if so, pass it in + partition_context.update_checkpoint() + + checkpoint_store = BlobCheckpointStore.from_connection_string(storage_conn_str, container_name) + consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group, checkpoint_store=checkpoint_store) + with consumer_client: + consumer_client.receive_batch(on_event_batch=on_event_batch) ``` -And in V5: -```python -import asyncio -import os -import logging -from collections import defaultdict -from azure.eventhub.aio import EventHubConsumerClient -from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore - -logging.basicConfig(level=logging.INFO) -CONNECTION_STR = os.environ["EVENT_HUB_CONN_STR"] -STORAGE_CONNECTION_STR = os.environ["AZURE_STORAGE_CONN_STR"] -BLOB_CONTAINER_NAME = "your-blob-container-name" -logger = logging.getLogger("azure.eventhub") - -events_processed = defaultdict(int) -async def on_event(partition_context, event): - partition_id = partition_context.partition_id - events_processed[partition_id] += 1 - logger.info("Partition id {}, Events processed {}".format(partition_id, events_processed[partition_id])) - await partition_context.update_checkpoint(event) - -async def on_partition_initialize(context): - logger.info("Partition {} initialized".format(context.partition_id)) - -async def on_partition_close(context, reason): - logger.info("Partition {} has closed, reason {})".format(context.partition_id, reason)) - -async def on_error(context, error): - if context: - logger.error("Partition {} has a partition related error {!r}.".format(context.partition_id, error)) - else: - logger.error("Receiving event has a non-partition error {!r}".format(error)) - -async def main(): - checkpoint_store = BlobCheckpointStore.from_connection_string(STORAGE_CONNECTION_STR, BLOB_CONTAINER_NAME) - client = EventHubConsumerClient.from_connection_string( - CONNECTION_STR, - consumer_group="$Default", - checkpoint_store=checkpoint_store, - ) - async with client: - await client.receive( - on_event, - on_error=on_error, # optional - on_partition_initialize=on_partition_initialize, # optional - on_partition_close=on_partition_close, # optional - starting_position="-1", # "-1" is from the beginning of the partition. - ) - -if __name__ == '__main__': - loop = asyncio.get_event_loop() - loop.run_until_complete(main()) -``` +## Additional samples + +More examples can be found at [Samples for azure-eventhub](https://github.com/Azure/azure-sdk-for-python/tree/master/sdk/eventhub/azure-eventhub/samples) \ No newline at end of file From 85031c450865badde70d3085c1846b7141a3b9ca Mon Sep 17 00:00:00 2001 From: Swathi Pillalamarri Date: Wed, 6 Jan 2021 21:25:21 -0500 Subject: [PATCH 2/7] made changes based on ramya/rakshith's comments --- .../azure-eventhub/migration_guide.md | 194 ++++++++++++++---- 1 file changed, 158 insertions(+), 36 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/migration_guide.md b/sdk/eventhub/azure-eventhub/migration_guide.md index 276ecbb9a463..143a789d4cb4 100644 --- a/sdk/eventhub/azure-eventhub/migration_guide.md +++ b/sdk/eventhub/azure-eventhub/migration_guide.md @@ -8,6 +8,7 @@ Familiarity with the `azure-eventhub` v1 package is assumed. For those new to th * [Migration benefits](#migration-benefits) - [Cross Service SDK improvements](#cross-service-sdk-improvements) + - [New features](#new-features) * [Important changes](#important-changes) - [Client hierarchy](#client-hierarchy) - [Client constructors](#client-constructors) @@ -26,17 +27,25 @@ To try and improve the development experience across Azure services, a set of un ### Cross Service SDK improvements -The modern Event Hubs client library also provides the ability to share in some of the cross-service improvements made to the Azure development experience, such as +The modern Event Hubs client library also provides the ability to share in some of the cross-service improvements made to the Azure development experience, such as: - using the new [`azure-identity`](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/identity/azure-identity/README.md) library to share a single authentication approach between clients - a unified logging and diagnostics pipeline offering a common view of the activities across each of the client libraries +### New features + +We have a variety of new features in version 5 of the Event Hubs library. + +- Ability to create a batch of messages with the `EventHubProducer.create_batch()` and `EventDataBatch.add()` APIs. This will help you manage events to be sent in the most optimal way. +- Ability to configure the retry policy used by operations on the clients. +- Authentication with AAD credentials using [`azure-identity`](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/identity/azure-identity/README.md). + +Refer to the [changelog](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhub/CHANGELOG.md) for more new features, changes and bug fixes. + ## Important changes ### Client hierarchy -In the interest of simplifying the API surface, we've made two distinct clients, rather than having a single `EventHubClient`: -* `EventHubProducerClient` for sending events. -* `EventHubConsumerClient` for receiving events. +In the interest of simplifying the API surface, we've made two distinct clients: the `EventHubProducerClient` for sending events and the `EventHubConsumerClient` for receiving events. This is in contrast to the single `EventHubClient` that was used to create senders and receivers. We've also merged the functionality from `EventProcessorHost` into `EventHubConsumerClient`. #### Approachability @@ -59,11 +68,28 @@ to share a single authentication solution between clients of different Azure ser In v1: ```python - # Authenticate with address + # Authenticate with address (full URI string - optionally includes URL-encoded access policy and key). For example: + # "amqps://:@.servicebus.windows.net/" eventhub_client = EventHubClient(address) # Authenticate with connection string eventhub_client = EventHubClient.from_connection_string(conn_str) + + # Authenticate with EventProcessorHost + from azure.eventprocessorhost import ( + AbstractEventProcessor, + EventHubConfig, + AzureStorageCheckpointLeaseManager, + EventProcessorHost) + + class EventProcessor(AbstractEventProcessor): + # Methods for opening connection, processing events, closing connection + ... + + eh_config = EventHubConfig(eh_namespace, eventhub_name, user, key, consumer_group="$default") + storage_manager = AzureStorageCheckpointLeaseManager(storage_account_name, storage_key, lease_container_name) + host = EventProcessorHost(EventProcessor, eh_config, storage_manager) + ``` In v5: ```python @@ -80,11 +106,15 @@ In v5: from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore checkpoint_store = BlobCheckpointStore.from_connection_string(storage_conn_str, container_name) consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group='$Default', checkpoint_store=checkpoint_store) + ``` ### Sending events -* `add_sender`, `run`, and `stop` methods are replaced by `from_connection_string` method on `EventGridProducerClient` to more approachably open a connection ready for sending. -* `send` method is replaced by `send_batch` method on `EventGridProducerClient` to clarify that, instead of single `EventData`, either an `EventDataBatch` or list of `EventData` are sent in one call not exceeding the event hub frame size limit. -* `EventDataBatch` is created using the `create_batch` method and `EventData` messages are added to the batch using the `add` method, until the size limit is reached. + +- The `run` and `stop` methods were previously used since the single `EventHubClient` controlled the lifecycle for all senders and receivers. In v5, the `run` and `stop` methods are deprecated since the `EventHubProducerClient` controls its own lifecycle. +- The `add_sender` method is no longer used to create sender clients. Instead, the `EventHubProducerClient` is used for sending events. +- The `send` method that allowed sending single events in each call is removed in favor of the `send_batch` to encourage sending events in batches for better throughput. +- The new `send_batch` method takes a list of `EventData` objects that is batched into a single message by the client before sending. +- The above approach fails if the list of events increase the size limit of the message. To safely send within size limits, use the `EventDataBatch` object to which you can add `EventData` objects until the size limit is reached after which you can send it using the same `send_batch` method. In v1: ```python @@ -110,9 +140,12 @@ In v5: ``` ### Receiving events -* `add_receiver`, `run`, and `stop` methods are replaced by `from_connection_string` method on `EventGridConsumerClient` to more approachably open a connection ready for receiving. -* `receive` method is renamed `receive_batch` on `EventGridConsumerClient` to be more consistent in the usage of `batch` suffix in other methods on the producer and consumer when receiving or sending batches. -* `receive` method on `EventGridConsumerClient` now receives only a single event as opposed to previously receiving a batch of events to more clearly reflect the naming, in which `batch` is not used as a suffix. + +- The `run` and `stop` methods were previously used since the single `EventHubClient` controlled the lifecycle for all senders and receivers. In v5, the `run` and `stop` methods are deprecated since the `EventHubConsumerClient` controls its own lifecycle. +- The `add_receiver` method is no longer used to create receiver clients. Instead, the `EventHubConsumerClient` is used for receiving events. +- The old `receive` method returned a list of `EventData`. +- The new `receive` calls the user callback `on_event` to process single events for easier and more clear interaction with event data when dealing with multiple partitions. +- The new `receive_batch` calls the user callback `on_event_batch` to process batches of events for easier and more clear interaction with event data when dealing with multiple partitions. In v1: ```python @@ -141,36 +174,125 @@ In v5: with consumer_client: consumer_client.receive_batch(on_event_batch=on_event_batch) ``` +### Migrating code from `EventProcessorHost` to `EventHubConsumerClient` for receiving events -### Receiving with checkpoints -Consuming events and saving checkpoints using a checkpoint store was not available in v1. +In V1, `EventProcessorHost` allowed you to balance the load between multiple instances of +your program when receiving events. -In v5: -```python - # Receive with checkpoint - from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore +In V5, `EventHubConsumerClient` allows you to do the same with the `receive()` method if you +pass a `CheckpointStore` to the constructor. - def on_event(partition_context, event): - print("Received event from partition: {}.".format(partition_context.partition_id)) - partition_context.update_checkpoint(event) - - checkpoint_store = BlobCheckpointStore.from_connection_string(storage_conn_str, container_name) - consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group, checkpoint_store=checkpoint_store) - with consumer_client: - consumer_client.receive(on_event=on_event) +> **Note:** V1 checkpoints are not compatible with V5 checkpoints. +If pointed at the same blob, consumption will begin at the first message. +V1 checkpoint json in the respective blobs can be manually converted (per-partition) if needed. +In V1 checkpoints (sequence_number and offset) are stored in the format of json along with ownership information +as the content of the blob, while in V5, checkpoints are kept in the metadata of a blob and the metadata is composed of name-value pairs. +Please check [update_checkpoint](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhub-checkpointstoreblob/azure/eventhub/extensions/checkpointstoreblob/_blobstoragecs.py#L231-L250) in V5 for implementation detail. - # Receive batch with checkpoint - from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore +So in V1: +```python + import logging + import asyncio + import os + from azure.eventprocessorhost import ( + AbstractEventProcessor, + AzureStorageCheckpointLeaseManager, + EventHubConfig, + EventProcessorHost, + EPHOptions) + logger = logging.getLogger("azure.eventhub") + class EventProcessor(AbstractEventProcessor): + def __init__(self, params=None): + super().__init__(params) + self._msg_counter = 0 + async def open_async(self, context): + logger.info("Connection established {}".format(context.partition_id)) + async def close_async(self, context, reason): + logger.info("Connection closed (reason {}, id {})".format( + reason, + context.partition_id)) + async def process_events_async(self, context, messages): + self._msg_counter += len(messages) + logger.info("Partition id {}, Events processed {}".format(context.partition_id, self._msg_counter)) + await context.checkpoint_async() + async def process_error_async(self, context, error): + logger.error("Event Processor Error {!r}".format(error)) + # Storage Account Credentials + STORAGE_ACCOUNT_NAME = os.environ.get('AZURE_STORAGE_ACCOUNT') + STORAGE_KEY = os.environ.get('AZURE_STORAGE_ACCESS_KEY') + LEASE_CONTAINER_NAME = "leases" + NAMESPACE = os.environ.get('EVENT_HUB_NAMESPACE') + EVENTHUB = os.environ.get('EVENT_HUB_NAME') + USER = os.environ.get('EVENT_HUB_SAS_POLICY') + KEY = os.environ.get('EVENT_HUB_SAS_KEY') + # Eventhub config and storage manager + eh_config = EventHubConfig(NAMESPACE, EVENTHUB, USER, KEY, consumer_group="$Default") + eh_options = EPHOptions() + eh_options.debug_trace = False + storage_manager = AzureStorageCheckpointLeaseManager( + STORAGE_ACCOUNT_NAME, STORAGE_KEY, LEASE_CONTAINER_NAME) + # Event loop and host + loop = asyncio.get_event_loop() + host = EventProcessorHost( + EventProcessor, + eh_config, + storage_manager, + ep_params=["param1","param2"], + eph_options=eh_options, + loop=loop) + try: + loop.run_until_complete(host.open_async()) + finally: + await host.close_async() + loop.stop() +``` - def on_event_batch(partition_context, event_batch): - print("Partition {}, Received count: {}".format(partition_context.partition_id, len(event_batch))) - # TODO: find out whether anything should be passed in, and if so, pass it in - partition_context.update_checkpoint() - - checkpoint_store = BlobCheckpointStore.from_connection_string(storage_conn_str, container_name) - consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group, checkpoint_store=checkpoint_store) - with consumer_client: - consumer_client.receive_batch(on_event_batch=on_event_batch) +And in V5: +```python + import asyncio + import os + import logging + from collections import defaultdict + from azure.eventhub.aio import EventHubConsumerClient + from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore + logging.basicConfig(level=logging.INFO) + CONNECTION_STR = os.environ["EVENT_HUB_CONN_STR"] + STORAGE_CONNECTION_STR = os.environ["AZURE_STORAGE_CONN_STR"] + BLOB_CONTAINER_NAME = "your-blob-container-name" + logger = logging.getLogger("azure.eventhub") + events_processed = defaultdict(int) + async def on_event(partition_context, event): + partition_id = partition_context.partition_id + events_processed[partition_id] += 1 + logger.info("Partition id {}, Events processed {}".format(partition_id, events_processed[partition_id])) + await partition_context.update_checkpoint(event) + async def on_partition_initialize(context): + logger.info("Partition {} initialized".format(context.partition_id)) + async def on_partition_close(context, reason): + logger.info("Partition {} has closed, reason {})".format(context.partition_id, reason)) + async def on_error(context, error): + if context: + logger.error("Partition {} has a partition related error {!r}.".format(context.partition_id, error)) + else: + logger.error("Receiving event has a non-partition error {!r}".format(error)) + async def main(): + checkpoint_store = BlobCheckpointStore.from_connection_string(STORAGE_CONNECTION_STR, BLOB_CONTAINER_NAME) + client = EventHubConsumerClient.from_connection_string( + CONNECTION_STR, + consumer_group="$Default", + checkpoint_store=checkpoint_store, + ) + async with client: + await client.receive( + on_event, + on_error=on_error, # optional + on_partition_initialize=on_partition_initialize, # optional + on_partition_close=on_partition_close, # optional + starting_position="-1", # "-1" is from the beginning of the partition. + ) + if __name__ == '__main__': + loop = asyncio.get_event_loop() + loop.run_until_complete(main()) ``` ## Additional samples From bda34b33b22454ce9cee74b9b3318353bcf1e686 Mon Sep 17 00:00:00 2001 From: Swathi Pillalamarri Date: Wed, 6 Jan 2021 21:41:17 -0500 Subject: [PATCH 3/7] fixed link in table of contents --- .../azure-eventhub/migration_guide.md | 319 +++++++++--------- 1 file changed, 160 insertions(+), 159 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/migration_guide.md b/sdk/eventhub/azure-eventhub/migration_guide.md index 143a789d4cb4..3bade517aa11 100644 --- a/sdk/eventhub/azure-eventhub/migration_guide.md +++ b/sdk/eventhub/azure-eventhub/migration_guide.md @@ -15,6 +15,7 @@ Familiarity with the `azure-eventhub` v1 package is assumed. For those new to th - [Sending](#sending-events) - [Receiving](#receiving-events) - [Receiving with checkpoints](#receiving-with-checkpoints) + - [Migrating code from EventProcessorHost to EventHubConsumerClient for receiving events](#migrating-code-from-eventprocessorhost-to-eventhubconsumerclient-for-receiving-events) * [Additional samples](#additional-samples) ## Migration benefits @@ -68,44 +69,44 @@ to share a single authentication solution between clients of different Azure ser In v1: ```python - # Authenticate with address (full URI string - optionally includes URL-encoded access policy and key). For example: - # "amqps://:@.servicebus.windows.net/" - eventhub_client = EventHubClient(address) +# Authenticate with address (full URI string - optionally includes URL-encoded access policy and key). For example: +# "amqps://:@.servicebus.windows.net/" +eventhub_client = EventHubClient(address) - # Authenticate with connection string - eventhub_client = EventHubClient.from_connection_string(conn_str) +# Authenticate with connection string +eventhub_client = EventHubClient.from_connection_string(conn_str) - # Authenticate with EventProcessorHost - from azure.eventprocessorhost import ( - AbstractEventProcessor, - EventHubConfig, - AzureStorageCheckpointLeaseManager, - EventProcessorHost) +# Authenticate with EventProcessorHost +from azure.eventprocessorhost import ( + AbstractEventProcessor, + EventHubConfig, + AzureStorageCheckpointLeaseManager, + EventProcessorHost) - class EventProcessor(AbstractEventProcessor): - # Methods for opening connection, processing events, closing connection - ... +class EventProcessor(AbstractEventProcessor): + # Methods for opening connection, processing events, closing connection + ... - eh_config = EventHubConfig(eh_namespace, eventhub_name, user, key, consumer_group="$default") - storage_manager = AzureStorageCheckpointLeaseManager(storage_account_name, storage_key, lease_container_name) - host = EventProcessorHost(EventProcessor, eh_config, storage_manager) +eh_config = EventHubConfig(eh_namespace, eventhub_name, user, key, consumer_group="$default") +storage_manager = AzureStorageCheckpointLeaseManager(storage_account_name, storage_key, lease_container_name) +host = EventProcessorHost(EventProcessor, eh_config, storage_manager) ``` In v5: ```python - # Authenticate with connection string - producer_client = EventHubProducerClient.from_connection_string(conn_str) - consumer_client = EventHubConsumerClient.from_connection_string(conn_str) +# Authenticate with connection string +producer_client = EventHubProducerClient.from_connection_string(conn_str) +consumer_client = EventHubConsumerClient.from_connection_string(conn_str) - # Authenticate with Active Directory - from azure.identity import EnvironmentCredential - producer_client = EventHubProducerClient(fully_qualified_namespace, eventhub_name, credential=EnvironmentCredential()) - consumer_client = EventHubConsumerClient(fully_qualified_namespace, eventhub_name, consumer_group='$Default', credential=EnvironmentCredential()) +# Authenticate with Active Directory +from azure.identity import EnvironmentCredential +producer_client = EventHubProducerClient(fully_qualified_namespace, eventhub_name, credential=EnvironmentCredential()) +consumer_client = EventHubConsumerClient(fully_qualified_namespace, eventhub_name, consumer_group='$Default', credential=EnvironmentCredential()) - # Authenticate consumer with connection string and checkpoint - from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore - checkpoint_store = BlobCheckpointStore.from_connection_string(storage_conn_str, container_name) - consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group='$Default', checkpoint_store=checkpoint_store) +# Authenticate consumer with connection string and checkpoint +from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore +checkpoint_store = BlobCheckpointStore.from_connection_string(storage_conn_str, container_name) +consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group='$Default', checkpoint_store=checkpoint_store) ``` ### Sending events @@ -118,25 +119,25 @@ In v5: In v1: ```python - client = EventHubClient(address) - sender = client.add_sender() - client.run() - sender.send(EventData('Single message')) - client.stop() +client = EventHubClient(address) +sender = client.add_sender() +client.run() +sender.send(EventData('Single message')) +client.stop() ``` In v5: ```python - producer_client = EventHubProducerClient.from_connection_string(conn_str, eventhub_name) +producer_client = EventHubProducerClient.from_connection_string(conn_str, eventhub_name) - # Send EventDataBatch - event_data_batch = producer.create_batch() - event_data_batch.add(EventData('Single message')) - producer.send_batch(event_data_batch) +# Send EventDataBatch +event_data_batch = producer.create_batch() +event_data_batch.add(EventData('Single message')) +producer.send_batch(event_data_batch) - # Send list of EventData - event_data_batch = [EventData('Single message')] - producer.send_batch(event_data_batch) +# Send list of EventData +event_data_batch = [EventData('Single message')] +producer.send_batch(event_data_batch) ``` ### Receiving events @@ -149,30 +150,30 @@ In v5: In v1: ```python - client = EventHubClient(address) - receiver = client.add_receiver(consumer_group, partition) - client.run() - batch = receiver.receive() - client.stop() +client = EventHubClient(address) +receiver = client.add_receiver(consumer_group, partition) +client.run() +batch = receiver.receive() +client.stop() ``` In v5: ```python - # Receive - def on_event(partition_context, event): - print("Received event from partition: {}.".format(partition_context.partition_id)) +# Receive +def on_event(partition_context, event): + print("Received event from partition: {}.".format(partition_context.partition_id)) + +consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group, eventhub_name=eh_name) +with consumer_client: + consumer_client.receive(on_event=on_event) + +# Receive batch +def on_event_batch(partition_context, event_batch): + print("Partition {}, Received count: {}".format(partition_context.partition_id, len(event_batch))) - consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group, eventhub_name=eh_name) - with consumer_client: - consumer_client.receive(on_event=on_event) - - # Receive batch - def on_event_batch(partition_context, event_batch): - print("Partition {}, Received count: {}".format(partition_context.partition_id, len(event_batch))) - - consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group, eventhub_name=eh_name) - with consumer_client: - consumer_client.receive_batch(on_event_batch=on_event_batch) +consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group, eventhub_name=eh_name) +with consumer_client: + consumer_client.receive_batch(on_event_batch=on_event_batch) ``` ### Migrating code from `EventProcessorHost` to `EventHubConsumerClient` for receiving events @@ -189,110 +190,110 @@ In V1 checkpoints (sequence_number and offset) are stored in the format of json as the content of the blob, while in V5, checkpoints are kept in the metadata of a blob and the metadata is composed of name-value pairs. Please check [update_checkpoint](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhub-checkpointstoreblob/azure/eventhub/extensions/checkpointstoreblob/_blobstoragecs.py#L231-L250) in V5 for implementation detail. -So in V1: +So in v1: ```python - import logging - import asyncio - import os - from azure.eventprocessorhost import ( - AbstractEventProcessor, - AzureStorageCheckpointLeaseManager, - EventHubConfig, - EventProcessorHost, - EPHOptions) - logger = logging.getLogger("azure.eventhub") - class EventProcessor(AbstractEventProcessor): - def __init__(self, params=None): - super().__init__(params) - self._msg_counter = 0 - async def open_async(self, context): - logger.info("Connection established {}".format(context.partition_id)) - async def close_async(self, context, reason): - logger.info("Connection closed (reason {}, id {})".format( - reason, - context.partition_id)) - async def process_events_async(self, context, messages): - self._msg_counter += len(messages) - logger.info("Partition id {}, Events processed {}".format(context.partition_id, self._msg_counter)) - await context.checkpoint_async() - async def process_error_async(self, context, error): - logger.error("Event Processor Error {!r}".format(error)) - # Storage Account Credentials - STORAGE_ACCOUNT_NAME = os.environ.get('AZURE_STORAGE_ACCOUNT') - STORAGE_KEY = os.environ.get('AZURE_STORAGE_ACCESS_KEY') - LEASE_CONTAINER_NAME = "leases" - NAMESPACE = os.environ.get('EVENT_HUB_NAMESPACE') - EVENTHUB = os.environ.get('EVENT_HUB_NAME') - USER = os.environ.get('EVENT_HUB_SAS_POLICY') - KEY = os.environ.get('EVENT_HUB_SAS_KEY') - # Eventhub config and storage manager - eh_config = EventHubConfig(NAMESPACE, EVENTHUB, USER, KEY, consumer_group="$Default") - eh_options = EPHOptions() - eh_options.debug_trace = False - storage_manager = AzureStorageCheckpointLeaseManager( - STORAGE_ACCOUNT_NAME, STORAGE_KEY, LEASE_CONTAINER_NAME) - # Event loop and host - loop = asyncio.get_event_loop() - host = EventProcessorHost( - EventProcessor, - eh_config, - storage_manager, - ep_params=["param1","param2"], - eph_options=eh_options, - loop=loop) - try: - loop.run_until_complete(host.open_async()) - finally: - await host.close_async() - loop.stop() +import logging +import asyncio +import os +from azure.eventprocessorhost import ( + AbstractEventProcessor, + AzureStorageCheckpointLeaseManager, + EventHubConfig, + EventProcessorHost, + EPHOptions) +logger = logging.getLogger("azure.eventhub") +class EventProcessor(AbstractEventProcessor): + def __init__(self, params=None): + super().__init__(params) + self._msg_counter = 0 + async def open_async(self, context): + logger.info("Connection established {}".format(context.partition_id)) + async def close_async(self, context, reason): + logger.info("Connection closed (reason {}, id {})".format( + reason, + context.partition_id)) + async def process_events_async(self, context, messages): + self._msg_counter += len(messages) + logger.info("Partition id {}, Events processed {}".format(context.partition_id, self._msg_counter)) + await context.checkpoint_async() + async def process_error_async(self, context, error): + logger.error("Event Processor Error {!r}".format(error)) +# Storage Account Credentials +STORAGE_ACCOUNT_NAME = os.environ.get('AZURE_STORAGE_ACCOUNT') +STORAGE_KEY = os.environ.get('AZURE_STORAGE_ACCESS_KEY') +LEASE_CONTAINER_NAME = "leases" +NAMESPACE = os.environ.get('EVENT_HUB_NAMESPACE') +EVENTHUB = os.environ.get('EVENT_HUB_NAME') +USER = os.environ.get('EVENT_HUB_SAS_POLICY') +KEY = os.environ.get('EVENT_HUB_SAS_KEY') +# Eventhub config and storage manager +eh_config = EventHubConfig(NAMESPACE, EVENTHUB, USER, KEY, consumer_group="$Default") +eh_options = EPHOptions() +eh_options.debug_trace = False +storage_manager = AzureStorageCheckpointLeaseManager( + STORAGE_ACCOUNT_NAME, STORAGE_KEY, LEASE_CONTAINER_NAME) +# Event loop and host +loop = asyncio.get_event_loop() +host = EventProcessorHost( + EventProcessor, + eh_config, + storage_manager, + ep_params=["param1","param2"], + eph_options=eh_options, + loop=loop) +try: + loop.run_until_complete(host.open_async()) +finally: + await host.close_async() + loop.stop() ``` -And in V5: +And in v5: ```python - import asyncio - import os - import logging - from collections import defaultdict - from azure.eventhub.aio import EventHubConsumerClient - from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore - logging.basicConfig(level=logging.INFO) - CONNECTION_STR = os.environ["EVENT_HUB_CONN_STR"] - STORAGE_CONNECTION_STR = os.environ["AZURE_STORAGE_CONN_STR"] - BLOB_CONTAINER_NAME = "your-blob-container-name" - logger = logging.getLogger("azure.eventhub") - events_processed = defaultdict(int) - async def on_event(partition_context, event): - partition_id = partition_context.partition_id - events_processed[partition_id] += 1 - logger.info("Partition id {}, Events processed {}".format(partition_id, events_processed[partition_id])) - await partition_context.update_checkpoint(event) - async def on_partition_initialize(context): - logger.info("Partition {} initialized".format(context.partition_id)) - async def on_partition_close(context, reason): - logger.info("Partition {} has closed, reason {})".format(context.partition_id, reason)) - async def on_error(context, error): - if context: - logger.error("Partition {} has a partition related error {!r}.".format(context.partition_id, error)) - else: - logger.error("Receiving event has a non-partition error {!r}".format(error)) - async def main(): - checkpoint_store = BlobCheckpointStore.from_connection_string(STORAGE_CONNECTION_STR, BLOB_CONTAINER_NAME) - client = EventHubConsumerClient.from_connection_string( - CONNECTION_STR, - consumer_group="$Default", - checkpoint_store=checkpoint_store, +import asyncio +import os +import logging +from collections import defaultdict +from azure.eventhub.aio import EventHubConsumerClient +from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore +logging.basicConfig(level=logging.INFO) +CONNECTION_STR = os.environ["EVENT_HUB_CONN_STR"] +STORAGE_CONNECTION_STR = os.environ["AZURE_STORAGE_CONN_STR"] +BLOB_CONTAINER_NAME = "your-blob-container-name" +logger = logging.getLogger("azure.eventhub") +events_processed = defaultdict(int) +async def on_event(partition_context, event): + partition_id = partition_context.partition_id + events_processed[partition_id] += 1 + logger.info("Partition id {}, Events processed {}".format(partition_id, events_processed[partition_id])) + await partition_context.update_checkpoint(event) +async def on_partition_initialize(context): + logger.info("Partition {} initialized".format(context.partition_id)) +async def on_partition_close(context, reason): + logger.info("Partition {} has closed, reason {})".format(context.partition_id, reason)) +async def on_error(context, error): + if context: + logger.error("Partition {} has a partition related error {!r}.".format(context.partition_id, error)) + else: + logger.error("Receiving event has a non-partition error {!r}".format(error)) +async def main(): + checkpoint_store = BlobCheckpointStore.from_connection_string(STORAGE_CONNECTION_STR, BLOB_CONTAINER_NAME) + client = EventHubConsumerClient.from_connection_string( + CONNECTION_STR, + consumer_group="$Default", + checkpoint_store=checkpoint_store, + ) + async with client: + await client.receive( + on_event, + on_error=on_error, # optional + on_partition_initialize=on_partition_initialize, # optional + on_partition_close=on_partition_close, # optional + starting_position="-1", # "-1" is from the beginning of the partition. ) - async with client: - await client.receive( - on_event, - on_error=on_error, # optional - on_partition_initialize=on_partition_initialize, # optional - on_partition_close=on_partition_close, # optional - starting_position="-1", # "-1" is from the beginning of the partition. - ) - if __name__ == '__main__': - loop = asyncio.get_event_loop() - loop.run_until_complete(main()) +if __name__ == '__main__': + loop = asyncio.get_event_loop() + loop.run_until_complete(main()) ``` ## Additional samples From b55761da9238a3b74d280949ff8d5a1a5933fe6c Mon Sep 17 00:00:00 2001 From: Swathi Pillalamarri Date: Wed, 6 Jan 2021 21:42:33 -0500 Subject: [PATCH 4/7] fix table of contents again --- sdk/eventhub/azure-eventhub/migration_guide.md | 1 - 1 file changed, 1 deletion(-) diff --git a/sdk/eventhub/azure-eventhub/migration_guide.md b/sdk/eventhub/azure-eventhub/migration_guide.md index 3bade517aa11..116334a0c91f 100644 --- a/sdk/eventhub/azure-eventhub/migration_guide.md +++ b/sdk/eventhub/azure-eventhub/migration_guide.md @@ -14,7 +14,6 @@ Familiarity with the `azure-eventhub` v1 package is assumed. For those new to th - [Client constructors](#client-constructors) - [Sending](#sending-events) - [Receiving](#receiving-events) - - [Receiving with checkpoints](#receiving-with-checkpoints) - [Migrating code from EventProcessorHost to EventHubConsumerClient for receiving events](#migrating-code-from-eventprocessorhost-to-eventhubconsumerclient-for-receiving-events) * [Additional samples](#additional-samples) From af4983b8768c43411ab75b5714a8a49e68d6fbe3 Mon Sep 17 00:00:00 2001 From: Swathi Pillalamarri Date: Thu, 7 Jan 2021 14:32:34 -0500 Subject: [PATCH 5/7] fixed ramyas comments --- .../azure-eventhub/migration_guide.md | 47 +++++++------------ 1 file changed, 18 insertions(+), 29 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/migration_guide.md b/sdk/eventhub/azure-eventhub/migration_guide.md index 116334a0c91f..db8fd9762699 100644 --- a/sdk/eventhub/azure-eventhub/migration_guide.md +++ b/sdk/eventhub/azure-eventhub/migration_guide.md @@ -68,45 +68,33 @@ to share a single authentication solution between clients of different Azure ser In v1: ```python -# Authenticate with address (full URI string - optionally includes URL-encoded access policy and key). For example: -# "amqps://:@.servicebus.windows.net/" +# Authenticate with address eventhub_client = EventHubClient(address) # Authenticate with connection string eventhub_client = EventHubClient.from_connection_string(conn_str) -# Authenticate with EventProcessorHost -from azure.eventprocessorhost import ( - AbstractEventProcessor, - EventHubConfig, - AzureStorageCheckpointLeaseManager, - EventProcessorHost) - -class EventProcessor(AbstractEventProcessor): - # Methods for opening connection, processing events, closing connection - ... - +# Authenticate the EventProcessorHost and StorageCheckpointLeaseManager eh_config = EventHubConfig(eh_namespace, eventhub_name, user, key, consumer_group="$default") storage_manager = AzureStorageCheckpointLeaseManager(storage_account_name, storage_key, lease_container_name) -host = EventProcessorHost(EventProcessor, eh_config, storage_manager) - +host = EventProcessorHost(EventProcessor, eventhub_config, storage_manager) ``` In v5: ```python +# Address is no longer used for authentication. + # Authenticate with connection string producer_client = EventHubProducerClient.from_connection_string(conn_str) consumer_client = EventHubConsumerClient.from_connection_string(conn_str) +checkpoint_store = BlobCheckpointStore.from_connection_string(storage_conn_str, container_name) +consumer_client_with_checkpoint_store = EventHubConsumerClient.from_connection_string(conn_str, consumer_group='$Default', checkpoint_store=checkpoint_store) # Authenticate with Active Directory from azure.identity import EnvironmentCredential producer_client = EventHubProducerClient(fully_qualified_namespace, eventhub_name, credential=EnvironmentCredential()) consumer_client = EventHubConsumerClient(fully_qualified_namespace, eventhub_name, consumer_group='$Default', credential=EnvironmentCredential()) - -# Authenticate consumer with connection string and checkpoint -from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore -checkpoint_store = BlobCheckpointStore.from_connection_string(storage_conn_str, container_name) -consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group='$Default', checkpoint_store=checkpoint_store) - +checkpoint_store = BlobCheckpointStore(blob_account_url, container_name, credential=EnvironmentCredential()) +consumer_client_with_checkpoint_store = EventHubConsumerClient(fully_qualified_namespace, eventhub_name, consumer_group='$Default', credential=EnvironmentCredential(), checkpoint_store=checkpoint_store) ``` ### Sending events @@ -129,13 +117,14 @@ In v5: ```python producer_client = EventHubProducerClient.from_connection_string(conn_str, eventhub_name) -# Send EventDataBatch -event_data_batch = producer.create_batch() -event_data_batch.add(EventData('Single message')) +# Send list of EventData. This can fail if the list exceeds size limit. +event_data_batch = [EventData('Single message')] producer.send_batch(event_data_batch) -# Send list of EventData -event_data_batch = [EventData('Single message')] +# Send EventDataBatch. Multiple messages will safely be sent by using `create_batch` to create a batch object. +# `add` will throw a ValueError if added size results in the batch exceeding the maximum batch size. +event_data_batch = producer.create_batch() +event_data_batch.add(EventData('Single message')) producer.send_batch(event_data_batch) ``` @@ -143,9 +132,9 @@ producer.send_batch(event_data_batch) - The `run` and `stop` methods were previously used since the single `EventHubClient` controlled the lifecycle for all senders and receivers. In v5, the `run` and `stop` methods are deprecated since the `EventHubConsumerClient` controls its own lifecycle. - The `add_receiver` method is no longer used to create receiver clients. Instead, the `EventHubConsumerClient` is used for receiving events. -- The old `receive` method returned a list of `EventData`. -- The new `receive` calls the user callback `on_event` to process single events for easier and more clear interaction with event data when dealing with multiple partitions. -- The new `receive_batch` calls the user callback `on_event_batch` to process batches of events for easier and more clear interaction with event data when dealing with multiple partitions. +- In v1, the `receive` method returned a list of `EventData`. You would call this method repeatedly every time you want receive a set of events. In v5, the new `receive` method takes user callback to process events and any resulting errors. This way, you call the method once and it continues to process incoming events until you stop it. +- Additionally, we have a method `receive_batch` which behaves the same as `receive`, but calls the user callback with a batch of events instead of single events. +- The same methods can be used whether you want to receive from a single partition or from all partitions. In v1: ```python From 23c2987e3bb56edaa0da8b9533a900cd344ba16d Mon Sep 17 00:00:00 2001 From: Swathi Pillalamarri Date: Thu, 7 Jan 2021 15:02:39 -0500 Subject: [PATCH 6/7] fixed last comment --- sdk/eventhub/azure-eventhub/migration_guide.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/migration_guide.md b/sdk/eventhub/azure-eventhub/migration_guide.md index db8fd9762699..ef0f7dd9bb1a 100644 --- a/sdk/eventhub/azure-eventhub/migration_guide.md +++ b/sdk/eventhub/azure-eventhub/migration_guide.md @@ -62,9 +62,11 @@ This provides consistency and predictability on the various features of the libr ### Client constructors -- While we continue to support connection strings when constructing a client, the main difference is when using Azure Active Directory. -We now use the new [`azure-identity`](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/identity/azure-identity/README.md) library +While we continue to support connection strings when constructing a client, below are the differences in the two versions: +- In v5, we now support the use of Azure Active Directory for authentication. +The new [`azure-identity`](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/identity/azure-identity/README.md) library allows us to share a single authentication solution between clients of different Azure services. +- The option to construct a client using an address of the form `amqps://:@/` is no longer supported in v5. This address is not readily available in the Azure portal or in any tooling and so was subject to human error. We instead recommend using the connection string if you want to use a SAS policy. In v1: ```python From 65112ea5f3275f6c0e68afc623611965e901cf4c Mon Sep 17 00:00:00 2001 From: Swathi Pillalamarri Date: Thu, 7 Jan 2021 15:30:40 -0500 Subject: [PATCH 7/7] adams comments --- sdk/eventhub/azure-eventhub/migration_guide.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/sdk/eventhub/azure-eventhub/migration_guide.md b/sdk/eventhub/azure-eventhub/migration_guide.md index ef0f7dd9bb1a..6510bfc6b79f 100644 --- a/sdk/eventhub/azure-eventhub/migration_guide.md +++ b/sdk/eventhub/azure-eventhub/migration_guide.md @@ -12,8 +12,8 @@ Familiarity with the `azure-eventhub` v1 package is assumed. For those new to th * [Important changes](#important-changes) - [Client hierarchy](#client-hierarchy) - [Client constructors](#client-constructors) - - [Sending](#sending-events) - - [Receiving](#receiving-events) + - [Sending events](#sending-events) + - [Receiving events](#receiving-events) - [Migrating code from EventProcessorHost to EventHubConsumerClient for receiving events](#migrating-code-from-eventprocessorhost-to-eventhubconsumerclient-for-receiving-events) * [Additional samples](#additional-samples) @@ -155,7 +155,7 @@ def on_event(partition_context, event): consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group, eventhub_name=eh_name) with consumer_client: - consumer_client.receive(on_event=on_event) + consumer_client.receive(on_event=on_even, partition_id=partition_id) # Receive batch def on_event_batch(partition_context, event_batch): @@ -163,7 +163,7 @@ def on_event_batch(partition_context, event_batch): consumer_client = EventHubConsumerClient.from_connection_string(conn_str, consumer_group, eventhub_name=eh_name) with consumer_client: - consumer_client.receive_batch(on_event_batch=on_event_batch) + consumer_client.receive_batch(on_event_batch=on_event_batch, partition_id=partition_id) ``` ### Migrating code from `EventProcessorHost` to `EventHubConsumerClient` for receiving events