[event-hubs] processEvents ignoring maxWaitTime after retryable disconnect event

[chradek]

Summary of issue

The subscription created by EventHubConsumerClient.subscribe() can get into a bad state where the processEvents handler is invoked rapidly with 0 events after a connection is disconnected due to a retryable error. When in this state, the maxWaitTimeInSeconds parameter is ignored and instead the processEvents handler is called as quickly as possible.

Steps to reproduce

The following script reproduces the issue by forcing the underlying connection's idle to fire which triggers a retryable disconnected event:

const { EventHubConsumerClient } = require('@azure/event-hubs');

const connectionString = "";
const eventHubName = "";

const client = new EventHubConsumerClient(EventHubConsumerClient.defaultConsumerGroupName, connectionString, eventHubName);

const clientConnectionContext = client["_context"];

let shouldDisconnect = true;

const subscription = client.subscribe("0", {
  async processError(error, context) {
  },
  async processEvents(events, context) {
    if (shouldDisconnect) {
      clientConnectionContext.connection["_connection"].idle();
    }
    shouldDisconnect = false;
  }
}, {
  maxBatchSize: 100,
  maxWaitTimeInSeconds: 5
});

Root cause

When a connection receives a disconnected event, the event-hubs library creates a new amqp connection and closes the links on the existing connection:

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

Lines 333 to 337 in 3f766f0

	for (const receiverName of Object.keys(connectionContext.receivers)) {
	const receiver = connectionContext.receivers[receiverName];
	await receiver.close().catch(() => {
	/* error already logged, swallow it here */
	});

Inside an already running subscription, receiveBatch is called on a loop until either a non-retryable error is encountered, or the subscription is closed:

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

Lines 75 to 81 in 3f766f0

	while (this._isReceiving) {
	try {
	const receivedEvents = await this._receiver.receiveBatch(
	this._processorOptions.maxBatchSize,
	this._processorOptions.maxWaitTimeInSeconds,
	this._abortController.signal
	);

The receiveBatch call does a check to see if the receiver has been closed, and returns any events that have been collected thus far:

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

Lines 677 to 679 in 3f766f0

	if (this._isClosed || this._context.wasConnectionCloseCalled) {
	return resolve(receivedEvents);
	}

Thus, when a disconnected event is triggered with a retryable error, the loop that calls receiveBatch doesn't exit, and receiveBatch returns immediately each time it is invoked.

Why didn't existing tests catch this?

We do have a test that is supposed to ensure that processEvents is called after a disconnected event occurs:

azure-sdk-for-js/sdk/eventhub/event-hubs/test/node/disconnects.spec.ts

Line 52 in 3f766f0

it("should receive after a disconnect", async () => {

However, it doesn't check that any events are returned on subsequent calls. It also doesn't verify that maxWaitTimeInSeconds is honored. The existing test should be updated to check both of these conditions.