Tỷ lệ yêu cầu CosmosDb lớn với insertMany

Trình điều khiển Mongo cho bạn biết bản ghi nào có lỗi và bản ghi nào chưa được xử lý. Nếu tất cả các lỗi (thường là một) có mã 16500, thì vấn đề của bạn là điều chỉnh và thử lại các lỗi và các bản ghi còn lại vẫn an toàn. Nếu không, lỗi của bạn là do nguyên nhân khác và bạn nên phân tích và quyết định xem có nên tiếp tục thử lại hay không.

Trình điều khiển Mongo không trả về tiêu đề HTTP trong đó Cosmos DB đề xuất độ trễ trước khi thử lại, nhưng đó không phải là vấn đề lớn. Dù sao thì độ trễ cũng không đảm bảo thành công, bởi vì các yêu cầu khác truy cập vào cùng một cơ sở dữ liệu có thể sử dụng hết RU. Tốt hơn hết bạn nên thử nghiệm và xác định quy tắc thử lại của riêng mình. Dưới đây là giải pháp đệ quy đơn giản tiếp tục thử lại cho đến khi tất cả đều ổn hoặc đạt đến giới hạn thử lại.

    private async Task InsertManyWithRetry(IMongoCollection<BsonDocument> collection, 
        IEnumerable<BsonDocument> batch, int retries = 10, int delay = 300)
    {
        var batchArray = batch.ToArray();

        try
        {
            await collection.InsertManyAsync(batchArray);
        }
        catch (MongoBulkWriteException<BsonDocument> e)
        {
            if (retries <= 0)
                throw;

            //Check if there were any errors other than throttling.
            var realErrors = e.WriteErrors.Where(we => we.Code != 16500).ToArray();
            //Re-throw original exception for now.
            //TODO: We can make it more sophisticated by continuing with unprocessed records and collecting all errors from all retries.
            if (realErrors.Any())
                throw;

            //Take all records that had errors.
            var errors = e.WriteErrors.Select(we => batchArray[we.Index]);
            //Take all unprocessed records.
            var unprocessed = e.UnprocessedRequests
                .Where(ur => ur.ModelType == WriteModelType.InsertOne)
                .OfType<InsertOneModel<BsonDocument>>() 
                .Select(ur => ur.Document);

            var retryBatchArray = errors.Union(unprocessed).ToArray();

            _logger($"Retry {retryBatchArray.Length} records after {delay} ms");

            await Task.Delay(delay);

            await InsertManyWithRetry(collection, retryBatchArray, retries - 1, delay);
        }
    }