[event-hubs] Better error handling and reporting

sdk/eventhub/event-hubs/changelog.md

@@ -1,3 +1,8 @@
+### 5.0.0-preview.8
+
+- Fixed potential issues with claims being mismanaged when subscriptions terminate.
+- Improved reporting of errors that occur when attempting to claim partitions from CheckpointStores.
+
 ### 2019-12-03 5.0.0-preview.7
 
 - Improves load-balancing capabilities to reduce the frequency that partitions are claimed by other running

sdk/eventhub/event-hubs/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@azure/event-hubs",
   "sdk-type": "client",
-  "version": "5.0.0-preview.7",
+  "version": "5.0.0-preview.8",
   "description": "Azure Event Hubs SDK for JS.",
   "author": "Microsoft Corporation",
   "license": "MIT",

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

@@ -4,7 +4,7 @@
 import uuid from "uuid/v4";
 import { EventHubClient } from "./impl/eventHubClient";
 import { EventPosition } from "./eventPosition";
-import { PumpManager } from "./pumpManager";
+import { PumpManager, PumpManagerImpl } from "./pumpManager";
 import { AbortController, AbortSignalLike } from "@azure/abort-controller";
 import * as log from "./log";
 import { FairPartitionLoadBalancer, PartitionLoadBalancer } from "./partitionLoadBalancer";
