create_batch feature implementation (#6256) #6324

yunhaoling · 2019-07-12T00:25:17Z

Init create batch event
create_batch implementation
Revert _set_partition_key method and update comment
Refacor EventDataBatch class
Revert logic when setting partition_key of event_data

* Init create batch event * create_batch implementation * Revert _set_partition_key method and update comment * Refacor EventDataBatch class * Revert logic when setting partition_key of event_data

adxsdk6 · 2019-07-12T00:25:18Z

Can one of the admins verify this patch?

annatisch

Added a couple of comments to the async clients - the same clients apply to the sync ones but I haven't duplicated them ;)

annatisch · 2019-07-16T13:50:31Z

sdk/eventhub/azure-eventhubs/azure/eventhub/aio/producer_async.py

@@ -110,6 +111,10 @@ def __init__(  # pylint: disable=super-init-not-called
            await self._connect()
            self.running = True

+            self._max_message_size_on_link = self._handler.message_handler._link.peer_max_message_size if\
+                self._handler.message_handler._link.peer_max_message_size\
+                else constants.MAX_MESSAGE_LENGTH_BYTES


Minor: This statement is a bit of mouthful - I think it would be cleaner just to do it with an "or" statement:

self._max_message_size = self._handler.message_handler._link.peer_max_message_size or constants.MAX_MESSAGE_LENGTH

Also I think it will need a pylint disable for protected-access.

good suggestion. Fixed

You fixed it in the sync class but not the async one :)

annatisch · 2019-07-16T13:52:56Z

sdk/eventhub/azure-eventhubs/azure/eventhub/aio/producer_async.py

+            await self._open()
+
+        if max_message_size and max_message_size > self._max_message_size_on_link:
+            raise EventDataError('Max message size: {} is too large, acceptable max batch size is: {} bytes.'


I think this should just be a ValueError

changed to ValueError

annatisch · 2019-07-16T13:53:43Z

sdk/eventhub/azure-eventhubs/azure/eventhub/aio/producer_async.py

+            raise EventDataError('Max message size: {} is too large, acceptable max batch size is: {} bytes.'
+                                 .format(max_message_size, self._max_message_size_on_link))
+
+        return EventDataBatch(max_message_size if max_message_size else self._max_message_size_on_link, partition_key)


This can also be done with an "or" statement:
EventDataBatch(max_message_size or self._max_message_size....)

changed to use "or"

annatisch · 2019-07-16T13:54:19Z

sdk/eventhub/azure-eventhubs/azure/eventhub/aio/producer_async.py

+                                 .format(max_message_size, self._max_message_size_on_link))
+
+        return EventDataBatch(max_message_size if max_message_size else self._max_message_size_on_link, partition_key)
+
    async def send(self, event_data, partition_key=None):
        # type:(Union[EventData, Union[List[EventData], Iterator[EventData], Generator[EventData]]], Union[str, bytes]) -> None


Type hint needs updating to include the EventDataBatch (assuming it can be passed in here)

Updated to
type:(Union[EventData, EventDataBatch, Iterable[EventData]], Union[str, bytes]) -> None

annatisch · 2019-07-16T13:55:17Z

sdk/eventhub/azure-eventhubs/azure/eventhub/aio/producer_async.py

-                event_data_with_pk,
-                partition_key=partition_key) if partition_key else _BatchSendEventData(event_data)
+            if isinstance(event_data, EventDataBatch):
+                wrapper_event_data = event_data


Does this mean that in the case of EventDataBatch we ignore the passed in partition key?

Yes, EventDataBatch wins. It also has partition_key.

In that case we should raise an error if a user tries to pass in the partition_key alongside a Batch object. Otherwise they might think the value they pass in being used when it's actually being ignored.

annatisch · 2019-07-16T13:57:22Z

sdk/eventhub/azure-eventhubs/azure/eventhub/common.py

        self._set_partition_key(partition_key)
+        self._size = self.message.gather()[0].get_message_encoded_size()


Can you please double check the uAMQP code for .gather()? I have a suspicion that it's not a repeatable operation.... though I could be wrong....

It works in our case

annatisch · 2019-07-16T13:58:25Z

sdk/eventhub/azure-eventhubs/azure/eventhub/common.py

@@ -260,6 +279,35 @@ def _set_partition_key(self, value):
            self.message.annotations = annotations
            self.message.header = header

+    def try_add(self, event_data):


Where is this function used?

The try_add function is part of the API and supposed to be used by customers not by us.

Other small fixes

bgklein · 2019-07-23T20:18:56Z

Improperly routed to batch. Concerns batching of event hub events. @annatisch if you wanna review.

annatisch · 2019-07-25T17:29:29Z

sdk/eventhub/azure-eventhubs/azure/eventhub/aio/producer_async.py

@@ -110,6 +111,10 @@ def __init__(  # pylint: disable=super-init-not-called
            await self._connect()
            self.running = True

+            self._max_message_size_on_link = self._handler.message_handler._link.peer_max_message_size if\
+                self._handler.message_handler._link.peer_max_message_size\
+                else constants.MAX_MESSAGE_LENGTH_BYTES


You fixed it in the sync class but not the async one :)

