Skip to content

[EventHubs] exception/kwargs testing #26867

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

[EventHubs] exception/kwargs testing #26867

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f92c51d
fix flushing list batches in bp
swathipil Oct 17, 2022
c78177f
add send list bp test
swathipil Oct 17, 2022
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
10 changes: 8 additions & 2 deletions sdk/eventhub/azure-eventhub/azure/eventhub/_buffered_producer/_buffered_producer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,15 +165,21 @@ def flush(self, timeout_time=None, raise_error=True):
self.partition_id,
len(batch),
)
self._on_success(batch._internal_events, self.partition_id)
try:
self._on_success(batch._internal_events, self.partition_id)
except AttributeError:
self._on_success(batch, self.partition_id)
except Exception as exc: # pylint: disable=broad-except
_LOGGER.info(
"Partition %r sending %r events failed due to exception: %r ",
self.partition_id,
len(batch),
exc,
)
self._on_error(batch._internal_events, self.partition_id, exc)
try:
self._on_error(batch._internal_events, self.partition_id, exc)
except AttributeError:
self._on_error(batch, self.partition_id, exc)
finally:
self._cur_buffered_len -= len(batch)
else:
Expand Down
10 changes: 8 additions & 2 deletions ...eventhub/azure-eventhub/azure/eventhub/aio/_buffered_producer/_buffered_producer_async.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,15 +167,21 @@ async def _flush(self, timeout_time=None, raise_error=True):
self.partition_id,
len(batch),
)
await self._on_success(batch._internal_events, self.partition_id)
try:
await self._on_success(batch._internal_events, self.partition_id)
except AttributeError:
await self._on_success(batch, self.partition_id)
except Exception as exc: # pylint: disable=broad-except
_LOGGER.info(
"Partition %r sending %r events failed due to exception: %r",
self.partition_id,
len(batch),
exc,
)
await self._on_error(batch._internal_events, self.partition_id, exc)
try:
await self._on_error(batch._internal_events, self.partition_id, exc)
except AttributeError:
await self._on_error(batch, self.partition_id, exc)
finally:
self._cur_buffered_len -= len(batch)
# If flush could not get the semaphore, we log and raise error if wanted
Expand Down
2 changes: 1 addition & 1 deletion sdk/eventhub/azure-eventhub/tests/livetest/asynctests/test_buffered_producer_async.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -407,7 +407,7 @@ async def on_error(events, pid, err):

async with producer:
partitions = await producer.get_partition_ids()
await producer.send_event(EventData('data'))
await producer.send_batch([EventData('data')])
await asyncio.sleep(5)
assert not sent_events
await asyncio.sleep(20)
Expand Down
2 changes: 1 addition & 1 deletion sdk/eventhub/azure-eventhub/tests/livetest/synctests/test_buffered_producer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -419,7 +419,7 @@ def on_error(events, pid, err):

with producer:
partitions = producer.get_partition_ids()
producer.send_event(EventData('data'))
producer.send_batch([EventData('data')])
time.sleep(5)
assert not sent_events
time.sleep(10)
Expand Down