diff --git a/sdk/eventhub/azure-eventhubs/HISTORY.md b/sdk/eventhub/azure-eventhubs/HISTORY.md index 11ae91dc04fb..0d7417d6cc28 100644 --- a/sdk/eventhub/azure-eventhubs/HISTORY.md +++ b/sdk/eventhub/azure-eventhubs/HISTORY.md @@ -4,16 +4,22 @@ **New features** -- Added ability to create and send EventDataBatch object with limited data size. +- Added new class `EventDataBatch` for publication of a batch of events with known size constraint. +- Added new method `create_batch` to producer for creating EventDataBatch objects. - Added new configuration parameters for exponential delay among each retry operation. - `retry_total`: The total number of attempts to redo the failed operation. - `backoff_factor`: The delay time factor. - `backoff_max`: The maximum delay time in total. +- Added support for context manager on `EventHubClient`. **Breaking changes** -- New `EventProcessor` design - - The `EventProcessorHost` was waived. +- Replaced `max_retries` configuration parameter of the EventHubClient with `retry_total`. +- Introduced the initial concept of a new version of the `EventProcessor`, intended as a neutral framework for processing events across all partitions for a given Event Hub and in the context of a specific Consumer Group. This early preview is intended to allow consumers to test the new design using a single instance that does not persist checkpoints to any durable store. + - `EventProcessor`: EventProcessor creates and runs consumers for all partitions of the eventhub. + - `PartitionManager`: PartitionManager defines the interface for getting/claiming ownerships of partitions and updating checkpoints. + - `PartitionProcessor`: PartitionProcessor defines the interface for processing events. + - `CheckpointManager`: CheckpointManager takes responsibility for updating checkpoints during events processing. ## 5.0.0b1 (2019-06-25) diff --git a/sdk/eventhub/azure-eventhubs/README.md b/sdk/eventhub/azure-eventhubs/README.md index ccb15d8f4bc3..2ab2ad758484 100644 --- a/sdk/eventhub/azure-eventhubs/README.md +++ b/sdk/eventhub/azure-eventhubs/README.md @@ -90,6 +90,7 @@ The following sections provide several code snippets covering some of the most c - [Consume events from an Event Hub](#consume-events-from-an-event-hub) - [Async publish events to an Event Hub](#async-publish-events-to-an-event-hub) - [Async consume events from an Event Hub](#async-consume-events-from-an-event-hub) +- [Consume events from all partitions of an Event Hub](#consume-events-from-all-partitions-of-an-event-hub) ### Inspect an Event Hub @@ -206,6 +207,54 @@ finally: pass ``` +### Consume events from all partitions of an Event Hub + +To consume events from all partitions of an Event Hub, you'll create an `EventProcessor` for a specific consumer group. When an Event Hub is created, it provides a default consumer group that can be used to get started. + +The `EventProcessor` will delegate processing of events to a `PartitionProcessor` implementation that you provide, allowing your logic to focus on the logic needed to provide value while the processor holds responsibility for managing the underlying consumer operations. In our example, we will focus on building the `EventProcessor` and use a very minimal partition processor that does no actual processing. + +```python +import asyncio + +from azure.eventhub.aio import EventHubClient +from azure.eventhub.eventprocessor import EventProcessor, PartitionProcessor, Sqlite3PartitionManager + +connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>' + +async def do_operation(event): + # do some sync or async operations. If the operation is i/o intensive, async will have better performance + print(event) + +class MyPartitionProcessor(PartitionProcessor): + def __init__(self, checkpoint_manager): + super(MyPartitionProcessor, self).__init__(checkpoint_manager) + + async def process_events(self, events): + if events: + await asyncio.gather(*[do_operation(event) for event in events]) + await self._checkpoint_manager.update_checkpoint(events[-1].offset, events[-1].sequence_number) + +def partition_processor_factory(checkpoint_manager): + return MyPartitionProcessor(checkpoint_manager) + +async def main(): + client = EventHubClient.from_connection_string(connection_str, receive_timeout=5, retry_total=3) + partition_manager = Sqlite3PartitionManager() + try: + event_processor = EventProcessor(client, "$default", MyPartitionProcessor, partition_manager) + # You can also define a callable object for creating PartitionProcessor like below: + # event_processor = EventProcessor(client, "$default", partition_processor_factory, partition_manager) + asyncio.ensure_future(event_processor.start()) + await asyncio.sleep(60) + await event_processor.stop() + finally: + await partition_manager.close() + +if __name__ == '__main__': + loop = asyncio.get_event_loop() + loop.run_until_complete(main()) +``` + ## Troubleshooting ### General @@ -230,7 +279,7 @@ These are the samples in our repo demonstraing the usage of the library. - [./examples/recv.py](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhubs/examples/recv.py) - use consumer to consume events - [./examples/async_examples/send_async.py](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhubs/examples/async_examples/send_async.py) - async/await support of a producer - [./examples/async_examples/recv_async.py](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhubs/examples/async_examples/recv_async.py) - async/await support of a consumer -- [./examples/eph.py](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhubs/examples/eph.py) - event processor host +- [./examples/eventprocessor/event_processor_example.py](https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhubs/examples/eventprocessor/event_processor_example.py) - event processor ### Documentation diff --git a/sdk/eventhub/azure-eventhubs/azure/eventhub/_consumer_producer_mixin.py b/sdk/eventhub/azure-eventhubs/azure/eventhub/_consumer_producer_mixin.py index 9124ff261949..8e3cae1e8a0b 100644 --- a/sdk/eventhub/azure-eventhubs/azure/eventhub/_consumer_producer_mixin.py +++ b/sdk/eventhub/azure-eventhubs/azure/eventhub/_consumer_producer_mixin.py @@ -15,9 +15,9 @@ def _retry_decorator(to_be_wrapped_func): def wrapped_func(self, *args, **kwargs): - timeout = kwargs.pop("timeout", None) + timeout = kwargs.pop("timeout", 100000) if not timeout: - timeout = 100000 # timeout None or 0 mean no timeout. 100000 seconds is equivalent to no timeout + timeout = 100000 # timeout equals to 0 means no timeout, set the value to be a large number. timeout_time = time.time() + timeout max_retries = self.client.config.max_retries retry_count = 0 diff --git a/sdk/eventhub/azure-eventhubs/azure/eventhub/aio/_consumer_producer_mixin_async.py b/sdk/eventhub/azure-eventhubs/azure/eventhub/aio/_consumer_producer_mixin_async.py index a90198f42f54..8ccdcdddcd47 100644 --- a/sdk/eventhub/azure-eventhubs/azure/eventhub/aio/_consumer_producer_mixin_async.py +++ b/sdk/eventhub/azure-eventhubs/azure/eventhub/aio/_consumer_producer_mixin_async.py @@ -15,9 +15,9 @@ def _retry_decorator(to_be_wrapped_func): async def wrapped_func(self, *args, **kwargs): - timeout = kwargs.pop("timeout", None) + timeout = kwargs.pop("timeout", 100000) if not timeout: - timeout = 100000 # timeout None or 0 mean no timeout. 100000 seconds is equivalent to no timeout + timeout = 100000 # timeout equals to 0 means no timeout, set the value to be a large number. timeout_time = time.time() + timeout max_retries = self.client.config.max_retries retry_count = 0 diff --git a/sdk/eventhub/azure-eventhubs/azure/eventhub/client_abstract.py b/sdk/eventhub/azure-eventhubs/azure/eventhub/client_abstract.py index efc69e870b54..dd62e3e76de1 100644 --- a/sdk/eventhub/azure-eventhubs/azure/eventhub/client_abstract.py +++ b/sdk/eventhub/azure-eventhubs/azure/eventhub/client_abstract.py @@ -118,9 +118,9 @@ def __init__(self, host, event_hub_path, credential, **kwargs): :type auth_timeout: float :param user_agent: The user agent that needs to be appended to the built in user agent string. :type user_agent: str - :param max_retries: The max number of attempts to redo the failed operation when an error happened. Default + :param retry_total: The total number of attempts to redo the failed operation when an error happened. Default value is 3. - :type max_retries: int + :type retry_total: int :param transport_type: The type of transport protocol that will be used for communicating with the Event Hubs service. Default is ~azure.eventhub.TransportType.Amqp. :type transport_type: ~azure.eventhub.TransportType @@ -239,9 +239,9 @@ def from_connection_string(cls, conn_str, **kwargs): :type auth_timeout: float :param user_agent: The user agent that needs to be appended to the built in user agent string. :type user_agent: str - :param max_retries: The max number of attempts to redo the failed operation when an error happened. Default + :param retry_total: The total number of attempts to redo the failed operation when an error happened. Default value is 3. - :type max_retries: int + :type retry_total: int :param transport_type: The type of transport protocol that will be used for communicating with the Event Hubs service. Default is ~azure.eventhub.TransportType.Amqp. :type transport_type: ~azure.eventhub.TransportType diff --git a/sdk/eventhub/azure-eventhubs/azure/eventhub/common.py b/sdk/eventhub/azure-eventhubs/azure/eventhub/common.py index 701b45484d75..9df6aa5d1c6a 100644 --- a/sdk/eventhub/azure-eventhubs/azure/eventhub/common.py +++ b/sdk/eventhub/azure-eventhubs/azure/eventhub/common.py @@ -63,8 +63,6 @@ def __init__(self, body=None, to_device=None): :param body: The data to send in a single message. :type body: str, bytes or list - :param batch: A data generator to send batched messages. - :type batch: Generator :param to_device: An IoT device to route to. :type to_device: str """ diff --git a/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/checkpoint_manager.py b/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/checkpoint_manager.py index 88a3c3616592..2714f675b28c 100644 --- a/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/checkpoint_manager.py +++ b/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/checkpoint_manager.py @@ -8,7 +8,9 @@ class CheckpointManager(object): - """Every PartitionProcessor has a CheckpointManager to save the partition's checkpoint. + """ + CheckpointManager is responsible for the creation of checkpoints. + The interaction with the chosen storage service is done via ~azure.eventhub.eventprocessor.PartitionManager. """ def __init__(self, partition_id: str, eventhub_name: str, consumer_group_name: str, owner_id: str, partition_manager: PartitionManager): @@ -19,10 +21,13 @@ def __init__(self, partition_id: str, eventhub_name: str, consumer_group_name: s self.partition_manager = partition_manager async def update_checkpoint(self, offset, sequence_number=None): - """Users call this method in PartitionProcessor.process_events() to save checkpoints + """ + Updates the checkpoint using the given information for the associated partition and consumer group in the chosen storage service. - :param offset: offset of the processed EventData - :param sequence_number: sequence_number of the processed EventData + :param offset: The offset of the ~azure.eventhub.EventData the new checkpoint will be associated with. + :type offset: str + :param sequence_number: The sequence_number of the ~azure.eventhub.EventData the new checkpoint will be associated with. + :type sequence_number: int :return: None """ await self.partition_manager.update_checkpoint( diff --git a/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/event_processor.py b/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/event_processor.py index 16786af91ff6..27f0e998835a 100644 --- a/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/event_processor.py +++ b/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/event_processor.py @@ -24,23 +24,26 @@ class EventProcessor(object): def __init__(self, eventhub_client: EventHubClient, consumer_group_name: str, partition_processor_factory: Callable[[CheckpointManager], PartitionProcessor], partition_manager: PartitionManager, **kwargs): - """An EventProcessor automatically creates and runs consumers for all partitions of the eventhub. + """ + An EventProcessor constantly receives events from all partitions of the Event Hub in the context of a given + consumer group. The received data will be sent to PartitionProcessor to be processed. It provides the user a convenient way to receive events from multiple partitions and save checkpoints. - If multiple EventProcessors are running for an event hub, they will automatically balance loading. - This load balancer won't be available until preview 3. - - :param eventhub_client: an instance of azure.eventhub.aio.EventClient object - :param consumer_group_name: the consumer group that is used to receive events from - :param partition_processor_factory: a callable (type or function) that is called to return a PartitionProcessor. - Users define their own PartitionProcessor by subclassing it implement abstract method process_events() to - implement their own operation logic. - :param partition_manager: an instance of a PartitionManager implementation. A partition manager claims ownership - of partitions and saves partition checkpoints to a data storage. For preview 2, sample Sqlite3PartitionManager is - already provided. For the future releases there will be more PartitionManager implementation provided to save - checkpoint to different data storage. Users can also implement their PartitionManager class to user their own - preferred data storage. - :param initial_event_position: the offset to start a partition consumer if the partition has no checkpoint yet + If multiple EventProcessors are running for an event hub, they will automatically balance load. + This load balancing won't be available until preview 3. + + :param eventhub_client: An instance of ~azure.eventhub.aio.EventClient object + :type eventhub_client: ~azure.eventhub.aio.EventClient + :param consumer_group_name: The name of the consumer group this event processor is associated with. Events will + be read only in the context of this group. + :type consumer_group_name: str + :param partition_processor_factory: A callable(type or function) object that creates an instance of a class + implementing the ~azure.eventhub.eventprocessor.PartitionProcessor interface. + :type partition_processor_factory: callable object + :param partition_manager: Interacts with the storage system, dealing with ownership and checkpoints. + For preview 2, sample Sqlite3PartitionManager is provided. + :type partition_manager: Class implementing the ~azure.eventhub.eventprocessor.PartitionManager interface. + :param initial_event_position: The offset to start a partition consumer if the partition has no checkpoint yet. :type initial_event_position: int or str Example: diff --git a/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_manager.py b/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_manager.py index 99222e264029..e4ecb1bec824 100644 --- a/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_manager.py +++ b/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_manager.py @@ -8,16 +8,21 @@ class PartitionManager(ABC): - """Subclass PartitionManager to implement the read/write access to storage service to list/claim ownership and save checkpoint. - + """ + PartitionManager deals with the interaction with the chosen storage service. + It's able to list/claim ownership and create checkpoint. """ @abstractmethod async def list_ownership(self, eventhub_name: str, consumer_group_name: str) -> Iterable[Dict[str, Any]]: """ + Retrieves a complete ownership list from the chosen storage service. - :param eventhub_name: - :param consumer_group_name: + :param eventhub_name: The name of the specific Event Hub the ownership are associated with, relative to + the Event Hubs namespace that contains it. + :type eventhub_name: str + :param consumer_group_name: The name of the consumer group the ownership are associated with. + :type consumer_group_name: str :return: Iterable of dictionaries containing the following partition ownership information: eventhub_name consumer_group_name @@ -33,11 +38,45 @@ async def list_ownership(self, eventhub_name: str, consumer_group_name: str) -> @abstractmethod async def claim_ownership(self, partitions: Iterable[Dict[str, Any]]) -> Iterable[Dict[str, Any]]: + """ + Tries to claim a list of specified ownership. + + :param partitions: Iterable of dictionaries containing all the ownership to claim. + :type partitions: Iterable of dict + :return: Iterable of dictionaries containing the following partition ownership information: + eventhub_name + consumer_group_name + owner_id + partition_id + owner_level + offset + sequence_number + last_modified_time + etag + """ pass @abstractmethod async def update_checkpoint(self, eventhub_name, consumer_group_name, partition_id, owner_id, offset, sequence_number) -> None: + """ + Updates the checkpoint using the given information for the associated partition and consumer group in the chosen storage service. + + :param eventhub_name: The name of the specific Event Hub the ownership are associated with, relative to + the Event Hubs namespace that contains it. + :type eventhub_name: str + :param consumer_group_name: The name of the consumer group the ownership are associated with. + :type consumer_group_name: str + :param partition_id: The partition id which the checkpoint is created for. + :type partition_id: str + :param owner_id: The identifier of the ~azure.eventhub.eventprocessor.EventProcessor. + :type owner_id: str + :param offset: The offset of the ~azure.eventhub.EventData the new checkpoint will be associated with. + :type offset: str + :param sequence_number: The sequence_number of the ~azure.eventhub.EventData the new checkpoint will be associated with. + :type sequence_number: int + :return: + """ pass async def close(self): diff --git a/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_processor.py b/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_processor.py index 1a17cd513876..ea97a5597be2 100644 --- a/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_processor.py +++ b/sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_processor.py @@ -18,6 +18,11 @@ class CloseReason(Enum): class PartitionProcessor(ABC): + """ + PartitionProcessor processes events received from the Azure Event Hubs service. A single instance of a class + implementing this interface will be created for every partition the associated ~azure.eventhub.eventprocessor.EventProcessor owns. + + """ def __init__(self, checkpoint_manager: CheckpointManager): self._checkpoint_manager = checkpoint_manager @@ -27,6 +32,9 @@ async def close(self, reason): There are different reasons to trigger the PartitionProcessor to close. Refer to enum class CloseReason + :param reason: Reason for closing the PartitionProcessor. + :type reason: CloseReason + """ pass @@ -34,11 +42,17 @@ async def close(self, reason): async def process_events(self, events: List[EventData]): """Called when a batch of events have been received. + :param events: Received events. + :type events: list[~azure.eventhub.common.EventData] + """ pass async def process_error(self, error): """Called when an error happens + :param error: The error that happens. + :type error: Exception + """ pass diff --git a/sdk/eventhub/azure-eventhubs/examples/eph.py b/sdk/eventhub/azure-eventhubs/examples/eph.py deleted file mode 100644 index 66e4e6aa866f..000000000000 --- a/sdk/eventhub/azure-eventhubs/examples/eph.py +++ /dev/null @@ -1,129 +0,0 @@ -# -------------------------------------------------------------------------------------------- -# Copyright (c) Microsoft Corporation. All rights reserved. -# Licensed under the MIT License. See License.txt in the project root for license information. -# ----------------------------------------------------------------------------------- - -import logging -import asyncio -import sys -import os -import signal -import functools - -from azure.eventprocessorhost import ( - AbstractEventProcessor, - AzureStorageCheckpointLeaseManager, - EventHubConfig, - EventProcessorHost, - EPHOptions) - -import examples -logger = examples.get_logger(logging.INFO) - - -class EventProcessor(AbstractEventProcessor): - """ - Example Implmentation of AbstractEventProcessor - """ - - def __init__(self, params=None): - """ - Init Event processor - """ - super().__init__(params) - self._msg_counter = 0 - - async def open_async(self, context): - """ - Called by processor host to initialize the event processor. - """ - logger.info("Connection established {}".format(context.partition_id)) - - async def close_async(self, context, reason): - """ - Called by processor host to indicate that the event processor is being stopped. - :param context: Information about the partition - :type context: ~azure.eventprocessorhost.PartitionContext - """ - logger.info("Connection closed (reason {}, id {}, offset {}, sq_number {})".format( - reason, - context.partition_id, - context.offset, - context.sequence_number)) - - async def process_events_async(self, context, messages): - """ - Called by the processor host when a batch of events has arrived. - This is where the real work of the event processor is done. - :param context: Information about the partition - :type context: ~azure.eventprocessorhost.PartitionContext - :param messages: The events to be processed. - :type messages: list[~azure.eventhub.common.EventData] - """ - logger.info("Events processed {}".format(context.sequence_number)) - await context.checkpoint_async() - - async def process_error_async(self, context, error): - """ - Called when the underlying client experiences an error while receiving. - EventProcessorHost will take care of recovering from the error and - continuing to pump messages,so no action is required from - :param context: Information about the partition - :type context: ~azure.eventprocessorhost.PartitionContext - :param error: The error that occured. - """ - logger.error("Event Processor Error {!r}".format(error)) - - -async def wait_and_close(host): - """ - Run EventProcessorHost for 2 minutes then shutdown. - """ - await asyncio.sleep(60) - await host.close_async() - - -try: - loop = asyncio.get_event_loop() - - # 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.release_pump_on_timeout = True - eh_options.debug_trace = False - storage_manager = AzureStorageCheckpointLeaseManager( - STORAGE_ACCOUNT_NAME, STORAGE_KEY, LEASE_CONTAINER_NAME) - - # Event loop and host - host = EventProcessorHost( - EventProcessor, - eh_config, - storage_manager, - ep_params=["param1", "param2"], - eph_options=eh_options, - loop=loop) - - tasks = asyncio.gather( - host.open_async(), - wait_and_close(host)) - loop.run_until_complete(tasks) - -except KeyboardInterrupt: - # Canceling pending tasks and stopping the loop - for task in asyncio.Task.all_tasks(): - task.cancel() - loop.run_forever() - tasks.exception() - -finally: - loop.stop() diff --git a/sdk/eventhub/azure-eventhubs/examples/eventprocessor/event_processor_example.py b/sdk/eventhub/azure-eventhubs/examples/eventprocessor/event_processor_example.py index e2254d05427d..55985a7aec68 100644 --- a/sdk/eventhub/azure-eventhubs/examples/eventprocessor/event_processor_example.py +++ b/sdk/eventhub/azure-eventhubs/examples/eventprocessor/event_processor_example.py @@ -29,11 +29,17 @@ async def process_events(self, events): await self._checkpoint_manager.update_checkpoint(events[-1].offset, events[-1].sequence_number) +def partition_processor_factory(checkpoint_manager): + return MyPartitionProcessor(checkpoint_manager) + + async def main(): client = EventHubClient.from_connection_string(CONNECTION_STR, receive_timeout=RECEIVE_TIMEOUT, retry_total=RETRY_TOTAL) partition_manager = Sqlite3PartitionManager() try: event_processor = EventProcessor(client, "$default", MyPartitionProcessor, partition_manager) + # You can also define a callable object for creating PartitionProcessor like below: + # event_processor = EventProcessor(client, "$default", partition_processor_factory, partition_manager) asyncio.ensure_future(event_processor.start()) await asyncio.sleep(TEST_DURATION) await event_processor.stop()