[Event Hubs] Fix memory leak in EHBufferedProducerClient

[HarshaNalluru]

Issue

First reported at #25426, thanks to @pkraj0 for reporting the issue.
EventHubBufferedProducerClient leaks memory when there are no packets to send to event-hubs.

EventHubBufferedProducerClient

• EventHubBufferedProducerClient is used to publish events to Event Hub
• It does not publish events immediately. Instead, events are buffered so they can be efficiently batched and published when the batch is full or the maxWaitTimeInMs has elapsed with no new events enqueued.

Stress Testing

The test sends about 5000 events and stops for a long duration, revealing the memory blowing up.

Investigation

Studying the heap snapshots, and comparing them at various timestamps during the test run hinted at the issue.
The problem is at the BatchingPartitionChannel#_startPublishLoop and the AwaitableQueue#shift implementations.

BatchingPartitionChannel#_startPublishLoop

• Starts the loop that creates batches and sends them to the Event Hub
• Runs forever or until the abortSignal is invoked

AwaitableQueue#shift

• Returns a Promise that will resolve with the next item in the queue.
• If there are no items in the queue, the pending Promise stays forever.

Problem

BatchingPartitionChannel#_startPublishLoop relies on a while loop to run forever.

• The root cause of the leak in the #_startPublishLoop is a race between two promises(AwaitableQueue#shift() and delay())
• AwaitableQueue#shift() never settles because there is no activity.
• Promise.race resolves as soon as one of the promises is fulfilled.
• Promise.race though resolves, keeps a reference for the pending promise from AwaitableQueue#shift(), references get accumulated because of the while loop causing the leak.

Solution

The fix is to abort the pending promise from AwaitableQueue#shift() once the race has been won by the other delay() promise.

What's in the PR?

@azure/core-util

• Added a helper method racePromisesAndAbortLosers, an abstraction that leverages "Promise.race()" and aborts the losers of the race as soon as the first promise fulfills.
• Changelog

@azure/event-hubs

• Updated BatchingPartitionChannel#_startPublishLoop to use racePromisesAndAbortLosers instead of Promise.race.
• Updated AwaitableQueue#shift() to return a promise that is cancellable so that the pending promise cancels once the first promise from Promise.race resolves .

Moved the stress testing to #27184

[azure-sdk]

API change check

APIView has identified API level changes in this PR and created following API reviews.

azure-core-util

[jeremymeng]

• So these pending promises from AwaitableQueue#shift() keep accumulating because of the while loop causing the leak.

In each loop iteration we only get the next one when there's an event; otherwise futureEvent is still the same one. How could it cause the accumulation?

azure-sdk-for-js/sdk/eventhub/event-hubs/src/batchingPartitionChannel.ts

Lines 186 to 188 in c1c67fa

	eventToAddToBatch = event;
	// We received an event, so get a promise for the next one.
	futureEvent = this._eventQueue.shift();

[HarshaNalluru]

/azp run js - event-hubs - tests

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[deyaaeldeen]

Looks great, thanks for the awesome fix!

[xirzec]

One concern with the new core-util code, but otherwise good

[xirzec]

I don't think this will work since the anonymous function () => aborter.abort() will have a different identity than the one registered above. I believe you need to pull it out into a local like

function abortHandler() {
    aborter.abort();
}
options?.abortSignal?.addEventListener("abort", abortHandler);
try {
// etc
} finally {
options?.abortSignal?.removeEventListener("abort", abortHandler);
}

in the future we can use addEventListener with the 'once' option to avoid this dance, but the above will work everywhere.

Maybe this event handler not being unregistered is the source of your small leak?

[HarshaNalluru]

Ah, a great catch, will check that, thanks.

[xirzec]

I'm a little wary of doing math with clocktime in tests unless we're mocking the clock to begin with.

[HarshaNalluru]

remove the delay function for an in-place function with my trackers in it :)
and got rid of the clocks and math, let me know if that looks good

[HarshaNalluru]

/azp run js - event-hubs - tests

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[jeremymeng]

Looks good!