Skip to content

Python EventHubs load balancing #6901

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
YijunXieMS merged 43 commits into from
Sep 7, 2019
Merged

Python EventHubs load balancing #6901

Changes from 1 commit
Commits
Show all changes
43 commits
Select commit Hold shift + click to select a range
3a32907
Draft EventProcessor Loadbalancing
Aug 22, 2019
39b1b86
EventProcessor Load balancing
Aug 22, 2019
17f5153
small changes from bryan's review
Aug 23, 2019
04ef548
remove checkpoint manager from initialize
Aug 23, 2019
9be1741
small changes
Aug 23, 2019
1b5753c
Draft EventProcessor Loadbalancing
Aug 22, 2019
b4b77f9
EventProcessor Load balancing
Aug 22, 2019
1787fdd
small changes from bryan's review
Aug 23, 2019
c2d0155
remove checkpoint manager from initialize
Aug 23, 2019
1074385
small changes
Aug 23, 2019
386baf0
Fix code review feedback
Aug 29, 2019
1afbf0c
Merge branch 'eventhubs_yx' of github.com:Azure/azure-sdk-for-python …
Aug 30, 2019
c126bea
Packaging update of azure-mgmt-datalake-analytics
AutorestCI Aug 30, 2019
40c7f03
Packaging update of azure-loganalytics
AutorestCI Aug 30, 2019
cf22c7c
Packaging update of azure-mgmt-storage
AutorestCI Aug 30, 2019
c7440b2
Merge branch 'eventhubs_preview3' into eventhubs_yx
Aug 30, 2019
fa804f4
code review fixes and pylint error
Aug 30, 2019
470cf7e
Merge branch 'eventhubs_yx' of github.com:Azure/azure-sdk-for-python …
Aug 30, 2019
e5f3b50
reduce dictionary access
Aug 30, 2019
8343876
Revert "Packaging update of azure-mgmt-storage"
Sep 2, 2019
66c5b31
Revert "Packaging update of azure-loganalytics"
Sep 2, 2019
bcd851a
Revert "Packaging update of azure-mgmt-datalake-analytics"
Sep 2, 2019
d740bb0
Trivial code change
Sep 2, 2019
aad6978
Refine exception handling for eventprocessor
Sep 3, 2019
a55dc13
Enable pylint for eventprocessor
Sep 3, 2019
a339985
Expose OwnershipLostError
Sep 3, 2019
9102713
Move eventprocessor to aio
Sep 4, 2019
278592c
change checkpoint_manager to partition context
Sep 4, 2019
665f28c
fix pylint error
Sep 4, 2019
0060f9d
fix a small issue
Sep 4, 2019
7b4273a
Catch list_ownership/claim_ownership exceptions and retry
Sep 5, 2019
bdf97c8
Fix code review issues
Sep 6, 2019
02a4daf
fix event processor long running test
Sep 6, 2019
a9446de
Remove utils.py
Sep 6, 2019
8dfdec9
Remove close() method
Sep 6, 2019
2aace82
Updated docstrings
Sep 6, 2019
36ba0a3
add pytest
Sep 7, 2019
7f95d9e
small fixes
Sep 7, 2019
f5870af
Merge branch 'eventhubs_preview3' into eventhubs_yx
Sep 7, 2019
f30d143
Revert "Remove utils.py"
Sep 7, 2019
893bee0
change asyncio.create_task to 3.5 friendly code
Sep 7, 2019
4b41fa5
Remove Callable
Sep 7, 2019
fef0551
raise CancelledError instead of break
Sep 7, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
small changes
  • Loading branch information
yijxie committed Aug 23, 2019
commit 9be17413316d4bf556472e950b3bfbeb56df166b
22 changes: 13 additions & 9 deletions sdk/eventhub/azure-eventhubs/azure/eventhub/eventprocessor/event_processor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,15 +164,19 @@ async def _receive(self, ownership):
await self.stop()
raise TypeError("Partition processor must has method process_events(events, checkpoint_manager")

partition_consumer = self._eventhub_client.create_consumer(ownership["consumer_group_name"],
ownership["partition_id"],
EventPosition(ownership.get("offset", self._initial_event_position))
)
checkpoint_manager = CheckpointManager(ownership["partition_id"],
ownership["eventhub_name"],
ownership["consumer_group_name"],
ownership["owner_id"],
self._partition_manager)
partition_consumer = self._eventhub_client.create_consumer(
ownership["consumer_group_name"],
ownership["partition_id"],
EventPosition(ownership.get("offset", self._initial_event_position))
)
checkpoint_manager = CheckpointManager(
ownership["partition_id"],
ownership["eventhub_name"],
ownership["consumer_group_name"],
ownership["owner_id"],
self._partition_manager
)

async def initialize():
if hasattr(partition_processor, "initialize"):
await partition_processor.initialize()
Expand Down