[EventHub] Fix race condition when buffered mode is enabled

[falcaopetri]

Description

This PR implements a fix to the race condition described in #34711.

The approach here is twofold:

1. Require the lock while trying to add new events to internal EventBatch buffer
2. Remove unnecessary clean up code

Implementing either of these would be enough to fix the related issue, but I would argue that both are necessary to make thinks consistent:

• Change 1. helps preventing future race condition issues, since it properly protects the shared resource self._cur_batch. It is especially important to synchronous BufferedProducer, since it relies on multi-threading, which can be even more unpredictable than an event loop.
• Change 2. improves the consistency of self._cur_batch and self._cur_buffered_len:
  • _cur_batch is "reset" (i.e., assigned to a new EventBatch object) when and only when it gets enqueued to self._buffered_queue
  • _cur_buffered_len is decreased solely as a consequence of a self._producer.send. PR [Eventhub] Fix Blocking Behavior of Buffered Producer Flush #25406 previously introduced a manual reset to this number, but it did not consider the timeout-based exit condition:
    

    azure-sdk-for-python/sdk/eventhub/azure-eventhub/azure/eventhub/aio/_buffered_producer/_buffered_producer_async.py

    Lines 190 to 200 in 23121a5

    	else:
    	_LOGGER.info(
    	"Partition %r fails to flush due to timeout.", self.partition_id
    	)
    	if raise_error:
    	raise OperationTimeoutError(
    	"Failed to flush {!r} within {}".format(
    	self.partition_id, timeout_time
    	)
    	)
    	break

A last note regarding Change 1.:

• it seems that the azure-sdk-for-net also updates the self._cur_batch with a synchronization tool, see RunPublishingAsync

I also have another small set of changes that could make sync and async BufferedProducer codebase more consistent. Since they do not have an impact on functionality, I thought it would be better to submit them in follow up PR.

All SDK Contribution checklist:

• The pull request does not introduce [breaking changes]
• CHANGELOG is updated for new features, bug fixes or other significant changes.
• I have read the contribution guidelines.

General Guidelines and Best Practices

• Title of the pull request is clear and informative.
• There are a small number of commits, each of which have an informative message. This means that previously merged commits do not appear in the history of the PR. For more information on cleaning up the commits in your PR, see this page.

Testing Guidelines

• Pull request includes test coverage for the included changes.

It would indeed be interesting to have some test code here, though I think it's hard to make determinist tests check this potential race condition. We would think something based on this test, though it is currently skipped:

azure-sdk-for-python/sdk/eventhub/azure-eventhub/tests/livetest/asynctests/test_buffered_producer_async.py

Lines 529 to 532 in 23121a5

	@pytest.mark.skip('not testing correctly + flaky, fix during MQ')
	@pytest.mark.liveTest
	@pytest.mark.asyncio
	async def test_long_wait_small_buffer(connection_str, uamqp_transport):

[github-actions]

Thank you for your contribution @falcaopetri! We will review the pull request and get back to you soon.

[falcaopetri]

@microsoft-github-policy-service agree

[kashifkhan]

/azp run python - eventhub - tests

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[annatisch]

Thank you so much for taking the time to add this @falcaopetri! Much appreciated :)