diff --git a/sdk/eventhub/event-hubs/changelog.md b/sdk/eventhub/event-hubs/changelog.md
index 0bca2e62590a..ab61d421508a 100644
--- a/sdk/eventhub/event-hubs/changelog.md
+++ b/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
diff --git a/sdk/eventhub/event-hubs/package.json b/sdk/eventhub/event-hubs/package.json
index b211d18fe6cb..8c12e0b2e335 100644
--- a/sdk/eventhub/event-hubs/package.json
+++ b/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",
diff --git a/sdk/eventhub/event-hubs/src/eventProcessor.ts b/sdk/eventhub/event-hubs/src/eventProcessor.ts
index 41bd69f689be..a63d4cb3f1b7 100644
--- a/sdk/eventhub/event-hubs/src/eventProcessor.ts
+++ b/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);
   }
 }
 
diff --git a/sdk/eventhub/event-hubs/src/partitionProcessor.ts b/sdk/eventhub/event-hubs/src/partitionProcessor.ts
index db3587e53d7d..cd0dea310fa8 100644
--- a/sdk/eventhub/event-hubs/src/partitionProcessor.ts
+++ b/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}`);
+      }
     }
   }
 
diff --git a/sdk/eventhub/event-hubs/src/partitionPump.ts b/sdk/eventhub/event-hubs/src/partitionPump.ts
index ac79c7414021..6cd522e46fb8 100644
--- a/sdk/eventhub/event-hubs/src/partitionPump.ts
+++ b/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;
     }
   }
diff --git a/sdk/eventhub/event-hubs/src/pumpManager.ts b/sdk/eventhub/event-hubs/src/pumpManager.ts
index 2f1ccede1c39..415a2e1e8878 100644
--- a/sdk/eventhub/event-hubs/src/pumpManager.ts
+++ b/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: {
diff --git a/sdk/eventhub/event-hubs/src/util/constants.ts b/sdk/eventhub/event-hubs/src/util/constants.ts
index 7be16bda7e84..58c3648d514e 100644
--- a/sdk/eventhub/event-hubs/src/util/constants.ts
+++ b/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"
 };
diff --git a/sdk/eventhub/event-hubs/test/eventProcessor.spec.ts b/sdk/eventhub/event-hubs/test/eventProcessor.spec.ts
index 5ded87805b92..d3f4184b72e4 100644
--- a/sdk/eventhub/event-hubs/test/eventProcessor.spec.ts
+++ b/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,
diff --git a/sdk/eventhub/eventhubs-checkpointstore-blob/changelog.md b/sdk/eventhub/eventhubs-checkpointstore-blob/changelog.md
index f19f6d6dc54d..9a914394d4b6 100644
--- a/sdk/eventhub/eventhubs-checkpointstore-blob/changelog.md
+++ b/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.
diff --git a/sdk/eventhub/eventhubs-checkpointstore-blob/package.json b/sdk/eventhub/eventhubs-checkpointstore-blob/package.json
index 486dc7ac5f8d..35a090546617 100644
--- a/sdk/eventhub/eventhubs-checkpointstore-blob/package.json
+++ b/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",
diff --git a/sdk/eventhub/eventhubs-checkpointstore-blob/src/blobCheckpointStore.ts b/sdk/eventhub/eventhubs-checkpointstore-blob/src/blobCheckpointStore.ts
index 8dbbf2f6a50c..5a105f16cd8c 100644
--- a/sdk/eventhub/eventhubs-checkpointstore-blob/src/blobCheckpointStore.ts
+++ b/sdk/eventhub/eventhubs-checkpointstore-blob/src/blobCheckpointStore.ts
@@ -2,7 +2,7 @@
 // Licensed under the MIT License.
 
 import { CheckpointStore, PartitionOwnership, Checkpoint } from "@azure/event-hubs";
-import { ContainerClient, Metadata } from "@azure/storage-blob";
+import { ContainerClient, Metadata, RestError } from "@azure/storage-blob";
 import * as log from "./log";
 import { throwTypeErrorIfParameterMissing } from "./util/error";
 
@@ -107,12 +107,22 @@ export class BlobCheckpointStore implements CheckpointStore {
           `LastModifiedTime: ${ownership.lastModifiedTimeInMs}, ETag: ${ownership.etag}`
         );
       } catch (err) {
-        // NOTE: there is some ordinary contention that can occur as different consumers battle over
-        // ownership. So the catching (and _only_ logging, not rethrowing) of this error is intentional.
+        const restError = err as RestError;
+
+        if (restError.statusCode === 412) {
+          // etag failures (precondition not met) aren't fatal errors. They happen
+          // as multiple consumers attempt to claim the same partition (first one wins)
+          // and losers get this error.
+          log.blobCheckpointStore(`[${ownership.ownerId}] Did not claim partition ${ownership.partitionId}. Another processor has already claimed it.`);
+          continue;
+        }
+
         log.error(
           `Error occurred while claiming ownership for partition: ${ownership.partitionId}`,
           err
         );
+
+        throw err;
       }
     }
     return partitionOwnershipArray;
diff --git a/sdk/eventhub/eventhubs-checkpointstore-blob/test/blob-partition-manager.spec.ts b/sdk/eventhub/eventhubs-checkpointstore-blob/test/blob-checkpointstore.spec.ts
similarity index 85%
rename from sdk/eventhub/eventhubs-checkpointstore-blob/test/blob-partition-manager.spec.ts
rename to sdk/eventhub/eventhubs-checkpointstore-blob/test/blob-checkpointstore.spec.ts
index 85399bebe91e..2bad74e16ba6 100644
--- a/sdk/eventhub/eventhubs-checkpointstore-blob/test/blob-partition-manager.spec.ts
+++ b/sdk/eventhub/eventhubs-checkpointstore-blob/test/blob-checkpointstore.spec.ts
@@ -11,13 +11,14 @@ import debugModule from "debug";
 const debug = debugModule("azure:event-hubs:partitionPump");
 import { EnvVarKeys, getEnvVars } from "./utils/testUtils";
 import { BlobCheckpointStore } from "../src";
-import { ContainerClient } from "@azure/storage-blob";
-import { PartitionOwnership, Checkpoint } from "@azure/event-hubs";
+import { ContainerClient, RestError } from "@azure/storage-blob";
+import { PartitionOwnership, Checkpoint, EventHubConsumerClient } from "@azure/event-hubs";
 import { Guid } from "guid-typescript";
 import { parseIntOrThrow } from "../src/blobCheckpointStore";
+import { fail } from 'assert';
 const env = getEnvVars();
 
-describe("Blob Partition Manager", function(): void {
+describe("Blob Checkpoint Store", function(): void {
   const service = {
     storageConnectionString: env[EnvVarKeys.STORAGE_CONNECTION_STRING]
   };
@@ -51,6 +52,63 @@ describe("Blob Partition Manager", function(): void {
     should.equal(listOwnership.length, 0);
   });
 
+  // these errors happen when we have multiple consumers starting up
+  // at the same time and load balancing amongst themselves. This is a 
+  // normal thing and shouldn't be reported to the user.
+  it("claimOwnership ignores errors about etags", async () => {
+    const checkpointStore = new BlobCheckpointStore(containerClient);
+
+    const originalClaimedOwnerships = await checkpointStore.claimOwnership([{
+      partitionId: "0",
+      consumerGroup: EventHubConsumerClient.defaultConsumerGroupName,
+      fullyQualifiedNamespace: "fqdn",
+      eventHubName: "ehname",
+      ownerId: "me"
+    }]);
+
+    const originalETag = originalClaimedOwnerships[0] && originalClaimedOwnerships[0].etag;
+    
+    const newClaimedOwnerships = await checkpointStore.claimOwnership(originalClaimedOwnerships);
+    newClaimedOwnerships.length.should.equal(1);
+
+    newClaimedOwnerships[0]!.etag!.should.not.equal(originalETag);
+
+    // we've now invalidated the previous ownership's etag so using the old etag will
+    // fail.
+    const shouldNotThrowButNothingWillClaim = await checkpointStore.claimOwnership([{
+      partitionId: "0",
+      consumerGroup: EventHubConsumerClient.defaultConsumerGroupName,
+      fullyQualifiedNamespace: "fqdn",
+      eventHubName: "ehname",
+      ownerId: "me",
+      etag: originalETag
+    }]);
+
+    shouldNotThrowButNothingWillClaim.length.should.equal(0);
+  });
+
+  it("claimOwnership will throw if the error is NOT an outdated etag", async () => {
+    const checkpointStore = new BlobCheckpointStore(containerClient);
+
+    // now let's induce a bad failure (removing the container)
+    await containerClient.delete();
+
+    try {
+      await checkpointStore.claimOwnership([{
+        partitionId: "0",
+        consumerGroup: EventHubConsumerClient.defaultConsumerGroupName,
+        fullyQualifiedNamespace: "fqdn",
+        eventHubName: "ehname",
+        ownerId: "me"
+      }]);
+      fail("Should have thrown an error - this isn't a normal claim collision issue");
+    } catch (err) {
+      (err instanceof RestError).should.be.ok;
+      // 404 because the container is missing (since we deleted it up above)
+      (err as RestError).statusCode!.should.equal(404);
+    }
+  });
+
   it("claimOwnership call should succeed, if it has been called for the first time", async function(): Promise<
     void
   > {