[Cosmos] Adds option for Bulk to continue on error (Azure#14277)

* Adds continue on error bulk option

* Formats

* Revert test config

* Changelog

* Adds docs for BulkOptions

* Fixes tsdoc escaping

Author: zfoster

sdk/cosmosdb/cosmos/CHANGELOG.md

@@ -1,5 +1,9 @@
 # Release History
 
+## 3.10.4 (2021-03-12)
+
+- FEATURE: Adds Bulk continueOnError option
+
 ## 3.10.3 (2021-03-12)
 
 - BUGFIX: Removes direct dependency on @azure/identity while retaining compatibility.

sdk/cosmosdb/cosmos/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@azure/cosmos",
-  "version": "3.10.3",
+  "version": "3.10.4",
   "description": "Microsoft Azure Cosmos DB Service Node.js SDK for SQL API",
   "sdk-type": "client",
   "keywords": [

sdk/cosmosdb/cosmos/review/cosmos.api.md

@@ -32,6 +32,12 @@ export const BulkOperationType: {
     readonly Replace: "Replace";
 };
 
+// @public
+export interface BulkOptions {
+    // (undocumented)
+    continueOnError?: boolean;
+}
+
 // @public
 export class ChangeFeedIterator<T> {
     fetchNext(): Promise<ChangeFeedResponse<Array<T & Resource>>>;
@@ -65,11 +71,12 @@ export class ChangeFeedResponse<T> {
 export class ClientContext {
     constructor(cosmosClientOptions: CosmosClientOptions, globalEndpointManager: GlobalEndpointManager);
     // (undocumented)
-    bulk<T>({ body, path, resourceId, partitionKeyRangeId, options }: {
+    bulk<T>({ body, path, partitionKeyRangeId, resourceId, bulkOptions, options }: {
         body: T;
         path: string;
         partitionKeyRangeId: string;
         resourceId: string;
+        bulkOptions?: BulkOptions;
         options?: RequestOptions;
     }): Promise<Response<any>>;
     // (undocumented)
@@ -352,6 +359,7 @@ export const Constants: {
         ALLOW_MULTIPLE_WRITES: string;
         IsBatchRequest: string;
         IsBatchAtomic: string;
+        BatchContinueOnError: string;
         ForceRefresh: string;
     };
     WritableLocations: string;
@@ -829,7 +837,7 @@ export class ItemResponse<T extends ItemDefinition> extends ResourceResponse<T &
 // @public
 export class Items {
     constructor(container: Container, clientContext: ClientContext);
-    bulk(operations: OperationInput[], options?: RequestOptions): Promise<OperationResponse[]>;
+    bulk(operations: OperationInput[], bulkOptions?: BulkOptions, options?: RequestOptions): Promise<OperationResponse[]>;
     changeFeed(partitionKey: string | number | boolean, changeFeedOptions?: ChangeFeedOptions): ChangeFeedIterator<any>;
     changeFeed(changeFeedOptions?: ChangeFeedOptions): ChangeFeedIterator<any>;
     changeFeed<T>(partitionKey: string | number | boolean, changeFeedOptions?: ChangeFeedOptions): ChangeFeedIterator<T>;

sdk/cosmosdb/cosmos/src/ClientContext.ts

@@ -22,6 +22,7 @@ import { RequestContext } from "./request/RequestContext";
 import { request as executeRequest } from "./request/RequestHandler";
 import { SessionContainer } from "./session/sessionContainer";
 import { SessionContext } from "./session/SessionContext";
+import { BulkOptions } from "./utils/batch";
 
 /** @hidden */
 const log = logger("ClientContext");
@@ -546,14 +547,16 @@ export class ClientContext {
   public async bulk<T>({
     body,
     path,
-    resourceId,
     partitionKeyRangeId,
+    resourceId,
+    bulkOptions = {},
     options = {}
   }: {
     body: T;
     path: string;
     partitionKeyRangeId: string;
     resourceId: string;
+    bulkOptions?: BulkOptions;
     options?: RequestOptions;
   }): Promise<Response<any>> {
     try {
@@ -576,6 +579,8 @@ export class ClientContext {
       request.headers[Constants.HttpHeaders.IsBatchRequest] = true;
       request.headers[Constants.HttpHeaders.PartitionKeyRangeID] = partitionKeyRangeId;
       request.headers[Constants.HttpHeaders.IsBatchAtomic] = false;
+      request.headers[Constants.HttpHeaders.BatchContinueOnError] =
+        bulkOptions.continueOnError || false;
 
       this.applySessionToken(request);
 

sdk/cosmosdb/cosmos/src/client/Item/Items.ts

@@ -20,7 +20,8 @@ import {
   getPartitionKeyToHash,
   decorateOperation,
   OperationResponse,
-  OperationInput
+  OperationInput,
+  BulkOptions
 } from "../../utils/batch";
 import { hashV1PartitionKey } from "../../utils/hashing/v1";
 import { hashV2PartitionKey } from "../../utils/hashing/v2";
@@ -394,10 +395,12 @@ export class Items {
    * ```
    *
    * @param operations - List of operations. Limit 100
+   * @param bulkOptions - Optional options object to modify bulk behavior. Pass \{ continueOnError: true \} to continue executing operations when one fails. (Defaults to false) ** NOTE: THIS WILL DEFAULT TO TRUE IN THE 4.0 RELEASE
    * @param options - Used for modifying the request.
    */
   public async bulk(
     operations: OperationInput[],
+    bulkOptions?: BulkOptions,
     options?: RequestOptions
   ): Promise<OperationResponse[]> {
     const {
@@ -443,6 +446,7 @@ export class Items {
               partitionKeyRangeId: batch.rangeId,
               path,
               resourceId: this.container.url,
+              bulkOptions,
               options
             });
             response.result.forEach((operationResponse: OperationResponse, index: number) => {

sdk/cosmosdb/cosmos/src/common/constants.ts

@@ -158,6 +158,7 @@ export const Constants = {
     // Bulk/Batch header
     IsBatchRequest: "x-ms-cosmos-is-batch-request",
     IsBatchAtomic: "x-ms-cosmos-batch-atomic",
+    BatchContinueOnError: "x-ms-cosmos-batch-continue-on-error",
 
     // Cache Refresh header
     ForceRefresh: "x-ms-force-refresh"

sdk/cosmosdb/cosmos/src/index.ts

@@ -7,6 +7,7 @@ export { setAuthorizationTokenHeaderUsingMasterKey } from "./auth";
 export {
   Operation,
   OperationResponse,
+  BulkOptions,
   CreateOperation,
   UpsertOperation,
   ReplaceOperation,

sdk/cosmosdb/cosmos/src/utils/batch.ts

@@ -29,6 +29,14 @@ export interface OperationResponse {
   resourceBody?: JSONObject;
 }
 
+/**
+ * Options object used to modify bulk execution.
+ * continueOnError (Default value: false) - Continues bulk execution when an operation fails ** NOTE THIS WILL DEFAULT TO TRUE IN the 4.0 RELEASE
+ */
+export interface BulkOptions {
+  continueOnError?: boolean;
+}
+
 export function isKeyInRange(min: string, max: string, key: string): boolean {
   const isAfterMinInclusive = key.localeCompare(min) >= 0;
   const isBeforeMax = key.localeCompare(max) < 0;

sdk/cosmosdb/cosmos/test/public/functional/item.spec.ts

@@ -409,7 +409,8 @@ describe("bulk item operations", function() {
         {
           operationType: BulkOperationType.Create,
           resourceBody: {
-            ttl: -10
+            ttl: -10,
+            key: "A"
           }
         },
         {
@@ -424,6 +425,27 @@ describe("bulk item operations", function() {
       const response = await v2Container.items.bulk(operations);
       assert.equal(response[1].statusCode, 424);
     });
+    it("Continues after errors with continueOnError true", async function() {
+      const operations = [
+        {
+          operationType: BulkOperationType.Create,
+          resourceBody: {
+            ttl: -10,
+            key: "A"
+          }
+        },
+        {
+          operationType: BulkOperationType.Create,
+          resourceBody: {
+            key: "A",
+            licenseType: "B",
+            id: "o239uroihndsf"
+          }
+        }
+      ];
+      const response = await v2Container.items.bulk(operations, { continueOnError: true });
+      assert.equal(response[1].statusCode, 201);
+    });
     it("autogenerates IDs for Create operations", async function() {
       const operations = [
         {