Skip to content

[EventHubs] Performance improvement #7533

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
yunhaoling merged 7 commits into from
Oct 2, 2019
Merged

[EventHubs] Performance improvement #7533

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
41cce74
performance improvement
yunhaoling Sep 29, 2019
25fa5d7
batch offset
yunhaoling Sep 29, 2019
24a181d
merge eventhub-preview4
yunhaoling Oct 1, 2019
647b256
Review feedback
yunhaoling Oct 2, 2019
64c7152
Remove lazy parse in EventData
yunhaoling Oct 2, 2019
3952b51
Merge branch 'eventhubs_preview4' into performance-improvement
yunhaoling Oct 2, 2019
803bc1e
Add annotation assertions
yunhaoling Oct 2, 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
Review feedback
  • Loading branch information
@yunhaoling
yunhaoling committed Oct 2, 2019
commit 647b2561318ae127870b0ca503b09b5fbe09a7f9
4 changes: 2 additions & 2 deletions sdk/eventhub/azure-eventhubs/azure/eventhub/aio/consumer_async.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,7 +119,7 @@ async def __anext__(self):
self._offset = EventPosition(event_data.offset, inclusive=False)
retried_times = 0
if self._track_last_enqueued_event_properties:
self._last_enqueued_event_properties = event_data._get_runtime_info() # pylint:disable=protected-access
self._last_enqueued_event_properties = event_data._get_last_enqueued_event_properties() # pylint:disable=protected-access
return event_data
except Exception as exception: # pylint:disable=broad-except
last_exception = await self._handle_exception(exception)
Expand Down Expand Up @@ -209,7 +209,7 @@ async def _receive(self, timeout_time=None, max_batch_size=None, **kwargs):
self._offset = EventPosition(data_batch[-1].offset)

if self._track_last_enqueued_event_properties and len(data_batch):
self._last_enqueued_event_properties = data_batch[-1]._get_runtime_info() # pylint:disable=protected-access
self._last_enqueued_event_properties = data_batch[-1]._get_last_enqueued_event_properties() # pylint:disable=protected-access

return data_batch

Expand Down
23 changes: 6 additions & 17 deletions sdk/eventhub/azure-eventhubs/azure/eventhub/common.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@ def __init__(self, body=None, to_device=None):
self._delivery_annotations = {}
self._app_properties = {}
self._msg_properties = MessageProperties()
self._runtime_info = {}
self._last_enqueued_event_properties = {}
self._need_further_parse = False
if to_device:
self._msg_properties.to = '/devices/{}/messages/devicebound'.format(to_device)
Expand Down Expand Up @@ -153,12 +153,12 @@ def _trace_link_message(self, parent_span=None):
if traceparent:
current_span.link(traceparent)

def _get_runtime_info(self):
if self._runtime_info:
return self._runtime_info
def _get_last_enqueued_event_properties(self):
if self._last_enqueued_event_properties:
return self._last_enqueued_event_properties

if self.message.delivery_annotations:
self._runtime_info = {
self._last_enqueued_event_properties = {
"sequence_number":
self.message.delivery_annotations.get(EventData.PROP_LAST_ENQUEUED_SEQUENCE_NUMBER, None),
"offset":
Expand All @@ -168,7 +168,7 @@ def _get_runtime_info(self):
"retrieval_time":
self.message.delivery_annotations.get(EventData.PROP_RUNTIME_INFO_RETRIEVAL_TIME_UTC, None)
}
return self._runtime_info
return self._last_enqueued_event_properties

return None

Expand All @@ -185,17 +185,6 @@ def _parse_message_properties(self):
self._annotations = self.message.annotations
self._app_properties = self.message.application_properties
self._delivery_annotations = self.message.delivery_annotations
if self._delivery_annotations:
self._runtime_info = {
"sequence_number":
self._delivery_annotations.get(EventData.PROP_LAST_ENQUEUED_SEQUENCE_NUMBER, None),
"offset":
self._delivery_annotations.get(EventData.PROP_LAST_ENQUEUED_OFFSET, None),
"enqueued_time":
self._delivery_annotations.get(EventData.PROP_LAST_ENQUEUED_TIME_UTC, None),
"retrieval_time":
self._delivery_annotations.get(EventData.PROP_RUNTIME_INFO_RETRIEVAL_TIME_UTC, None)
}
self._need_further_parse = False

@property
Expand Down
4 changes: 2 additions & 2 deletions sdk/eventhub/azure-eventhubs/azure/eventhub/consumer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,7 +115,7 @@ def __next__(self):
self._offset = EventPosition(event_data.offset, inclusive=False)
retried_times = 0
if self._track_last_enqueued_event_properties:
self._last_enqueued_event_properties = event_data._get_runtime_info() # pylint:disable=protected-access
self._last_enqueued_event_properties = event_data._get_last_enqueued_event_properties() # pylint:disable=protected-access
return event_data
except Exception as exception: # pylint:disable=broad-except
last_exception = self._handle_exception(exception)
Expand Down Expand Up @@ -203,7 +203,7 @@ def _receive(self, timeout_time=None, max_batch_size=None, **kwargs):
self._offset = EventPosition(data_batch[-1].offset)

if self._track_last_enqueued_event_properties and len(data_batch):
self._last_enqueued_event_properties = data_batch[-1]._get_runtime_info() # pylint:disable=protected-access
self._last_enqueued_event_properties = data_batch[-1]._get_last_enqueued_event_properties() # pylint:disable=protected-access

return data_batch

Expand Down