[event-hubs] fixes issue where processEvents ignores maxWaitTime after retryable disconnect event (Azure#12280)

Fixes Azure#12278 

## Description
This PR adds a check within the `PartitionPump`'s receive events loop to determine whether the receiver has been closed. If the receiver has been closed, it will create a new receiver using an event start position that matches the last event seen by the pump.

The receiver is explicitly closed when a disconnected event is received on the underlying AMQP connection, which causes calls to `receiveBatch` to immediately return any events it had collected up to this point. Note that once the receiver is closed, the receiver's onAmqpMessage handler is removed so it won't receive any additional events.

## Updates to testing
I manually tested the changes in this PR against the sample code in the linked issue.

I also updated the existing `disconnect` test to confirm that
- the `maxWaitTimeInSeconds` is honoured after a disconnected event is encountered.
- new events can be received on subsequent `processEvents` invocations.

Author: chradek

sdk/eventhub/event-hubs/CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## 5.3.1 (Unreleased)
 
+- Fixes issue [#12278](https://github.com/Azure/azure-sdk-for-js/issues/12278)
+  where the `processEvents` handler could ignore the `maxWaitTimeInSeconds`
+  parameter after a disconnection.
 
 ## 5.3.0 (2020-09-08)
 

sdk/eventhub/event-hubs/src/eventHubReceiver.ts

@@ -176,6 +176,13 @@ export class EventHubReceiver extends LinkEntity {
     return this._isReceivingMessages;
   }
 
+  /**
+   * Indicates if the receiver has been closed.
+   */
+  get isClosed(): boolean {
+    return this._isClosed;
+  }
+
   /**
    * @property The last enqueued event information. This property will only
    * be enabled when `trackLastEnqueuedEventProperties` option is set to true

sdk/eventhub/event-hubs/src/partitionPump.ts

@@ -59,38 +59,77 @@ export class PartitionPump {
     );
   }
 
-  private async _receiveEvents(partitionId: string): Promise<void> {
+  /**
+   * Creates a new `EventHubReceiver` and replaces any existing receiver.
+   * @param partitionId The partition the receiver should read messages from.
+   * @param lastSeenSequenceNumber The sequence number to begin receiving messages from (exclusive).
+   * If `-1`, then the PartitionPump's startPosition will be used instead.
+   */
+  private _setOrReplaceReceiver(
+    partitionId: string,
+    lastSeenSequenceNumber: number
+  ): EventHubReceiver {
+    // Determine what the new EventPosition should be.
+    // If this PartitionPump has received events, we'll start from the last
+    // seen sequenceNumber (exclusive).
+    // Otherwise, use the `_startPosition`.
+    const currentEventPosition: EventPosition =
+      lastSeenSequenceNumber >= 0
+        ? {
+            sequenceNumber: lastSeenSequenceNumber,
+            isInclusive: false
+          }
+        : this._startPosition;
+
+    // Set or replace the PartitionPump's receiver.
     this._receiver = new EventHubReceiver(
       this._context,
       this._partitionProcessor.consumerGroup,
       partitionId,
-      this._startPosition,
+      currentEventPosition,
       {
         ownerLevel: this._processorOptions.ownerLevel,
         trackLastEnqueuedEventProperties: this._processorOptions.trackLastEnqueuedEventProperties,
         retryOptions: this._processorOptions.retryOptions
       }
     );
 
+    return this._receiver;
+  }
+
+  private async _receiveEvents(partitionId: string): Promise<void> {
+    let lastSeenSequenceNumber = -1;
+    let receiver = this._setOrReplaceReceiver(partitionId, lastSeenSequenceNumber);
+
     while (this._isReceiving) {
       try {
-        const receivedEvents = await this._receiver.receiveBatch(
+        // Check if the receiver was closed so we can recreate it.
+        if (receiver.isClosed) {
+          receiver = this._setOrReplaceReceiver(partitionId, lastSeenSequenceNumber);
+        }
+
+        const receivedEvents = await receiver.receiveBatch(
           this._processorOptions.maxBatchSize,
           this._processorOptions.maxWaitTimeInSeconds,
           this._abortController.signal
         );
 
         if (
           this._processorOptions.trackLastEnqueuedEventProperties &&
-          this._receiver.lastEnqueuedEventProperties
+          receiver.lastEnqueuedEventProperties
         ) {
-          this._partitionProcessor.lastEnqueuedEventProperties = this._receiver.lastEnqueuedEventProperties;
+          this._partitionProcessor.lastEnqueuedEventProperties =
+            receiver.lastEnqueuedEventProperties;
         }
         // avoid calling user's processEvents handler if the pump was stopped while receiving events
         if (!this._isReceiving) {
           return;
         }
 
+        if (receivedEvents.length) {
+          lastSeenSequenceNumber = receivedEvents[receivedEvents.length - 1].sequenceNumber;
+        }
+
         const span = createProcessingSpan(
           receivedEvents,
           {

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

@@ -50,30 +50,86 @@ describe("disconnected", function() {
     });
 
     it("should receive after a disconnect", async () => {
-      const client = new EventHubConsumerClient(
+      /**
+       * This test validates that an `EventHubConsumerClient.subscribe()` call continues
+       * receiving events after a `disconnected` event occurs on the underlying connection.
+       *
+       * https://github.com/Azure/azure-sdk-for-js/pull/12280 describes an issue where `processEvents`
+       * would be invoked with 0 events and ignoring the `maxWaitTimeInSeconds` after a `disconnected` event.
+       *
+       * For a single `subscribe()` call, this test does the following:
+       * 1. Ensure events can be received normally before the `disconnected` event.
+       * 2. Ensure that the `maxWaitTimeInSeconds` is honoured after a `disconnected` event.
+       * 3. Ensure that events can be received normally after the `disconnected` event.
+       */
+      const consumer = new EventHubConsumerClient(
         EventHubConsumerClient.defaultConsumerGroupName,
         service.connectionString,
         service.path
       );
+
+      const producer = new EventHubProducerClient(service.connectionString, service.path);
+      const eventSentBeforeDisconnect = { body: "the first event" };
+      const eventSentAfterDisconnect = { body: "the second event" };
+
+      const maxWaitTimeInSeconds = 10;
       const partitionId = "0";
-      const partitionProperties = await client.getPartitionProperties(partitionId);
-      const clientConnectionContext = client["_context"];
+      const partitionProperties = await consumer.getPartitionProperties(partitionId);
+      const clientConnectionContext = consumer["_context"];
+
+      // Send the first event after getting partition properties so that we can expect to receive it.
+      await producer.sendBatch([eventSentBeforeDisconnect], { partitionId });
 
       let subscription: Subscription | undefined;
       let originalConnectionId: string;
 
+      let processEventsInvocationCount = 0;
+      let firstInvocationEndTime = 0;
       await new Promise((resolve, reject) => {
-        subscription = client.subscribe(
+        subscription = consumer.subscribe(
           partitionId,
           {
             processEvents: async (data) => {
+              processEventsInvocationCount++;
               should.exist(data);
-              should.equal(data.length, 0);
-              if (!originalConnectionId) {
+              if (processEventsInvocationCount === 1) {
+                // 1. Ensure events can be received normally before the `disconnected` event.
+                should.equal(
+                  data.length,
+                  1,
+                  "Expected to receive 1 event in first processEvents invocation."
+                );
+                should.equal(data[0].body, eventSentBeforeDisconnect.body);
                 originalConnectionId = clientConnectionContext.connectionId;
                 // Trigger a disconnect on the underlying connection.
                 clientConnectionContext.connection["_connection"].idle();
-              } else {
+                firstInvocationEndTime = Date.now();
+              } else if (processEventsInvocationCount === 2) {
+                // 2. Ensure that the `maxWaitTimeInSeconds` is honoured after a `disconnected` event.
+                // No new events should have been received at this point since we received the last event in the previous invocation.
+                should.equal(
+                  data.length,
+                  0,
+                  "Expected to receive 0 events in second processEvents invocation."
+                );
+                // The elapsed time since the last processEvents invocation should be >= maxWaitTimeInSeconds
+                should.equal(
+                  Date.now() - firstInvocationEndTime >= maxWaitTimeInSeconds,
+                  true,
+                  "Expected elapsed time between first and second processEvents invocations to be >= maxWaitTimeInSeconds."
+                );
+                const newConnectionId = clientConnectionContext.connectionId;
+                should.not.equal(originalConnectionId, newConnectionId);
+                // Send a new event that will be immediately receivable.
+                await producer.sendBatch([eventSentAfterDisconnect], { partitionId });
+              } else if (processEventsInvocationCount === 3) {
+                // 3. Ensure that events can be received normally after the `disconnected` event.
+                should.equal(
+                  data.length,
+                  1,
+                  "Expected to receive 1 event in third processEvents invocation."
+                );
+                should.equal(data[0].body, eventSentAfterDisconnect.body);
                 const newConnectionId = clientConnectionContext.connectionId;
                 should.not.equal(originalConnectionId, newConnectionId);
                 resolve();
@@ -84,13 +140,16 @@ describe("disconnected", function() {
             }
           },
           {
-            startPosition: { sequenceNumber: partitionProperties.lastEnqueuedSequenceNumber },
-            maxWaitTimeInSeconds: 2
+            startPosition: {
+              sequenceNumber: partitionProperties.lastEnqueuedSequenceNumber
+            },
+            maxWaitTimeInSeconds
           }
         );
       });
       await subscription!.close();
-      await client.close();
+      await consumer.close();
+      await producer.close();
     });
   });