Skip to content
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

[Event Hubs] Add a 20% random jitter to the EP load balancing interval. #9713

Merged
merged 1 commit into from
Mar 6, 2020
Merged

[Event Hubs] Add a 20% random jitter to the EP load balancing interval. #9713

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
7 changes: 5 additions & 2 deletions sdk/eventhub/azure-eventhub/azure/eventhub/_eventprocessor/event_processor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
import random
import uuid
import logging
import time
Expand Down Expand Up @@ -224,6 +225,8 @@ def _load_balancing(self):

"""
while self._running:
random_jitter = self._load_balancing_interval * random.random() * 0.2
load_balancing_interval = self._load_balancing_interval + random_jitter
try:
claimed_partition_ids = self._ownership_manager.claim_ownership()
if claimed_partition_ids:
Expand Down Expand Up @@ -266,11 +269,11 @@ def _load_balancing(self):
self._eventhub_name,
self._consumer_group,
err,
self._load_balancing_interval
load_balancing_interval
)
self._process_error(None, err) # type: ignore

time.sleep(self._load_balancing_interval)
time.sleep(load_balancing_interval)

def _close_consumer(self, partition_id, consumer, reason):
# type: (str, EventHubConsumer, CloseReason) -> None
Expand Down
7 changes: 5 additions & 2 deletions sdk/eventhub/azure-eventhub/azure/eventhub/aio/_eventprocessor/event_processor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
import random
from typing import (
Dict,
Callable,
Expand Down Expand Up @@ -310,6 +311,8 @@ async def start(self) -> None:
if not self._running:
self._running = True
while self._running:
random_jitter = self._load_balancing_interval * random.random() * 0.2
Copy link
Contributor

@yunhaoling yunhaoling Feb 4, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems Java gives a fixed base jitter (2 seconds) time and use n * BASE (n in [0,1]) as the final jitter.
https://github.com/Azure/azure-sdk-for-java/blob/master/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventProcessorClient.java#L43

So it's
load_balancing_interval * (1 + ratio) VS load_balancing_interval + ratio * base_jitter

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A fixed base jitter 2 seconds would be too large if the load balancing interval is small. A propotional value is better I think. Java fixes the interval at 10 so the base jitter is 2, not a proportional value.

load_balancing_interval = self._load_balancing_interval + random_jitter
try:
claimed_partition_ids = (
await self._ownership_manager.claim_ownership()
Expand Down Expand Up @@ -355,11 +358,11 @@ async def start(self) -> None:
self._eventhub_name,
self._consumer_group,
err,
self._load_balancing_interval
load_balancing_interval
)
await self._process_error(None, err) # type: ignore

await asyncio.sleep(self._load_balancing_interval, loop=self._loop)
await asyncio.sleep(load_balancing_interval, loop=self._loop)

async def stop(self) -> None:
"""Stop the EventProcessor.
Expand Down