@@ -258,7 +258,7 @@ export class EventProcessor {
     this._consumerGroup = consumerGroup;
     this._eventHubClient = eventHubClient;
     this._processorOptions = options;
-    this._pumpManager = options.pumpManager || new PumpManager(this._id, this._processorOptions);
+    this._pumpManager = options.pumpManager || new PumpManagerImpl(this._id, this._processorOptions);
     const inactiveTimeLimitInMS = options.inactiveTimeLimitInMs || this._inactiveTimeLimitInMs;
     this._partitionLoadBalancer =
       options.partitionLoadBalancer ||
@@ -305,9 +305,11 @@ export class EventProcessor {
     );
     try {
       const claimedOwnerships = await this._checkpointStore.claimOwnership([ownershipRequest]);
-      // since we only claim one ownership at a time, check the array length and throw
+
+      // can happen if the partition was claimed out from underneath us - we shouldn't
+      // attempt to spin up a processor.
       if (!claimedOwnerships.length) {
-        throw new Error(`Failed to claim ownership of partition ${ownershipRequest.partitionId}`);
+        return;
       }
 
       log.partitionLoadBalancer(
@@ -498,8 +500,6 @@ export class EventProcessor {
    *
    */
   async stop(): Promise<void> {
-    await this.abandonPartitionOwnerships();
-
     log.eventProcessor(`[${this._id}] Stopping an EventProcessor.`);
     if (this._abortController) {
       // cancel the event processor loop
@@ -519,6 +519,8 @@ export class EventProcessor {
     } finally {
       log.eventProcessor(`[${this._id}] EventProcessor stopped.`);
     }
+
+    await this.abandonPartitionOwnerships();
   }
 
   private async abandonPartitionOwnerships() {
@@ -528,7 +530,7 @@ export class EventProcessor {
     for (const ownership of ourOwnerships) {
       ownership.ownerId = "";
     }
-    this._checkpointStore.claimOwnership(ourOwnerships);
+    return this._checkpointStore.claimOwnership(ourOwnerships);
   }
 }
 

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

@@ -6,7 +6,8 @@ import {
   InitializationContext,
   BasicPartitionProperties
 } from "./eventHubConsumerClientModels";
-import { EventPosition } from ".";
+import { EventPosition } from "./eventPosition";
+import * as log from "./log";
 
 /**
  * A checkpoint is meant to represent the last successfully processed event by the user from a particular
@@ -176,7 +177,11 @@ export class PartitionProcessor implements InitializationContext {
    */
   async processError(error: Error): Promise<void> {
     if (this._eventHandlers.processError) {
-      await this._eventHandlers.processError(error, this);
+      try {
+        await this._eventHandlers.processError(error, this);
+      } catch (err) {
+        log.partitionPump(`Error thrown from user's processError handler : ${err}`);
+      }
     }
   }
 

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

@@ -42,8 +42,9 @@ export class PartitionPump {
     let userRequestedDefaultPosition: EventPosition | undefined;
     try {
       userRequestedDefaultPosition = await this._partitionProcessor.initialize();
-    } catch {
+    } catch (err) {
       // swallow the error from the user-defined code
+      this._partitionProcessor.processError(err);
     }
 
     const startingPosition = getStartingPosition(
@@ -137,6 +138,7 @@ export class PartitionPump {
       await this._partitionProcessor.close(reason);
     } catch (err) {
       log.error("An error occurred while closing the receiver.", err);
+      this._partitionProcessor.processError(err);
       throw err;
     }
   }

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

@@ -13,8 +13,39 @@ import * as log from "./log";
  * It also starts a PartitionPump when it is created, and stops a
  * PartitionPump when it is removed.
  * @ignore
+ * @internal
  */
-export class PumpManager {
+export interface PumpManager {
+  /**
+   * Creates and starts a PartitionPump.
+   * @param eventHubClient The EventHubClient to forward to the PartitionPump.
+   * @param initialEventPosition The EventPosition to forward to the PartitionPump.
+   * @param partitionProcessor The PartitionProcessor to forward to the PartitionPump.
+   * @param abortSignal Used to cancel pump creation.
+   * @ignore
+   */
+  createPump(
+    eventHubClient: EventHubClient,
+    initialEventPosition: EventPosition | undefined,
+    partitionProcessor: PartitionProcessor
+  ): Promise<void>;
+
+  /**
+   * Stops all PartitionPumps and removes them from the internal map.
+   * @param reason The reason for removing the pump.
+   * @ignore
+   */
+  removeAllPumps(reason: CloseReason): Promise<void>;
+}
+
+/**
+ * The PumpManager handles the creation and removal of PartitionPumps.
+ * It also starts a PartitionPump when it is created, and stops a
+ * PartitionPump when it is removed.
+ * @ignore
+ * @internal
+ */
+export class PumpManagerImpl implements PumpManager {
   private readonly _eventProcessorName: string;
   private readonly _options: FullEventProcessorOptions;
   private _partitionIdToPumps: {

sdk/eventhub/event-hubs/src/util/constants.ts

@@ -6,5 +6,5 @@
  */
 export const packageJsonInfo = {
   name: "@azure/event-hubs",
-  version: "5.0.0-preview.7"
+  version: "5.0.0-preview.8"
 };

sdk/eventhub/event-hubs/test/eventProcessor.spec.ts

@@ -15,7 +15,7 @@ import {
   LastEnqueuedEventProperties,
   SubscriptionEventHandlers,
   EventPosition,
-  CheckpointStore,
+  CheckpointStore
 } from "../src";
 import { EventHubClient } from "../src/impl/eventHubClient";
 import { EnvVarKeys, getEnvVars, loopUntil } from "./utils/testUtils";
@@ -34,7 +34,6 @@ import { GreedyPartitionLoadBalancer } from "../src/partitionLoadBalancer";
 import { AbortError } from "@azure/abort-controller";
 import { FakeSubscriptionEventHandlers } from './utils/fakeSubscriptionEventHandlers';
 import sinon from 'sinon';
-import { PumpManager } from '../src/pumpManager';
 const env = getEnvVars();
 
 describe("Event Processor", function(): void {
@@ -188,6 +187,63 @@ describe("Event Processor", function(): void {
       });
     });
 
+    it("if we fail to claim partitions we don't start up new processors", async () => {
+      const checkpointStore = {
+        claimOwnershipCalled: false,
+
+        // the important thing is that the EventProcessor won't be able to claim
+        // any partitions, causing it to go down the "I tried but failed" path.
+        async claimOwnership(_: PartitionOwnership[]): Promise<PartitionOwnership[]> {
+          checkpointStore.claimOwnershipCalled = true;
+          return [];
+        },
+
+        // (these aren't used for this test)
+        async listOwnership(): Promise<PartitionOwnership[]> { return []; },
+        async updateCheckpoint(): Promise<void> { },
+        async listCheckpoints(): Promise<Checkpoint[]> { return []; }        
+      };
+
+      const pumpManager = {
+        createPumpCalled: false,
+
+        async createPump() {
+          pumpManager.createPumpCalled = true;
+        },
+
+        async removeAllPumps() { }
+      }
+
+      const eventProcessor = new EventProcessor(
+        EventHubClient.defaultConsumerGroupName,
+        client,
+        {
+          processEvents: async () => { },
+          processError: async () => { },
+        },
+        checkpointStore,
+        {
+          ...defaultOptions,
+          pumpManager: pumpManager
+        }
+      );
+
+      await eventProcessor['_claimOwnership']({
+        consumerGroup: "cgname",
+        eventHubName: "ehname",
+        fullyQualifiedNamespace: "fqdn",
+        ownerId: "owner",
+        partitionId: "0"
+      });
+
+      // when we fail to claim a partition we should _definitely_
+      // not attempt to start a pump.
+      pumpManager.createPumpCalled.should.be.false;
+
+      // we'll attempt to claim a partition (but won't succeed)
+      checkpointStore.claimOwnershipCalled.should.be.true;
+    });
+
     it("abandoned claims are treated as unowned claims", async () => {
       const commonFields = {
         fullyQualifiedNamespace: "irrelevant namespace",
@@ -215,13 +271,14 @@ describe("Event Processor", function(): void {
       sinon.replaceGetter(fakeEventHubClient, 'eventHubName', () => commonFields.eventHubName);
       sinon.replaceGetter(fakeEventHubClient, 'fullyQualifiedNamespace', () => commonFields.fullyQualifiedNamespace);
 
-      const fakePumpManager = sinon.createStubInstance(PumpManager);
-
       const ep = new EventProcessor(commonFields.consumerGroup, fakeEventHubClient as any, handlers, checkpointStore, {
         maxBatchSize: 1,
         loopIntervalInMs: 1,
         maxWaitTimeInSeconds: 1,
-        pumpManager: fakePumpManager as any
+        pumpManager: {
+          async createPump() { },
+          async removeAllPumps(): Promise<void> { }
+        }
       });
 
       // allow three iterations through the loop - one for each partition that 
@@ -270,6 +327,106 @@ describe("Event Processor", function(): void {
     });
   });
 
+  it("claimOwnership throws and is reported to the user", async () => {
+    const errors = [];
+
+    const faultyCheckpointStore: CheckpointStore = {
+      listOwnership: async () => [],
+      claimOwnership: async () => {
+        throw new Error("Some random failure!");
+      },
+      updateCheckpoint: async () => {},
+      listCheckpoints: async () => []
+    };
+
+    const eventProcessor = new EventProcessor(
+      EventHubClient.defaultConsumerGroupName,
+      client,
+      {
+        processEvents: async () => {},
+        processError: async (err, _) => {
+          errors.push(err);
+        }
+      },
+      faultyCheckpointStore,
+      {
+        ...defaultOptions,
+        partitionLoadBalancer: new GreedyPartitionLoadBalancer(["0"])
+      }
+    );
+
+    // claimOwnership() calls that fail in the runloop of eventProcessor
+    // will get directed to the user's processError handler.
+    eventProcessor.start();
+
+    try {
+      await loopUntil({
+        name: "waiting for checkpoint store errors to show up",
+        timeBetweenRunsMs: 1000,
+        maxTimes: 30,
+        until: async () => errors.length !== 0
+      });
+
+      errors.length.should.equal(1);
+    } finally {
+      // this will also fail - we "abandon" all claimed partitions at
+      // when a processor is stopped (which requires us to claim them
+      // with an empty owner ID).
+      //
+      // Note that this one gets thrown directly from stop(), rather
+      // than reporting to processError() since we have a direct
+      // point of contact with the user.
+      await eventProcessor.stop().should.be.rejectedWith(/Some random failure!/);
+    }
+  });
+
+  it("errors thrown from the user's handlers are reported to processError()", async () => {
+    const errors = new Set<Error>();
+
+    const eventProcessor = new EventProcessor(
+      EventHubClient.defaultConsumerGroupName,
+      client,
+      {
+        processClose: async () => { throw new Error("processClose() error") },
+        processEvents: async () => { throw new Error("processEvents() error"); },
+        processInitialize: async () => { throw new Error("processInitialize() error") },
+        processError: async (err, _) => {
+          errors.add(err);
+          throw new Error("These are logged but ignored");
+        }
+      },
+      new InMemoryCheckpointStore(),
+      {
+        ...defaultOptions,
+        partitionLoadBalancer: new GreedyPartitionLoadBalancer(["0"])
+      }
+    );
+
+    // errors that occur within the user's own event handlers will get
+    // routed to their processError() handler
+    eventProcessor.start();
+
+    try {
+      await loopUntil({
+        name: "waiting for errors thrown from user's handlers",
+        timeBetweenRunsMs: 1000,
+        maxTimes: 30,
+        until: async () => errors.size >= 3
+      });
+
+      const messages = [...errors].map(e => e.message);
+      messages.sort();
+
+      messages.should.deep.equal([
+        "processClose() error",
+        "processEvents() error",
+        "processInitialize() error"
+      ]);
+    } finally {
+      await eventProcessor.stop();
+    }
+  });
+
   it("should expose an id #RunnableInBrowser", async function(): Promise<void> {
     const processor = new EventProcessor(
       EventHubClient.defaultConsumerGroupName,

sdk/eventhub/eventhubs-checkpointstore-blob/changelog.md

@@ -1,3 +1,8 @@
+### 1.0.0-preview.6
+
+- `claimOwnership()` will throw on underlying issues with storage, rather than
+   failing silently.
+
 ### 2019-12-03 - 1.0.0-preview.5
 
 - Updated to use the latest version of the `@azure/event-hubs` package.

sdk/eventhub/eventhubs-checkpointstore-blob/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@azure/eventhubs-checkpointstore-blob",
   "sdk-type": "client",
-  "version": "1.0.0-preview.5",
+  "version": "1.0.0-preview.6",
   "description": "An Azure Storage Blob solution to store checkpoints when using Event Hubs.",
   "author": "Microsoft Corporation",
   "license": "MIT",
@@ -63,7 +63,7 @@
     "unit-test": "npm run unit-test:node && npm run unit-test:browser"
   },
   "dependencies": {
-    "@azure/event-hubs": "5.0.0-preview.7",
+    "@azure/event-hubs": "5.0.0-preview.8",
     "@azure/storage-blob": "^12.0.0",
     "debug": "^4.1.1",
     "events": "^3.0.0",