Azure Cosmos DB - Ignoring serialization failures in Cosmos DB's FeedIterator.ReadNextAsync()

tval 0
Viewed 23

When using a Cosmos DB FeedIterator, how can I ignore serialization failures and continue reading the rest of the valid documents?

The Cosmos DB method FeedIterator.ReadNextAsync() can return multiple documents at a time. In case of a serialization error in one of the documents, the rest of the valid documents will be dropped (a JsonSerializationException exception is thrown).

SDK version: SQL API for .NET via the NuGet package Microsoft.Azure.Cosmos version 3.28.0.

public async Task<ICollection<T>> FetchAll(CancellationToken cancellationToken)
{
    var items = new List<T>();
    using var iterator = this.container.GetItemQueryIterator<T>();
    while (iterator.HasMoreResults)
    {
        try
        {
            foreach (var item in await iterator.ReadNextAsync(cancellationToken).ConfigureAwait(false))
            {
                items.Add(item);
            }
         }
         catch (JsonSerializationException ex)
         {
                    // When getting to here - the entire bulk has failed (100 documents) even if only a single document is invalid.
         }
    }

    return items;
}

this.container is a CosmosClient with default cosmosClientOptions.

azure-cosmosdb-sqlapi azure-cosmosdb .net
1 Answers

You can instead use the Stream APIs and handle serialization yourself, and decide to skip the ones that fail.

public async Task<ICollection<T>> FetchAll(CancellationToken cancellationToken)
{
    var items = new List<T>();
    using var iterator = this.container.GetItemQueryStreamIterator();
    while (iterator.HasMoreResults)
    {
        using (ResponseMessage response = await feedIterator.ReadNextAsync(cancellationToken).ConfigureAwait(false))
        {
            response.EnsureSuccessStatusCode();
            // deserialize the response.Content Stream using your serializer of choice and catch any errors
        }
    }

    return items;
}

Or you can deserialize into a generic Json object and then try to convert:

public async Task<ICollection<T>> FetchAll(CancellationToken cancellationToken)
{
    var items = new List<T>();
    using var iterator = this.container.GetItemQueryIterator<JObject>();
    while (iterator.HasMoreResults)
    {
        foreach (var jobject in await iterator.ReadNextAsync(cancellationToken).ConfigureAwait(false))
        {
            
            try
            {
                var item = jobject.ToObject<T>();
                items.Add(item);
            }
            catch (JsonSerializationException ex)
            {
                    // When getting to here - the entire bulk has failed (100 documents) even if only a single document is invalid.
            }
            
        }
    }

    return items;
}
Matias Quaranta0
Related