Refine exception handling for eventprocessor

sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/event_processor.py

@@ -11,7 +11,7 @@
 from azure.eventhub import EventPosition, EventHubError
 from azure.eventhub.aio import EventHubClient
 from .checkpoint_manager import CheckpointManager
-from .partition_manager import PartitionManager
+from .partition_manager import PartitionManager, OwnershipLostError
 from ._ownership_manager import OwnershipManager
 from .partition_processor import CloseReason, PartitionProcessor
 
@@ -182,13 +182,43 @@ async def _receive(self, ownership):
             EventPosition(ownership.get("offset", self._initial_event_position))
         )
 
+        async def process_error(err):
+            log.error(
+                "PartitionProcessor of EventProcessor instance %r of eventhub %r partition %r consumer group %r"
+                " has met an error. The exception is %r.",
+                owner_id, eventhub_name, partition_id, consumer_group_name, err
+            )
+            try:
+                await partition_processor.process_error(err, checkpoint_manager)
+            except Exception as err_again:  # pylint:disable=broad-except
+                log.error(
+                    "PartitionProcessor of EventProcessor instance %r of eventhub %r partition %r consumer group %r"
+                    " has another error during running process_error(). The exception is %r.",
+                    owner_id, eventhub_name, partition_id, consumer_group_name, err_again
+                )
+
+        async def close(reason):
+            log.info(
+                "PartitionProcessor of EventProcessor instance %r of eventhub %r partition %r consumer group %r"
+                " is being closed. Reason is: %r",
+                owner_id, eventhub_name, partition_id, consumer_group_name, reason
+            )
+            try:
+                await partition_processor.close(reason, checkpoint_manager)
+            except Exception as err:  # pylint:disable=broad-except
+                log.error(
+                    "PartitionProcessor of EventProcessor instance %r of eventhub %r partition %r consumer group %r"
+                    " has an error during running close(). The exception is %r.",
+                    owner_id, eventhub_name, partition_id, consumer_group_name, err
+                )
+
         try:
             while True:
                 try:
                     await partition_processor.initialize()
                     events = await partition_consumer.receive(timeout=self._receive_timeout)
                     await partition_processor.process_events(events, checkpoint_manager)
-                except asyncio.CancelledError as cancelled_error:
+                except asyncio.CancelledError:
                     log.info(
                         "PartitionProcessor of EventProcessor instance %r of eventhub %r partition %r consumer group %r"
                         " is cancelled",
@@ -197,28 +227,25 @@ async def _receive(self, ownership):
                         partition_id,
                         consumer_group_name
                     )
-                    await partition_processor.process_error(cancelled_error, checkpoint_manager)
-                    await partition_processor.close(CloseReason.SHUTDOWN, checkpoint_manager)
-                    # TODO: release the ownership immediately via partition manager
+                    if self._running is False:
+                        await close(CloseReason.SHUTDOWN)
+                    else:
+                        await close(CloseReason.OWNERSHIP_LOST)
+                    # TODO: release the ownership immediately via partition manager in preview 4
                     break
                 except EventHubError as eh_err:
-                    reason = CloseReason.OWNERSHIP_LOST if eh_err.error == "link:stolen" \
-                        else CloseReason.EVENTHUB_EXCEPTION
-                    log.warning(
-                        "PartitionProcessor of EventProcessor instance %r of eventhub %r partition %r consumer group %r"
-                        " has met an exception receiving events. It's being closed. The exception is %r.",
-                        owner_id,
-                        eventhub_name,
-                        partition_id,
-                        consumer_group_name,
-                        eh_err
-                    )
-                    await partition_processor.process_error(eh_err, checkpoint_manager)
-                    await partition_processor.close(reason, checkpoint_manager)
+                    await process_error(eh_err)
+                    await close(CloseReason.EVENTHUB_EXCEPTION)
                     # An EventProcessor will pick up this partition again after the ownership is released
-                    # TODO: release the ownership immediately via partition manager
+                    # TODO: release the ownership immediately via partition manager in preview 4
+                    break
+                except OwnershipLostError:
+                    await close(CloseReason.OWNERSHIP_LOST)
+                    break
+                except Exception as other_error:  # pylint:disable=broad-except
+                    await process_error(other_error)
+                    await close(CloseReason.PROCESS_EVENTS_ERROR)
+                    # TODO: release the ownership immediately via partition manager in preview 4
                     break
-                except Exception as exp:  # pylint:disable=broad-except
-                    log.warning(exp)
         finally:
             await partition_consumer.close()

sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_manager.py

@@ -77,8 +77,15 @@ async def update_checkpoint(self, eventhub_name, consumer_group_name, partition_
         :param sequence_number: The sequence_number of the ~azure.eventhub.EventData the new checkpoint
         will be associated with.
         :type sequence_number: int
-        :return:
+        :return: None
+        :raise: `OwnershipLostError`, `CheckpointError`
         """
 
     async def close(self):
         pass
+
+
+class OwnershipLostError(Exception):
+    """Raises when update_checkpoint detects the ownership has been lost
+
+    """

sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/partition_processor.py

@@ -14,6 +14,7 @@ class CloseReason(Enum):
     SHUTDOWN = 0  # user call EventProcessor.stop()
     OWNERSHIP_LOST = 1  # lose the ownership of a partition.
     EVENTHUB_EXCEPTION = 2  # Exception happens during receiving events
+    PROCESS_EVENTS_ERROR = 3  # Exception happens during process_events
 
 
 class PartitionProcessor(ABC):

sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/sqlite3_partition_manager.py

@@ -7,7 +7,7 @@
 import uuid
 import sqlite3
 import logging
-from .partition_manager import PartitionManager
+from .partition_manager import PartitionManager, OwnershipLostError
 
 logger = logging.getLogger(__name__)
 
@@ -134,6 +134,7 @@ async def update_checkpoint(self, eventhub_name, consumer_group_name, partition_
             else:
                 logger.info("EventProcessor couldn't checkpoint to partition %r because it no longer has the ownership",
                             partition_id)
+                raise OwnershipLostError()
 
         finally:
             cursor.close()