annatisch · 2019-07-25T17:32:18Z

sdk/eventhub/azure-eventhubs/azure/eventhub/aio/producer_async.py

@@ -301,8 +306,25 @@ def _set_partition_key(event_datas, partition_key):
            ed._set_partition_key(partition_key)
            yield ed

+    async def create_batch(self, max_size=None, partition_key=None):


I wonder if this function needs to be async.
I think it might be better to raise an error if this is called before the link is open, rather than have it open the connection.
Then the function doesn't need to be async and can simply return the Batch object.

As we're not exposing the producer.open api to user, if users want to use create_batch before they send, we need to open the connection for them internally to get the right peer link message size.

annatisch · 2019-07-25T17:33:15Z

sdk/eventhub/azure-eventhubs/azure/eventhub/aio/producer_async.py

-                event_data_with_pk,
-                partition_key=partition_key) if partition_key else _BatchSendEventData(event_data)
+            if isinstance(event_data, EventDataBatch):
+                wrapper_event_data = event_data


In that case we should raise an error if a user tries to pass in the partition_key alongside a Batch object. Otherwise they might think the value they pass in being used when it's actually being ignored.

annatisch · 2019-07-25T17:34:25Z

sdk/eventhub/azure-eventhubs/azure/eventhub/producer.py

+        :return:
+        """
+        if not self._max_message_size_on_link:
+            self._open()


If we change the behaviour for the async create_batch (to raise an error if the connection isn't open) we will need to do the same thing here.

Same to the above async one. We should open internally first to get the right peer link message size.

annatisch · 2019-07-25T17:35:28Z

sdk/eventhub/azure-eventhubs/azure/eventhub/producer.py

@@ -298,16 +302,34 @@ def _error(outcome, condition):
        if outcome != constants.MessageSendResult.Ok:
            raise condition

+    def create_batch(self, max_size=None, partition_key=None):
+        """
+        Create an EventDataBatch object with max message size being max_message_size.


The parameter is called max_size not max_message_size

Parameter name and description updated.

annatisch · 2019-07-25T17:35:51Z

sdk/eventhub/azure-eventhubs/azure/eventhub/producer.py

+        Create an EventDataBatch object with max message size being max_message_size.
+        The max_message_size should be no greater than the max allowed message size defined by the service side.
+        :param max_message_size:
+        :param partition_key:


Could we add types and descriptions of the parameters.

Types and descriptions added.

annatisch · 2019-07-25T17:36:42Z

sdk/eventhub/azure-eventhubs/azure/eventhub/producer.py

-                event_data_with_pk,
-                partition_key=partition_key) if partition_key else _BatchSendEventData(event_data)
+            if isinstance(event_data, EventDataBatch):  # The partition_key in the param will be omitted.
+                wrapper_event_data = event_data


Same comment as above regarding raising an error if partition_key is going to be ignored.

Code updated, an EventDataError will be raised if the keys don't match.

annatisch · 2019-07-25T17:37:15Z

sdk/eventhub/azure-eventhubs/azure/eventhub/common.py

+    Do not instantiate an EventDataBatch object directly.
+    """
+
+    log = logging.getLogger(__name__)


This needs to be at the module level, not class.

Code updated.

annatisch · 2019-07-25T17:38:29Z

sdk/eventhub/azure-eventhubs/azure/eventhub/common.py

+    log = logging.getLogger(__name__)
+
+    def __init__(self, max_size=None, partition_key=None):
+        self.max_size = max_size if max_size else constants.MAX_MESSAGE_LENGTH_BYTES


This can be a simple or-statement:
self.max_size = max_size or contants.MAX_MESSAGE_LENGTH_BYTES

thanks for the suggestion, code updated.

annatisch · 2019-07-25T17:41:27Z

sdk/eventhub/azure-eventhubs/azure/eventhub/common.py

+        :return:
+        """
+        if event_data is None:
+            self.log.warning("event_data is None when calling EventDataBatch.try_add. Ignored")


Should use module-level logger

Code updated.

* Update according to the review * Update comment

…g changes

* Change back to normal number writings as not supported by python under 3.6 * small fix

* Eventhubs conn (#6394) * Shared connection (sync) draft * Shared connection (sync) draft 2 * Shared connection (sync) test update * Shared connection * Fix an issue * add retry exponential delay and timeout to exception handling * put module method before class def * fixed Client.get_properties error * Improve send timeout (#6481) * Add timeout information to the link prop during link creation process (#6485) * Update optional parameters in public api into kwargs and update comments (#6510) * Update optional parameters in public api into kwargs and update some comments * Update more optional parameters into kwargs paramter * create_batch feature implementation (#6256) (#6324) * create_batch feature implementation (#6256) * Init create batch event * create_batch implementation * Revert _set_partition_key method and update comment * Refacor EventDataBatch class * Revert logic when setting partition_key of event_data * Misc fixes from code review * Rename max_message_size to max_size Other small fixes * Warn if event_data is None when call try_add * Create batch event update (#6509) * Update according to the review * Update comment * Revert some kwargs backto optional parameters as it may cause breaking changes * Small fixes (#6520) * Change back to normal number writings as not supported by python under 3.6 * small fix * Add missing return (#6522) * Fix livetest (#6523) * Eventhubs new EventProcessor (#6550) * Shared connection (sync) draft * Shared connection (sync) draft 2 * Shared connection (sync) test update * Shared connection * Fix an issue * add retry exponential delay and timeout to exception handling * put module method before class def * fixed Client.get_properties error * new eph (draft) * new eph (draft2) * remove in memory partition manager * EventProcessor draft 3 * small format change * Fix logging * Add EventProcessor example * use decorator to implement retry logic and update some tests (#6544) * Update livetest (#6547) * Remove legacy code and update livetest (#6549) * Update livetest * Remove legacy code and update livetest * make sync longrunning multi-threaded * small changes on async long running test * reset retry_count for iterator * Don't return early when open a ReceiveClient or SendClient * type annotation change * Update kwargs and remove unused import * Misc changes from EventProcessor PR review * raise asyncio.CancelledError out instead of supressing it. * Update livetest and small fixed (#6594) * Add missing close in livetest * Update livetest to wait longer * Close handler each time before retry * Fix feedback from PR (1) * Revert "Merge branch 'eventhubs_dev' into eventhubs_eph" This reverts commit 19a5539, reversing changes made to 9d18dd9. * Fix feedback from PR (2) * Update code according to the review (#6623) * Wait longer for reconnect op * Raise authentication error when open timeout * Optimize retry decorator * Update code according to review * Small fix * Fix feedback from PR (3) * small bug fixing * Remove old EPH * Update decorator implementation (#6642) * Update decorator implementation * Remove old EPH pytest * Revert "Revert "Merge branch 'eventhubs_dev' into eventhubs_eph"" This reverts commit d688090. * Update sample codes and docstring (#6643) * Check tablename to prevent sql injection * PR review update * Removed old EPH stuffs. Added new EPH stuffs. * Small fix (#6650) * Draft for changelog * Improve syntax for kwargs * keep partition manager open for call restart() again * Example to process async operations * Update version to 5.0.0b2 * fix mypy problem * fix small issue on max_retries * compatible with python < 3.7 * Update docstring of event processor * small changes on max_retries * small changes on max_retries * small changes * new EventProcessor long-running live test * change offset to text * Updating docstings, docs, samples (#6673) * Draft for updating documentations * Small improvement * Updating docstrings, docs and sample * support 3.5 type hint * fix 3.5 compatibility * Update docs (#6678) * Add a run_awhile example function * small fix on example * small fix on example * Update documentations (#6694) * Small update (#6696)

create_batch feature implementation (#6256)

2d22a1a

* Init create batch event * create_batch implementation * Revert _set_partition_key method and update comment * Refacor EventDataBatch class * Revert logic when setting partition_key of event_data

yunhaoling requested review from annatisch and YijunXieMS July 12, 2019 00:25

annatisch reviewed Jul 16, 2019

View reviewed changes

kurtzeborn added Batch Mgmt This issue is related to a management-plane library. labels Jul 16, 2019

yijxie added 3 commits July 17, 2019 15:28

Misc fixes from code review

11b3333

Rename max_message_size to max_size

4f310ea

Other small fixes

Warn if event_data is None when call try_add

47df21b

bgklein added Event Hubs and removed Batch labels Jul 23, 2019

annatisch reviewed Jul 25, 2019

View reviewed changes

Create batch event update (#6509)

dfd4e39

* Update according to the review * Update comment

yunhaoling force-pushed the eventhubs_dev branch from aece9c2 to dfd4e39 Compare July 27, 2019 02:29

yunhaoling and others added 5 commits July 26, 2019 19:55

Merged shared connection

7e58a0c

Revert some kwargs backto optional parameters as it may cause breakin…

fc69d42

…g changes

Small fixes (#6520)

e4be05e

* Change back to normal number writings as not supported by python under 3.6 * small fix

Add missing return (#6522)

e702c76

Fix livetest (#6523)

aa56698

yunhaoling merged commit 5bfd08f into eventhubs_preview2 Jul 29, 2019

abdulrahman305 mentioned this pull request Oct 11, 2024

[Snyk] Security upgrade cspell from 6.31.2 to 8.15.0 abdulrahman305/azure-sdk-for-python#18

Open

		self._set_partition_key(partition_key)
		self._size = self.message.gather()[0].get_message_encoded_size()

create_batch feature implementation (#6256) #6324

create_batch feature implementation (#6256) #6324

Conversation

yunhaoling commented Jul 12, 2019

adxsdk6 commented Jul 12, 2019

annatisch left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

YijunXieMS Jul 17, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bgklein commented Jul 23, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yunhaoling Jul 26, 2019 • edited Loading

Choose a reason for hiding this comment

YijunXieMS Jul 17, 2019 •

edited

Loading

yunhaoling Jul 26, 2019 •

edited

Loading