Change feed processor in Azure Cosmos DB

The change feed processor in .NET is available for latest version mode and all versions and deletes mode. All versions and deletes mode is in preview and is supported for the change feed processor beginning in version 3.40.0-preview.0. The point of entry for both modes is always the monitored container.

To read using latest version mode, in a Container instance, you call GetChangeFeedProcessorBuilder:

using System;
using System.Collections.Generic;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Azure.Cosmos;
using Microsoft.Azure.Cosmos.Fluent;
using Microsoft.Extensions.Configuration;

namespace ChangeFeedSample
{
    class Program
    {
        static async Task Main(string[] args)
        {
            IConfiguration configuration = BuildConfiguration();

            CosmosClient cosmosClient = BuildCosmosClient(configuration);

            await InitializeContainersAsync(cosmosClient, configuration);

            ChangeFeedProcessor processor = await StartChangeFeedProcessorAsync(cosmosClient, configuration);

            await GenerateItemsAsync(cosmosClient, processor, configuration);
        }

        // <Delegate>
        /// <summary>
        /// The delegate receives batches of changes as they are generated in the change feed and can process them.
        /// </summary>
        static async Task HandleChangesAsync(
            ChangeFeedProcessorContext context,
            IReadOnlyCollection<ToDoItem> changes,
            CancellationToken cancellationToken)
        {
            Console.WriteLine($"Started handling changes for lease {context.LeaseToken}...");
            Console.WriteLine($"Change Feed request consumed {context.Headers.RequestCharge} RU.");
            // SessionToken if needed to enforce Session consistency on another client instance
            Console.WriteLine($"SessionToken ${context.Headers.Session}");

            // We may want to track any operation's Diagnostics that took longer than some threshold
            if (context.Diagnostics.GetClientElapsedTime() > TimeSpan.FromSeconds(1))
            {
                Console.WriteLine($"Change Feed request took longer than expected. Diagnostics:" + context.Diagnostics.ToString());
            }

            foreach (ToDoItem item in changes)
            {
                Console.WriteLine($"Detected operation for item with id {item.id}, created at {item.creationTime}.");
                // Simulate some asynchronous operation
                await Task.Delay(10);
            }

            Console.WriteLine("Finished handling changes.");
        }
        // </Delegate>

        /// <summary>
        /// Create required containers for the sample.
        /// Change Feed processing requires a source container to read the Change Feed from, and a container to store the state on, called leases.
        /// </summary>
        private static async Task InitializeContainersAsync(
            CosmosClient cosmosClient,
            IConfiguration configuration)
        {
            string databaseName = configuration["SourceDatabaseName"];
            string sourceContainerName = configuration["SourceContainerName"];
            string leaseContainerName = configuration["LeasesContainerName"];

            if (string.IsNullOrEmpty(databaseName)
                || string.IsNullOrEmpty(sourceContainerName)
                || string.IsNullOrEmpty(leaseContainerName))
            {
                throw new ArgumentNullException("'SourceDatabaseName', 'SourceContainerName', and 'LeasesContainerName' settings are required. Verify your configuration.");
            }

            Database database = await cosmosClient.CreateDatabaseIfNotExistsAsync(databaseName);

            await database.CreateContainerIfNotExistsAsync(new ContainerProperties(sourceContainerName, "/id"));

            await database.CreateContainerIfNotExistsAsync(new ContainerProperties(leaseContainerName, "/id"));
        }

        // <DefineProcessor>
        /// <summary>
        /// Start the Change Feed Processor to listen for changes and process them with the HandleChangesAsync implementation.
        /// </summary>
        private static async Task<ChangeFeedProcessor> StartChangeFeedProcessorAsync(
            CosmosClient cosmosClient,
            IConfiguration configuration)
        {
            string databaseName = configuration["SourceDatabaseName"];
            string sourceContainerName = configuration["SourceContainerName"];
            string leaseContainerName = configuration["LeasesContainerName"];

            Container leaseContainer = cosmosClient.GetContainer(databaseName, leaseContainerName);
            ChangeFeedProcessor changeFeedProcessor = cosmosClient.GetContainer(databaseName, sourceContainerName)
                .GetChangeFeedProcessorBuilder<ToDoItem>(processorName: "changeFeedSample", onChangesDelegate: HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();

            Console.WriteLine("Starting Change Feed Processor...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");
            return changeFeedProcessor;
        }
        // </DefineProcessor>

        /// <summary>
        /// Generate sample items based on user input.
        /// </summary>
        private static async Task GenerateItemsAsync(
            CosmosClient cosmosClient,
            ChangeFeedProcessor changeFeedProcessor,
            IConfiguration configuration)
        {
            string databaseName = configuration["SourceDatabaseName"];
            string sourceContainerName = configuration["SourceContainerName"];
            Container sourceContainer = cosmosClient.GetContainer(databaseName, sourceContainerName);
            while(true)
            {
                Console.WriteLine("Enter a number of items to insert in the container or 'exit' to stop:");
                string command = Console.ReadLine();
                if ("exit".Equals(command, StringComparison.InvariantCultureIgnoreCase))
                {
                    Console.WriteLine();
                    break;
                }

                if (int.TryParse(command, out int itemsToInsert))
                {
                    Console.WriteLine($"Generating {itemsToInsert} items...");
                    for (int i = 0; i < itemsToInsert; i++)
                    {
                        string id = Guid.NewGuid().ToString();
                        await sourceContainer.CreateItemAsync<ToDoItem>(
                            new ToDoItem()
                            {
                                id = id,
                                creationTime = DateTime.UtcNow
                            },
                            new PartitionKey(id));
                    }
                }
            }

            Console.WriteLine("Stopping Change Feed Processor...");
            await changeFeedProcessor.StopAsync();
            Console.WriteLine("Stopped Change Feed Processor.");
        }

        private static IConfiguration BuildConfiguration()
        {
            return new ConfigurationBuilder()
                .SetBasePath(Directory.GetCurrentDirectory())
                .AddJsonFile("appsettings.json", optional: true, reloadOnChange: true)
                .Build();
        }

        private static CosmosClient BuildCosmosClient(IConfiguration configuration)
        {
            if (string.IsNullOrEmpty(configuration["ConnectionString"]) || "<Your-Connection-String>".Equals(configuration["ConnectionString"]))
            {
                throw new ArgumentNullException("Missing 'ConnectionString' setting in configuration.");
            }

            return new CosmosClientBuilder(configuration["ConnectionString"])
                .Build();
        }
    }
}

To read using all versions and deletes mode, call GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes from the Container instance:

namespace ChangeFeedAllVersionsAndDeletes
{
    using System.Net;
    using Microsoft.Azure.Cosmos;
    using Microsoft.Extensions.Configuration;

    // ----------------------------------------------------------------------------------------------------------
    // Prerequisites - 
    // 
    // 1. An Azure Cosmos account - 
    //    https://learn.microsoft.com/azure/cosmos-db/create-cosmosdb-resources-portal
    //
    // 2. Microsoft.Azure.Cosmos NuGet package - 
    //    http://www.nuget.org/packages/Microsoft.Azure.Cosmos/ 
    // ----------------------------------------------------------------------------------------------------------
    // Sample - demonstrates common Change Feed operations using All Versions and Deletes mode
    //
    // 1. Listening for changes that happen after a Change Feed Processor is started.
    //
    // 2. Generate Estimator metrics to expose current Change Feed Processor progress as a push notification
    //
    // 3. Generate Estimator metrics to expose current Change Feed Processor progress on demand
    //
    // 4. Error handling and advanced logging
    //-----------------------------------------------------------------------------------------------------------

    internal class Program
    {
        private static readonly string databaseName = "changefeed-db";
        private static readonly string monitoredContainerPrefix = "monitored-";
        private static readonly string leasesContainer = "leases";
        private static readonly string partitionKeyPath = "/id";
        static async Task Main(string[] args)
        {
            try
            {
                IConfigurationRoot configuration = new ConfigurationBuilder()
                    .AddJsonFile("appSettings.json")
                    .Build();

                string endpoint = configuration["EndPointUrl"];
                if (string.IsNullOrEmpty(endpoint))
                {
                    throw new ArgumentNullException("Please specify a valid endpoint in the appSettings.json");
                }

                string authKey = configuration["AuthorizationKey"];
                if (string.IsNullOrEmpty(authKey) || string.Equals(authKey, "Super secret key"))
                {
                    throw new ArgumentException("Please specify a valid AuthorizationKey in the appSettings.json");
                }

                using (CosmosClient client = new CosmosClient(endpoint, authKey))
                {
                    Console.WriteLine($"\n1. Listening for changes that happen after a Change Feed Processor is started.");
                    await Program.RunBasicChangeFeed(monitoredContainerPrefix + "changefeed-basic", client);
                    Console.WriteLine($"\n2. Generate Estimator metrics to expose current Change Feed Processor progress as a push notification.");
                    await Program.RunEstimatorChangeFeed(monitoredContainerPrefix + "changefeed-estimator", client);
                    Console.WriteLine($"\n3. Generate Estimator metrics to expose current Change Feed Processor progress on demand.");
                    await Program.RunEstimatorPullChangeFeed(monitoredContainerPrefix + "changefeed-estimator-detailed", client);
                    Console.WriteLine($"\n4. Error handling and advanced logging.");
                    await Program.RunWithNotifications(monitoredContainerPrefix + "changefeed-logging", client);
                }
            }
            finally
            {
                Console.WriteLine("End of demo, press any key to exit.");
                Console.ReadKey();
            }
        }

        /// <summary>
        /// Basic change feed functionality using all versions and deletes mode.
        /// </summary>
        /// <remarks>
        /// When StartAsync is called, the Change Feed Processor starts an initialization process that can take several milliseconds, 
        /// in which it starts connections and initializes locks in the leases container.
        /// </remarks>
        public static async Task RunBasicChangeFeed(
            string containerName,
            CosmosClient client)
        {
            await Program.InitializeAsync(containerName, client);

            // <BasicInitialization>
            Container leaseContainer = client.GetContainer(Program.databaseName, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(Program.databaseName, containerName);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes<ToDoItem>(processorName: "changeFeedBasic", onChangesDelegate: Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();
            // </BasicInitialization>

            Console.WriteLine("Starting Change Feed Processor...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            Console.WriteLine("Generating 10 items that will be picked up by the delegate...");
            await Program.GenerateItems(10, monitoredContainer);
            Console.WriteLine("Updating 10 items that will be picked up by the delegate...");
            await Program.UpdateItems(10, monitoredContainer);
            Console.WriteLine("Deleting 10 items that will be picked up by the delegate...");
            await Program.DeleteItems(10, monitoredContainer);

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);
            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// Exposing progress with the Estimator.
        /// </summary>
        /// <remarks>
        /// The Estimator uses the same processorName and the same lease configuration as the existing processor to measure progress.
        /// </remarks>
        public static async Task RunEstimatorChangeFeed(
            string containerName,
            CosmosClient client)
        {
            await Program.InitializeAsync(containerName, client);

            // <StartProcessorEstimator>
            Container leaseContainer = client.GetContainer(Program.databaseName, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(Program.databaseName, containerName);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes<ToDoItem>(processorName: "changeFeedEstimator", onChangesDelegate: Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();
            // </StartProcessorEstimator>

            Console.WriteLine($"Starting Change Feed Processor...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            // Wait random time for the delegate to output all messages
            await Task.Delay(1000);

            // <StartEstimator>
            ChangeFeedProcessor changeFeedEstimator = monitoredContainer
                .GetChangeFeedEstimatorBuilder("changeFeedEstimator", Program.HandleEstimationAsync, TimeSpan.FromMilliseconds(1000))
                .WithLeaseContainer(leaseContainer)
                .Build();
            // </StartEstimator>

            Console.WriteLine($"Starting Change Feed Estimator...");
            await changeFeedEstimator.StartAsync();
            Console.WriteLine("Change Feed Estimator started.");

            Console.WriteLine("Generating 10 items that will be picked up by the delegate and reported by the Estimator...");
            await Program.GenerateItems(10, monitoredContainer);
            Console.WriteLine("Updating 10 items that will be picked up by the delegate and reported by the Estimator...");
            await Program.UpdateItems(10, monitoredContainer);
            Console.WriteLine("Deleting 10 items that will be picked up by the delegate and reported by the Estimator...");
            await Program.DeleteItems(10, monitoredContainer);

            // Wait random time for the delegate to output all messages
            await Task.Delay(5000);

            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
            await changeFeedEstimator.StopAsync();
        }

        /// <summary>
        /// Exposing progress with the Estimator with the detailed iterator.
        /// </summary>
        /// <remarks>
        /// The Estimator uses the same processorName and the same lease configuration as the existing processor to measure progress.
        /// The iterator exposes detailed, per-lease, information on estimation and ownership.
        /// </remarks>
        public static async Task RunEstimatorPullChangeFeed(
            string containerName,
            CosmosClient client)
        {
            await Program.InitializeAsync(containerName, client);

            // <StartProcessorEstimatorDetailed>
            Container leaseContainer = client.GetContainer(Program.databaseName, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(Program.databaseName, containerName);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes<ToDoItem>("changeFeedEstimatorPull", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();
            // </StartProcessorEstimatorDetailed>

            Console.WriteLine($"Starting Change Feed Processor...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            // Wait some seconds for instances to acquire leases
            await Task.Delay(5000);

            Console.WriteLine("Generating 10 items that will be picked up by the delegate and reported by the Estimator...");
            await Program.GenerateItems(10, monitoredContainer);
            Console.WriteLine("Updating 10 items that will be picked up by the delegate and reported by the Estimator...");
            await Program.UpdateItems(10, monitoredContainer);
            Console.WriteLine("Deleting 10 items that will be picked up by the delegate and reported by the Estimator...");
            await Program.DeleteItems(10, monitoredContainer);

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);

            // <StartEstimatorDetailed>
            ChangeFeedEstimator changeFeedEstimator = monitoredContainer
                .GetChangeFeedEstimator("changeFeedEstimator", leaseContainer);
            // </StartEstimatorDetailed>

            // <GetIteratorEstimatorDetailed>
            Console.WriteLine("Checking estimation...");
            using FeedIterator<ChangeFeedProcessorState> estimatorIterator = changeFeedEstimator.GetCurrentStateIterator();
            while (estimatorIterator.HasMoreResults)
            {
                FeedResponse<ChangeFeedProcessorState> states = await estimatorIterator.ReadNextAsync();
                foreach (ChangeFeedProcessorState leaseState in states)
                {
                    string host = leaseState.InstanceName == null ? $"not owned by any host currently" : $"owned by host {leaseState.InstanceName}";
                    Console.WriteLine($"Lease [{leaseState.LeaseToken}] {host} reports {leaseState.EstimatedLag} as estimated lag.");
                }
            }
            // </GetIteratorEstimatorDetailed>

            Console.WriteLine("Stopping processor.");
            await changeFeedProcessor.StopAsync();

            // Wait for processor to shutdown completely so the next items generate lag
            await Task.Delay(7500);

            Console.WriteLine("Generating 10 items that will be picked up by the delegate and reported by the Estimator...");
            await Program.GenerateItems(10, monitoredContainer);
            Console.WriteLine("Updating 10 items that will be picked up by the delegate and reported by the Estimator...");
            await Program.UpdateItems(10, monitoredContainer);
            Console.WriteLine("Deleting 10 items that will be picked up by the delegate and reported by the Estimator...");
            await Program.DeleteItems(10, monitoredContainer);

            Console.WriteLine("Checking estimation...");
            using FeedIterator<ChangeFeedProcessorState> estimatorIteratorAfter = changeFeedEstimator.GetCurrentStateIterator();
            while (estimatorIteratorAfter.HasMoreResults)
            {
                FeedResponse<ChangeFeedProcessorState> states = await estimatorIteratorAfter.ReadNextAsync();
                foreach (ChangeFeedProcessorState leaseState in states)
                {
                    // Host ownership should be empty as we have already stopped the estimator
                    string host = leaseState.InstanceName == null ? $"not owned by any host currently" : $"owned by host {leaseState.InstanceName}";
                    Console.WriteLine($"Lease [{leaseState.LeaseToken}] {host} reports {leaseState.EstimatedLag} as estimated lag.");
                }
            }

            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
        }

        /// <summary>
        /// Setup notification APIs for events.
        /// </summary>
        public static async Task RunWithNotifications(
            string containerName,
            CosmosClient client)
        {
            await Program.InitializeAsync(containerName, client);

            Container leaseContainer = client.GetContainer(Program.databaseName, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(Program.databaseName, containerName);

            ManualResetEventSlim manualResetEventSlim = new ManualResetEventSlim();
            Container.ChangeFeedHandler<ChangeFeedItem<ToDoItem>> handleChanges = (ChangeFeedProcessorContext context, IReadOnlyCollection<ChangeFeedItem<ToDoItem>> changes, CancellationToken cancellationToken) =>
            {
                Console.WriteLine($"Started handling changes for lease {context.LeaseToken} but throwing an exception to bubble to notifications.");
                manualResetEventSlim.Set();
                throw new Exception("This is an unhandled exception from inside the delegate");
            };

            // <StartWithNotifications>
            Container.ChangeFeedMonitorLeaseAcquireDelegate onLeaseAcquiredAsync = (string leaseToken) =>
            {
                Console.WriteLine($"Lease {leaseToken} is acquired and will start processing");
                return Task.CompletedTask;
            };

            Container.ChangeFeedMonitorLeaseReleaseDelegate onLeaseReleaseAsync = (string leaseToken) =>
            {
                Console.WriteLine($"Lease {leaseToken} is released and processing is stopped");
                return Task.CompletedTask;
            };

            Container.ChangeFeedMonitorErrorDelegate onErrorAsync = (string LeaseToken, Exception exception) =>
            {
                if (exception is ChangeFeedProcessorUserException userException)
                {
                    Console.WriteLine($"Lease {LeaseToken} processing failed with unhandled exception from user delegate {userException.InnerException}");
                }
                else
                {
                    Console.WriteLine($"Lease {LeaseToken} failed with {exception}");
                }

                return Task.CompletedTask;
            };

            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes<ToDoItem>("changeFeedNotifications", handleChanges)
                    .WithLeaseAcquireNotification(onLeaseAcquiredAsync)
                    .WithLeaseReleaseNotification(onLeaseReleaseAsync)
                    .WithErrorNotification(onErrorAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();
            // </StartWithNotifications>

            Console.WriteLine($"Starting Change Feed Processor with logging enabled...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            Console.WriteLine("Generating 10 items that will be picked up by the delegate and reported by the Estimator...");
            await Program.GenerateItems(10, monitoredContainer);
            Console.WriteLine("Updating 10 items that will be picked up by the delegate and reported by the Estimator...");
            await Program.UpdateItems(10, monitoredContainer);
            Console.WriteLine("Deleting 10 items that will be picked up by the delegate and reported by the Estimator...");
            await Program.DeleteItems(10, monitoredContainer);

            // Wait random time for the delegate to output all messages after initialization is done
            manualResetEventSlim.Wait();
            await Task.Delay(1000);
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// The delegate receives batches of changes as they are generated in the change feed and can process them.
        /// </summary>
        // <Delegate>
        static async Task HandleChangesAsync(ChangeFeedProcessorContext context, IReadOnlyCollection<ChangeFeedItem<ToDoItem>> changes, CancellationToken cancellationToken)
        {
            Console.WriteLine($"Started handling changes for lease {context.LeaseToken}...");
            Console.WriteLine($"Change Feed request consumed {context.Headers.RequestCharge} RU.");
            // SessionToken if needed to enforce Session consistency on another client instance
            Console.WriteLine($"SessionToken ${context.Headers.Session}");

            // We may want to track any operation's Diagnostics that took longer than some threshold
            if (context.Diagnostics.GetClientElapsedTime() > TimeSpan.FromSeconds(1))
            {
                Console.WriteLine($"Change Feed request took longer than expected. Diagnostics:" + context.Diagnostics.ToString());
            }

            foreach (ChangeFeedItem<ToDoItem> item in changes)
            {
                if (item.Metadata.OperationType == ChangeFeedOperationType.Delete)
                {
                    Console.WriteLine($"\tDetected {item.Metadata.OperationType} operation for item with id {item.Previous.id}.");
                }
                else
                {
                    Console.WriteLine($"\tDetected {item.Metadata.OperationType} operation for item with id {item.Current.id}.");
                }
                // Simulate work
                await Task.Delay(1);
            }
        }
        // </Delegate>

        /// <summary>
        /// The delegate for the Estimator receives a numeric representation of items pending to be read. 
        /// This is an estimate only and is not an exact count of outstanding items.
        /// </summary>
        // <EstimationDelegate>
        static async Task HandleEstimationAsync(long estimation, CancellationToken cancellationToken)
        {
            if (estimation > 0)
            {
                Console.WriteLine($"\tEstimator detected {estimation} items pending to be read by the Processor.");
            }

            await Task.Delay(0);
        }
        // </EstimationDelegate>

        private static async Task InitializeAsync(string containerName, CosmosClient client)
        {
            Database database = await client.CreateDatabaseIfNotExistsAsync(id: Program.databaseName, throughput: 1000);

            await database.CreateContainerIfNotExistsAsync(new ContainerProperties(containerName, Program.partitionKeyPath));

            await database.CreateContainerIfNotExistsAsync(new ContainerProperties(Program.leasesContainer, Program.partitionKeyPath));
        }

        private static async Task GenerateItems(int itemsToInsert, Container container)
        {
            await Task.Delay(500);
            for (int i = 0; i < itemsToInsert; i++)
            {
                await container.CreateItemAsync<ToDoItem>(
                    new ToDoItem()
                    {
                        id = i.ToString(),
                        CreationTime = DateTime.UtcNow
                    },
                    new PartitionKey(i.ToString()));
            }
        }

        private static async Task UpdateItems(int itemsToUpdate, Container container)
        {
            await Task.Delay(500);
            for (int i = 0; i < itemsToUpdate; i++)
            {
                await container.ReplaceItemAsync<ToDoItem>(
                    new ToDoItem()
                    {
                        id = i.ToString(),
                        CreationTime = DateTime.UtcNow,
                        Status = "updated"
                    },
                    i.ToString());
            }
        }

        private static async Task DeleteItems(int itemsToDelete, Container container)
        {
            await Task.Delay(500);
            for (int i = 0; i < itemsToDelete; i++)
            {
                await container.DeleteItemAsync<ToDoItem>(i.ToString(), new PartitionKey(i.ToString()));
            }
        }
    }

    // <Model>
    public class ToDoItem
    {
        public string? id { get; set; }
        public DateTime CreationTime { get; set; }
        public string? Status { get; set; }
    }
    // </Model>
}

For both modes, the first parameter is a distinct name that describes the goal of this processor. The second name is the delegate implementation that handles changes.

Here's an example of a delegate for latest version mode:

using System;
using System.Collections.Generic;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Azure.Cosmos;
using Microsoft.Azure.Cosmos.Fluent;
using Microsoft.Extensions.Configuration;

namespace ChangeFeedSample
{
    class Program
    {
        static async Task Main(string[] args)
        {
            IConfiguration configuration = BuildConfiguration();

            CosmosClient cosmosClient = BuildCosmosClient(configuration);

            await InitializeContainersAsync(cosmosClient, configuration);

            ChangeFeedProcessor processor = await StartChangeFeedProcessorAsync(cosmosClient, configuration);

            await GenerateItemsAsync(cosmosClient, processor, configuration);
        }

        // <Delegate>
        /// <summary>
        /// The delegate receives batches of changes as they are generated in the change feed and can process them.
        /// </summary>
        static async Task HandleChangesAsync(
            ChangeFeedProcessorContext context,
            IReadOnlyCollection<ToDoItem> changes,
            CancellationToken cancellationToken)
        {
            Console.WriteLine($"Started handling changes for lease {context.LeaseToken}...");
            Console.WriteLine($"Change Feed request consumed {context.Headers.RequestCharge} RU.");
            // SessionToken if needed to enforce Session consistency on another client instance
            Console.WriteLine($"SessionToken ${context.Headers.Session}");

            // We may want to track any operation's Diagnostics that took longer than some threshold
            if (context.Diagnostics.GetClientElapsedTime() > TimeSpan.FromSeconds(1))
            {
                Console.WriteLine($"Change Feed request took longer than expected. Diagnostics:" + context.Diagnostics.ToString());
            }

            foreach (ToDoItem item in changes)
            {
                Console.WriteLine($"Detected operation for item with id {item.id}, created at {item.creationTime}.");
                // Simulate some asynchronous operation
                await Task.Delay(10);
            }

            Console.WriteLine("Finished handling changes.");
        }
        // </Delegate>

        /// <summary>
        /// Create required containers for the sample.
        /// Change Feed processing requires a source container to read the Change Feed from, and a container to store the state on, called leases.
        /// </summary>
        private static async Task InitializeContainersAsync(
            CosmosClient cosmosClient,
            IConfiguration configuration)
        {
            string databaseName = configuration["SourceDatabaseName"];
            string sourceContainerName = configuration["SourceContainerName"];
            string leaseContainerName = configuration["LeasesContainerName"];

            if (string.IsNullOrEmpty(databaseName)
                || string.IsNullOrEmpty(sourceContainerName)
                || string.IsNullOrEmpty(leaseContainerName))
            {
                throw new ArgumentNullException("'SourceDatabaseName', 'SourceContainerName', and 'LeasesContainerName' settings are required. Verify your configuration.");
            }

            Database database = await cosmosClient.CreateDatabaseIfNotExistsAsync(databaseName);

            await database.CreateContainerIfNotExistsAsync(new ContainerProperties(sourceContainerName, "/id"));

            await database.CreateContainerIfNotExistsAsync(new ContainerProperties(leaseContainerName, "/id"));
        }

        // <DefineProcessor>
        /// <summary>
        /// Start the Change Feed Processor to listen for changes and process them with the HandleChangesAsync implementation.
        /// </summary>
        private static async Task<ChangeFeedProcessor> StartChangeFeedProcessorAsync(
            CosmosClient cosmosClient,
            IConfiguration configuration)
        {
            string databaseName = configuration["SourceDatabaseName"];
            string sourceContainerName = configuration["SourceContainerName"];
            string leaseContainerName = configuration["LeasesContainerName"];

            Container leaseContainer = cosmosClient.GetContainer(databaseName, leaseContainerName);
            ChangeFeedProcessor changeFeedProcessor = cosmosClient.GetContainer(databaseName, sourceContainerName)
                .GetChangeFeedProcessorBuilder<ToDoItem>(processorName: "changeFeedSample", onChangesDelegate: HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();

            Console.WriteLine("Starting Change Feed Processor...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");
            return changeFeedProcessor;
        }
        // </DefineProcessor>

        /// <summary>
        /// Generate sample items based on user input.
        /// </summary>
        private static async Task GenerateItemsAsync(
            CosmosClient cosmosClient,
            ChangeFeedProcessor changeFeedProcessor,
            IConfiguration configuration)
        {
            string databaseName = configuration["SourceDatabaseName"];
            string sourceContainerName = configuration["SourceContainerName"];
            Container sourceContainer = cosmosClient.GetContainer(databaseName, sourceContainerName);
            while(true)
            {
                Console.WriteLine("Enter a number of items to insert in the container or 'exit' to stop:");
                string command = Console.ReadLine();
                if ("exit".Equals(command, StringComparison.InvariantCultureIgnoreCase))
                {
                    Console.WriteLine();
                    break;
                }

                if (int.TryParse(command, out int itemsToInsert))
                {
                    Console.WriteLine($"Generating {itemsToInsert} items...");
                    for (int i = 0; i < itemsToInsert; i++)
                    {
                        string id = Guid.NewGuid().ToString();
                        await sourceContainer.CreateItemAsync<ToDoItem>(
                            new ToDoItem()
                            {
                                id = id,
                                creationTime = DateTime.UtcNow
                            },
                            new PartitionKey(id));
                    }
                }
            }

            Console.WriteLine("Stopping Change Feed Processor...");
            await changeFeedProcessor.StopAsync();
            Console.WriteLine("Stopped Change Feed Processor.");
        }

        private static IConfiguration BuildConfiguration()
        {
            return new ConfigurationBuilder()
                .SetBasePath(Directory.GetCurrentDirectory())
                .AddJsonFile("appsettings.json", optional: true, reloadOnChange: true)
                .Build();
        }

        private static CosmosClient BuildCosmosClient(IConfiguration configuration)
        {
            if (string.IsNullOrEmpty(configuration["ConnectionString"]) || "<Your-Connection-String>".Equals(configuration["ConnectionString"]))
            {
                throw new ArgumentNullException("Missing 'ConnectionString' setting in configuration.");
            }

            return new CosmosClientBuilder(configuration["ConnectionString"])
                .Build();
        }
    }
}

Afterward, you define the compute instance name or unique identifier by using WithInstanceName. The compute instance name should be unique and different for each compute instance you're deploying. You set the container to maintain the lease state by using WithLeaseContainer.

Calling Build gives you the processor instance that you can start by calling StartAsync.

Processing life cycle

The normal life cycle of a host instance is:

1. Read the change feed.
2. If there are no changes, sleep for a predefined amount of time (customizable by using WithPollInterval in the Builder) and go to #1.
3. If there are changes, send them to the delegate.
4. When the delegate finishes processing the changes successfully, update the lease store with the latest processed point in time and go to #1.

Error handling

The change feed processor is resilient to user code errors. If your delegate implementation has an unhandled exception (step #4), the thread that is processing that particular batch of changes stops, and a new thread is eventually created. The new thread checks the latest point in time that the lease store has saved for that range of partition key values. The new thread restarts from there, effectively sending the same batch of changes to the delegate. This behavior continues until your delegate processes the changes correctly, and it's the reason the change feed processor has an "at least once" guarantee.

To prevent your change feed processor from getting "stuck" continuously retrying the same batch of changes, you should add logic in your delegate code to write documents, upon exception, to an errored-message queue. This design ensures that you can keep track of unprocessed changes while still being able to continue to process future changes. The errored-message queue might be another Azure Cosmos DB container. The exact data store doesn't matter. You simply want the unprocessed changes to be persisted.

You also can use the change feed estimator to monitor the progress of your change feed processor instances as they read the change feed, or you can use life cycle notifications to detect underlying failures.

Life cycle notifications

You can connect the change feed processor to any relevant event in its life cycle. You can choose to be notified to one or all of them. The recommendation is to at least register the error notification:

• Register a handler for WithLeaseAcquireNotification to be notified when the current host acquires a lease to start processing it.
• Register a handler for WithLeaseReleaseNotification to be notified when the current host releases a lease and stops processing it.
• Register a handler for WithErrorNotification to be notified when the current host encounters an exception during processing. You need to be able to distinguish whether the source is the user delegate (an unhandled exception) or an error that the processor encounters when it tries to access the monitored container (for example, networking issues).

Life cycle notifications are available in both change feed modes. Here's an example of life cycle notifications in latest version mode:

namespace Cosmos.Samples.ChangeFeed
{
    using System;
    using System.Collections.Generic;
    using System.Net;
    using System.Threading;
    using System.Threading.Tasks;
    using Microsoft.Azure.Cosmos;
    using Microsoft.Azure.Documents.ChangeFeedProcessor.FeedProcessing;
    using Microsoft.Extensions.Configuration;
    using ChangeFeedProcessorLibrary = Microsoft.Azure.Documents.ChangeFeedProcessor;

    // ----------------------------------------------------------------------------------------------------------
    // Prerequisites - 
    // 
    // 1. An Azure Cosmos account - 
    //    https://learn.microsoft.com/azure/cosmos-db/create-cosmosdb-resources-portal
    //
    // 2. Microsoft.Azure.Cosmos NuGet package - 
    //    http://www.nuget.org/packages/Microsoft.Azure.Cosmos/ 
    // ----------------------------------------------------------------------------------------------------------
    // Sample - demonstrates common Change Feed operations
    //
    // 1. Listening for changes that happen after a Change Feed Processor is started.
    //
    // 2. Listening for changes that happen after a certain point in time.
    //
    // 3. Listening for changes that happen since the container was created.
    //
    // 4. Generate Estimator metrics to expose current Change Feed Processor progress as a push notification
    //
    // 5. Generate Estimator metrics to expose current Change Feed Processor progress on demand
    //
    // 6. Code migration template from existing Change Feed Processor library V2
    //
    // 7. Error handling and advanced logging
    //-----------------------------------------------------------------------------------------------------------


    class Program
    {
        private static readonly string monitoredContainer = "monitored";
        private static readonly string leasesContainer = "leases";
        private static readonly string partitionKeyPath = "/id";

        // Async main requires c# 7.1 which is set in the csproj with the LangVersion attribute
        static async Task Main(string[] args)
        {
            try
            {
                IConfigurationRoot configuration = new ConfigurationBuilder()
                    .AddJsonFile("appSettings.json")
                    .Build();

                string endpoint = configuration["EndPointUrl"];
                if (string.IsNullOrEmpty(endpoint))
                {
                    throw new ArgumentNullException("Please specify a valid endpoint in the appSettings.json");
                }

                string authKey = configuration["AuthorizationKey"];
                if (string.IsNullOrEmpty(authKey) || string.Equals(authKey, "Super secret key"))
                {
                    throw new ArgumentException("Please specify a valid AuthorizationKey in the appSettings.json");
                }

                using (CosmosClient client = new CosmosClient(endpoint, authKey))
                {
                    Console.WriteLine($"\n1. Listening for changes that happen after a Change Feed Processor is started.");
                    await Program.RunBasicChangeFeed("changefeed-basic", client);
                    Console.WriteLine($"\n2. Listening for changes that happen after a certain point in time.");
                    await Program.RunStartTimeChangeFeed("changefeed-time", client);
                    Console.WriteLine($"\n3. Listening for changes that happen since the container was created.");
                    await Program.RunStartFromBeginningChangeFeed("changefeed-beginning", client);
                    Console.WriteLine($"\n4. Generate Estimator metrics to expose current Change Feed Processor progress as a push notification.");
                    await Program.RunEstimatorChangeFeed("changefeed-estimator", client);
                    Console.WriteLine($"\n5. Generate Estimator metrics to expose current Change Feed Processor progress on demand.");
                    await Program.RunEstimatorPullChangeFeed("changefeed-estimator-detailed", client);
                    Console.WriteLine($"\n6. Code migration template from existing Change Feed Processor library V2.");
                    await Program.RunMigrationSample("changefeed-migration", client, configuration);
                    Console.WriteLine($"\n7. Error handling and advanced logging.");
                    await Program.RunWithNotifications("changefeed-logging", client);
                }
            }
            finally
            {
                Console.WriteLine("End of demo, press any key to exit.");
                Console.ReadKey();
            }
        }

        /// <summary>
        /// Basic change feed functionality.
        /// </summary>
        /// <remarks>
        /// When StartAsync is called, the Change Feed Processor starts an initialization process that can take several milliseconds, 
        /// in which it starts connections and initializes locks in the leases container.
        /// </remarks>
        public static async Task RunBasicChangeFeed(
            string databaseId, 
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);

            // <BasicInitialization>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedBasic", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();
            // </BasicInitialization>

            Console.WriteLine("Starting Change Feed Processor...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            Console.WriteLine("Generating 10 items that will be picked up by the delegate...");
            await Program.GenerateItems(10, monitoredContainer);

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);
            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// StartTime will make the Change Feed Processor start processing changes at a particular point in time, all previous changes are ignored.
        /// </summary>
        /// <remarks>
        /// StartTime only works if the leaseContainer is empty or contains no leases for the particular processor name.
        /// </remarks>
        public static async Task RunStartTimeChangeFeed(
            string databaseId,
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);
            Console.WriteLine("Generating 5 items that will not be picked up.");
            await Program.GenerateItems(5, client.GetContainer(databaseId, Program.monitoredContainer));
            Console.WriteLine($"Items generated at {DateTime.UtcNow}");
            // Generate a future point in time
            await Task.Delay(2000);
            DateTime particularPointInTime = DateTime.UtcNow;

            Console.WriteLine("Generating 5 items that will be picked up by the delegate...");
            await Program.GenerateItems(5, client.GetContainer(databaseId, Program.monitoredContainer));

            // <TimeInitialization>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedTime", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .WithStartTime(particularPointInTime)
                    .Build();
            // </TimeInitialization>

            Console.WriteLine($"Starting Change Feed Processor with changes after {particularPointInTime}...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);
            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// Reading the Change Feed since the beginning of time.
        /// </summary>
        /// <remarks>
        /// StartTime only works if the leaseContainer is empty or contains no leases for the particular processor name.
        /// </remarks>
        public static async Task RunStartFromBeginningChangeFeed(
            string databaseId,
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);
            Console.WriteLine("Generating 10 items that will be picked up by the delegate...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            // <StartFromBeginningInitialization>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedBeginning", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .WithStartTime(DateTime.MinValue.ToUniversalTime())
                    .Build();
            // </StartFromBeginningInitialization>

            Console.WriteLine($"Starting Change Feed Processor with changes since the beginning...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);
            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// Exposing progress with the Estimator.
        /// </summary>
        /// <remarks>
        /// The Estimator uses the same processorName and the same lease configuration as the existing processor to measure progress.
        /// </remarks>
        public static async Task RunEstimatorChangeFeed(
            string databaseId,
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);

            // <StartProcessorEstimator>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedEstimator", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();
            // </StartProcessorEstimator>

            Console.WriteLine($"Starting Change Feed Processor...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            Console.WriteLine("Generating 10 items that will be picked up by the delegate...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);

            // <StartEstimator>
            ChangeFeedProcessor changeFeedEstimator = monitoredContainer
                .GetChangeFeedEstimatorBuilder("changeFeedEstimator", Program.HandleEstimationAsync, TimeSpan.FromMilliseconds(1000))
                .WithLeaseContainer(leaseContainer)
                .Build();
            // </StartEstimator>

            Console.WriteLine($"Starting Change Feed Estimator...");
            await changeFeedEstimator.StartAsync();
            Console.WriteLine("Change Feed Estimator started.");

            Console.WriteLine("Generating 10 items that will be picked up by the delegate and reported by the Estimator...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
            await changeFeedEstimator.StopAsync();
        }

        /// <summary>
        /// Exposing progress with the Estimator with the detailed iterator.
        /// </summary>
        /// <remarks>
        /// The Estimator uses the same processorName and the same lease configuration as the existing processor to measure progress.
        /// The iterator exposes detailed, per-lease, information on estimation and ownership.
        /// </remarks>
        public static async Task RunEstimatorPullChangeFeed(
            string databaseId,
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);

            // <StartProcessorEstimatorDetailed>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedEstimator", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();
            // </StartProcessorEstimatorDetailed>

            Console.WriteLine($"Starting Change Feed Processor...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            // Wait some seconds for instances to acquire leases
            await Task.Delay(5000);

            Console.WriteLine("Generating 10 items that will be picked up by the delegate...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);

            // <StartEstimatorDetailed>
            ChangeFeedEstimator changeFeedEstimator = monitoredContainer
                .GetChangeFeedEstimator("changeFeedEstimator", leaseContainer);
            // </StartEstimatorDetailed>

            // <GetIteratorEstimatorDetailed>
            Console.WriteLine("Checking estimation...");
            using FeedIterator<ChangeFeedProcessorState> estimatorIterator = changeFeedEstimator.GetCurrentStateIterator();
            while (estimatorIterator.HasMoreResults)
            {
                FeedResponse<ChangeFeedProcessorState> states = await estimatorIterator.ReadNextAsync();
                foreach (ChangeFeedProcessorState leaseState in states)
                {
                    string host = leaseState.InstanceName == null ? $"not owned by any host currently" : $"owned by host {leaseState.InstanceName}";
                    Console.WriteLine($"Lease [{leaseState.LeaseToken}] {host} reports {leaseState.EstimatedLag} as estimated lag.");
                }
            }
            // </GetIteratorEstimatorDetailed>

            Console.WriteLine("Stopping processor to show how the lag increases if no processing is happening.");
            await changeFeedProcessor.StopAsync();

            // Wait for processor to shutdown completely so the next items generate lag
            await Task.Delay(5000);

            Console.WriteLine("Generating 10 items that will be seen by the Estimator...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            Console.WriteLine("Checking estimation...");
            using FeedIterator<ChangeFeedProcessorState> estimatorIteratorAfter = changeFeedEstimator.GetCurrentStateIterator();
            while (estimatorIteratorAfter.HasMoreResults)
            {
                FeedResponse<ChangeFeedProcessorState> states = await estimatorIteratorAfter.ReadNextAsync();
                foreach (ChangeFeedProcessorState leaseState in states)
                {
                    // Host ownership should be empty as we have already stopped the estimator
                    string host = leaseState.InstanceName == null ? $"not owned by any host currently" : $"owned by host {leaseState.InstanceName}";
                    Console.WriteLine($"Lease [{leaseState.LeaseToken}] {host} reports {leaseState.EstimatedLag} as estimated lag.");
                }
            }



            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
        }

        /// <summary>
        /// Example of a code migration template from Change Feed Processor V2 to SDK V3.
        /// </summary>
        /// <returns></returns>
        public static async Task RunMigrationSample(
            string databaseId,
            CosmosClient client,
            IConfigurationRoot configuration)
        {
            await Program.InitializeAsync(databaseId, client);

            Console.WriteLine("Generating 10 items that will be picked up by the old Change Feed Processor library...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            // This is how you would initialize the processor in V2
            // <ChangeFeedProcessorLibrary>
            ChangeFeedProcessorLibrary.DocumentCollectionInfo monitoredCollectionInfo = new ChangeFeedProcessorLibrary.DocumentCollectionInfo()
            {
                DatabaseName = databaseId,
                CollectionName = Program.monitoredContainer,
                Uri = new Uri(configuration["EndPointUrl"]),
                MasterKey = configuration["AuthorizationKey"]
            };

            ChangeFeedProcessorLibrary.DocumentCollectionInfo leaseCollectionInfo = new ChangeFeedProcessorLibrary.DocumentCollectionInfo()
            {
                DatabaseName = databaseId,
                CollectionName = Program.leasesContainer,
                Uri = new Uri(configuration["EndPointUrl"]),
                MasterKey = configuration["AuthorizationKey"]
            };

            ChangeFeedProcessorLibrary.ChangeFeedProcessorBuilder builder = new ChangeFeedProcessorLibrary.ChangeFeedProcessorBuilder();
            var oldChangeFeedProcessor = await builder
                .WithHostName("consoleHost")
                .WithProcessorOptions(new ChangeFeedProcessorLibrary.ChangeFeedProcessorOptions
                {
                    StartFromBeginning = true,
                    LeasePrefix = "MyLeasePrefix",
                    MaxItemCount = 10,
                    FeedPollDelay = TimeSpan.FromSeconds(1)
                })
                .WithFeedCollection(monitoredCollectionInfo)
                .WithLeaseCollection(leaseCollectionInfo)
                .WithObserver<ChangeFeedObserver>()
                .BuildAsync();
            // </ChangeFeedProcessorLibrary>

            await oldChangeFeedProcessor.StartAsync();

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);
            Console.WriteLine("Now we will stop the V2 Processor and start a V3 with the same parameters to pick up from the same state, press any key to continue...");
            Console.ReadKey();
            await oldChangeFeedProcessor.StopAsync();

            Console.WriteLine("Generating 5 items that will be picked up by the new Change Feed Processor...");
            await Program.GenerateItems(5, client.GetContainer(databaseId, Program.monitoredContainer));

            // This is how you would do the same initialization in V3
            // <ChangeFeedProcessorMigrated>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("MyLeasePrefix", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .WithMaxItems(10)
                    .WithPollInterval(TimeSpan.FromSeconds(1))
                    .WithStartTime(DateTime.MinValue.ToUniversalTime())
                    .Build();
            // </ChangeFeedProcessorMigrated>

            await changeFeedProcessor.StartAsync();

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);
            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// Setup notification APIs for events.
        /// </summary>
        public static async Task RunWithNotifications(
            string databaseId,
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);

            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);

            ManualResetEventSlim manualResetEventSlim = new ManualResetEventSlim();
            Container.ChangeFeedHandler<ToDoItem> handleChanges = (ChangeFeedProcessorContext context, IReadOnlyCollection<ToDoItem> changes, CancellationToken cancellationToken) =>
            {
                Console.WriteLine($"Started handling changes for lease {context.LeaseToken} but throwing an exception to bubble to notifications.");
                manualResetEventSlim.Set();
                throw new Exception("This is an unhandled exception from inside the delegate");
            };

            // <StartWithNotifications>
            Container.ChangeFeedMonitorLeaseAcquireDelegate onLeaseAcquiredAsync = (string leaseToken) =>
            {
                Console.WriteLine($"Lease {leaseToken} is acquired and will start processing");
                return Task.CompletedTask;
            };

            Container.ChangeFeedMonitorLeaseReleaseDelegate onLeaseReleaseAsync = (string leaseToken) =>
            {
                Console.WriteLine($"Lease {leaseToken} is released and processing is stopped");
                return Task.CompletedTask;
            };

            Container.ChangeFeedMonitorErrorDelegate onErrorAsync = (string LeaseToken, Exception exception) =>
            {
                if (exception is ChangeFeedProcessorUserException userException)
                {
                    Console.WriteLine($"Lease {LeaseToken} processing failed with unhandled exception from user delegate {userException.InnerException}");
                }
                else
                {
                    Console.WriteLine($"Lease {LeaseToken} failed with {exception}");
                }

                return Task.CompletedTask;
            };

            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedNotifications", handleChanges)
                    .WithLeaseAcquireNotification(onLeaseAcquiredAsync)
                    .WithLeaseReleaseNotification(onLeaseReleaseAsync)
                    .WithErrorNotification(onErrorAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();
            // </StartWithNotifications>

            Console.WriteLine($"Starting Change Feed Processor with logging enabled...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");
            Console.WriteLine("Generating 10 items that will be picked up by the delegate...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));
            // Wait random time for the delegate to output all messages after initialization is done
            manualResetEventSlim.Wait();
            await Task.Delay(1000);
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// The delegate receives batches of changes as they are generated in the change feed and can process them.
        /// </summary>
        // <Delegate>
        static async Task HandleChangesAsync(ChangeFeedProcessorContext context, IReadOnlyCollection<ToDoItem> changes, CancellationToken cancellationToken)
        {
            Console.WriteLine($"Started handling changes for lease {context.LeaseToken}...");
            Console.WriteLine($"Change Feed request consumed {context.Headers.RequestCharge} RU.");
            // SessionToken if needed to enforce Session consistency on another client instance
            Console.WriteLine($"SessionToken ${context.Headers.Session}");

            // We may want to track any operation's Diagnostics that took longer than some threshold
            if (context.Diagnostics.GetClientElapsedTime() > TimeSpan.FromSeconds(1))
            {
                Console.WriteLine($"Change Feed request took longer than expected. Diagnostics:" + context.Diagnostics.ToString());
            }

            foreach (ToDoItem item in changes)
            {
                Console.WriteLine($"\tDetected operation for item with id {item.id}, created at {item.creationTime}.");
                // Simulate work
                await Task.Delay(1);
            }
        }
        // </Delegate>

        /// <summary>
        /// The delegate for the Estimator receives a numeric representation of items pending to be read.
        /// </summary>
        // <EstimationDelegate>
        static async Task HandleEstimationAsync(long estimation, CancellationToken cancellationToken)
        {
            if (estimation > 0)
            {
                Console.WriteLine($"\tEstimator detected {estimation} items pending to be read by the Processor.");
            }

            await Task.Delay(0);
        }
        // </EstimationDelegate>

        private static async Task GenerateItems(
            int itemsToInsert, 
            Container container)
        {
            await Task.Delay(500);
            for (int i = 0; i < itemsToInsert; i++)
            {
                string id = Guid.NewGuid().ToString();
                await container.CreateItemAsync<ToDoItem>(
                    new ToDoItem()
                    {
                        id = id,
                        creationTime = DateTime.UtcNow
                    },
                    new PartitionKey(id));
            }
        }

        private static async Task InitializeAsync(
            string databaseId,
            CosmosClient client)
        {
            Database database;
            // Recreate database
            try
            {
                database = await client.GetDatabase(databaseId).ReadAsync();
                await database.DeleteAsync();
            }
            catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.NotFound)
            {
            }

            database = await client.CreateDatabaseAsync(databaseId);

            await database.CreateContainerIfNotExistsAsync(new ContainerProperties(Program.monitoredContainer, Program.partitionKeyPath));

            await database.CreateContainerIfNotExistsAsync(new ContainerProperties(Program.leasesContainer, Program.partitionKeyPath));
        }
    }

    // <Model>
    public class ToDoItem
    {
        public string id { get; set; }

        public DateTime creationTime { get; set; }
    }
    // </Model>

    internal class ChangeFeedObserver : IChangeFeedObserver
    {
        public Task CloseAsync(IChangeFeedObserverContext context, ChangeFeedObserverCloseReason reason)
        {
            return Task.CompletedTask;
        }

        public Task OpenAsync(IChangeFeedObserverContext context)
        {
            return Task.CompletedTask;
        }

        public Task ProcessChangesAsync(IChangeFeedObserverContext context, IReadOnlyList<Microsoft.Azure.Documents.Document> docs, CancellationToken cancellationToken)
        {
            foreach (Microsoft.Azure.Documents.Document doc in docs)
            {
                Console.WriteLine($"\t[OLD Processor] Detected operation for item with id {doc.Id}, created at {doc.GetPropertyValue<DateTime>("creationTime")}.");
            }

            return Task.CompletedTask;
        }
    }
}

Deployment unit

A single change feed processor deployment unit consists of one or more compute instances that have the same value for processorName and the same lease container configuration, but different instance names. You can have many deployment units in which each unit has a different business flow for the changes and each deployment unit consists of one or more instances.

For example, you might have one deployment unit that triggers an external API each time there's a change in your container. Another deployment unit might move data in real time each time there's a change. When a change happens in your monitored container, all your deployment units are notified.

Dynamic scaling

As mentioned earlier, within a deployment unit, you can have one or more compute instances. To take advantage of the compute distribution within the deployment unit, the only key requirements are that:

• All instances should have the same lease container configuration.
• All instances should have the same value for processorName.
• Each instance needs to have a different instance name (WithInstanceName).

If these three conditions apply, then the change feed processor distributes all the leases that are in the lease container across all running instances of that deployment unit, and it parallelizes compute by using an equal-distribution algorithm. A lease is owned by one instance at any time, so the number of instances shouldn't be greater than the number of leases.

The number of instances can grow and shrink. The change feed processor dynamically adjusts the load by redistributing it accordingly.

Moreover, the change feed processor can dynamically adjust a container's scale if the container's throughput or storage increases. When your container grows, the change feed processor transparently handles the scenario by dynamically increasing the leases and distributing the new leases among existing instances.

Starting time

By default, when a change feed processor starts for the first time, it initializes the lease container and starts its processing life cycle. Any changes that happened in the monitored container before the change feed processor is initialized for the first time aren't detected.

Reading from a previous date and time

It's possible to initialize the change feed processor to read changes starting at a specific date and time by passing an instance of DateTime to the WithStartTime builder extension:

namespace Cosmos.Samples.ChangeFeed
{
    using System;
    using System.Collections.Generic;
    using System.Net;
    using System.Threading;
    using System.Threading.Tasks;
    using Microsoft.Azure.Cosmos;
    using Microsoft.Azure.Documents.ChangeFeedProcessor.FeedProcessing;
    using Microsoft.Extensions.Configuration;
    using ChangeFeedProcessorLibrary = Microsoft.Azure.Documents.ChangeFeedProcessor;

    // ----------------------------------------------------------------------------------------------------------
    // Prerequisites - 
    // 
    // 1. An Azure Cosmos account - 
    //    https://learn.microsoft.com/azure/cosmos-db/create-cosmosdb-resources-portal
    //
    // 2. Microsoft.Azure.Cosmos NuGet package - 
    //    http://www.nuget.org/packages/Microsoft.Azure.Cosmos/ 
    // ----------------------------------------------------------------------------------------------------------
    // Sample - demonstrates common Change Feed operations
    //
    // 1. Listening for changes that happen after a Change Feed Processor is started.
    //
    // 2. Listening for changes that happen after a certain point in time.
    //
    // 3. Listening for changes that happen since the container was created.
    //
    // 4. Generate Estimator metrics to expose current Change Feed Processor progress as a push notification
    //
    // 5. Generate Estimator metrics to expose current Change Feed Processor progress on demand
    //
    // 6. Code migration template from existing Change Feed Processor library V2
    //
    // 7. Error handling and advanced logging
    //-----------------------------------------------------------------------------------------------------------


    class Program
    {
        private static readonly string monitoredContainer = "monitored";
        private static readonly string leasesContainer = "leases";
        private static readonly string partitionKeyPath = "/id";

        // Async main requires c# 7.1 which is set in the csproj with the LangVersion attribute
        static async Task Main(string[] args)
        {
            try
            {
                IConfigurationRoot configuration = new ConfigurationBuilder()
                    .AddJsonFile("appSettings.json")
                    .Build();

                string endpoint = configuration["EndPointUrl"];
                if (string.IsNullOrEmpty(endpoint))
                {
                    throw new ArgumentNullException("Please specify a valid endpoint in the appSettings.json");
                }

                string authKey = configuration["AuthorizationKey"];
                if (string.IsNullOrEmpty(authKey) || string.Equals(authKey, "Super secret key"))
                {
                    throw new ArgumentException("Please specify a valid AuthorizationKey in the appSettings.json");
                }

                using (CosmosClient client = new CosmosClient(endpoint, authKey))
                {
                    Console.WriteLine($"\n1. Listening for changes that happen after a Change Feed Processor is started.");
                    await Program.RunBasicChangeFeed("changefeed-basic", client);
                    Console.WriteLine($"\n2. Listening for changes that happen after a certain point in time.");
                    await Program.RunStartTimeChangeFeed("changefeed-time", client);
                    Console.WriteLine($"\n3. Listening for changes that happen since the container was created.");
                    await Program.RunStartFromBeginningChangeFeed("changefeed-beginning", client);
                    Console.WriteLine($"\n4. Generate Estimator metrics to expose current Change Feed Processor progress as a push notification.");
                    await Program.RunEstimatorChangeFeed("changefeed-estimator", client);
                    Console.WriteLine($"\n5. Generate Estimator metrics to expose current Change Feed Processor progress on demand.");
                    await Program.RunEstimatorPullChangeFeed("changefeed-estimator-detailed", client);
                    Console.WriteLine($"\n6. Code migration template from existing Change Feed Processor library V2.");
                    await Program.RunMigrationSample("changefeed-migration", client, configuration);
                    Console.WriteLine($"\n7. Error handling and advanced logging.");
                    await Program.RunWithNotifications("changefeed-logging", client);
                }
            }
            finally
            {
                Console.WriteLine("End of demo, press any key to exit.");
                Console.ReadKey();
            }
        }

        /// <summary>
        /// Basic change feed functionality.
        /// </summary>
        /// <remarks>
        /// When StartAsync is called, the Change Feed Processor starts an initialization process that can take several milliseconds, 
        /// in which it starts connections and initializes locks in the leases container.
        /// </remarks>
        public static async Task RunBasicChangeFeed(
            string databaseId, 
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);

            // <BasicInitialization>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedBasic", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();
            // </BasicInitialization>

            Console.WriteLine("Starting Change Feed Processor...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            Console.WriteLine("Generating 10 items that will be picked up by the delegate...");
            await Program.GenerateItems(10, monitoredContainer);

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);
            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// StartTime will make the Change Feed Processor start processing changes at a particular point in time, all previous changes are ignored.
        /// </summary>
        /// <remarks>
        /// StartTime only works if the leaseContainer is empty or contains no leases for the particular processor name.
        /// </remarks>
        public static async Task RunStartTimeChangeFeed(
            string databaseId,
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);
            Console.WriteLine("Generating 5 items that will not be picked up.");
            await Program.GenerateItems(5, client.GetContainer(databaseId, Program.monitoredContainer));
            Console.WriteLine($"Items generated at {DateTime.UtcNow}");
            // Generate a future point in time
            await Task.Delay(2000);
            DateTime particularPointInTime = DateTime.UtcNow;

            Console.WriteLine("Generating 5 items that will be picked up by the delegate...");
            await Program.GenerateItems(5, client.GetContainer(databaseId, Program.monitoredContainer));

            // <TimeInitialization>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedTime", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .WithStartTime(particularPointInTime)
                    .Build();
            // </TimeInitialization>

            Console.WriteLine($"Starting Change Feed Processor with changes after {particularPointInTime}...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);
            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// Reading the Change Feed since the beginning of time.
        /// </summary>
        /// <remarks>
        /// StartTime only works if the leaseContainer is empty or contains no leases for the particular processor name.
        /// </remarks>
        public static async Task RunStartFromBeginningChangeFeed(
            string databaseId,
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);
            Console.WriteLine("Generating 10 items that will be picked up by the delegate...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            // <StartFromBeginningInitialization>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedBeginning", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .WithStartTime(DateTime.MinValue.ToUniversalTime())
                    .Build();
            // </StartFromBeginningInitialization>

            Console.WriteLine($"Starting Change Feed Processor with changes since the beginning...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);
            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// Exposing progress with the Estimator.
        /// </summary>
        /// <remarks>
        /// The Estimator uses the same processorName and the same lease configuration as the existing processor to measure progress.
        /// </remarks>
        public static async Task RunEstimatorChangeFeed(
            string databaseId,
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);

            // <StartProcessorEstimator>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedEstimator", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();
            // </StartProcessorEstimator>

            Console.WriteLine($"Starting Change Feed Processor...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            Console.WriteLine("Generating 10 items that will be picked up by the delegate...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);

            // <StartEstimator>
            ChangeFeedProcessor changeFeedEstimator = monitoredContainer
                .GetChangeFeedEstimatorBuilder("changeFeedEstimator", Program.HandleEstimationAsync, TimeSpan.FromMilliseconds(1000))
                .WithLeaseContainer(leaseContainer)
                .Build();
            // </StartEstimator>

            Console.WriteLine($"Starting Change Feed Estimator...");
            await changeFeedEstimator.StartAsync();
            Console.WriteLine("Change Feed Estimator started.");

            Console.WriteLine("Generating 10 items that will be picked up by the delegate and reported by the Estimator...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
            await changeFeedEstimator.StopAsync();
        }

        /// <summary>
        /// Exposing progress with the Estimator with the detailed iterator.
        /// </summary>
        /// <remarks>
        /// The Estimator uses the same processorName and the same lease configuration as the existing processor to measure progress.
        /// The iterator exposes detailed, per-lease, information on estimation and ownership.
        /// </remarks>
        public static async Task RunEstimatorPullChangeFeed(
            string databaseId,
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);

            // <StartProcessorEstimatorDetailed>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedEstimator", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();
            // </StartProcessorEstimatorDetailed>

            Console.WriteLine($"Starting Change Feed Processor...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            // Wait some seconds for instances to acquire leases
            await Task.Delay(5000);

            Console.WriteLine("Generating 10 items that will be picked up by the delegate...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);

            // <StartEstimatorDetailed>
            ChangeFeedEstimator changeFeedEstimator = monitoredContainer
                .GetChangeFeedEstimator("changeFeedEstimator", leaseContainer);
            // </StartEstimatorDetailed>

            // <GetIteratorEstimatorDetailed>
            Console.WriteLine("Checking estimation...");
            using FeedIterator<ChangeFeedProcessorState> estimatorIterator = changeFeedEstimator.GetCurrentStateIterator();
            while (estimatorIterator.HasMoreResults)
            {
                FeedResponse<ChangeFeedProcessorState> states = await estimatorIterator.ReadNextAsync();
                foreach (ChangeFeedProcessorState leaseState in states)
                {
                    string host = leaseState.InstanceName == null ? $"not owned by any host currently" : $"owned by host {leaseState.InstanceName}";
                    Console.WriteLine($"Lease [{leaseState.LeaseToken}] {host} reports {leaseState.EstimatedLag} as estimated lag.");
                }
            }
            // </GetIteratorEstimatorDetailed>

            Console.WriteLine("Stopping processor to show how the lag increases if no processing is happening.");
            await changeFeedProcessor.StopAsync();

            // Wait for processor to shutdown completely so the next items generate lag
            await Task.Delay(5000);

            Console.WriteLine("Generating 10 items that will be seen by the Estimator...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            Console.WriteLine("Checking estimation...");
            using FeedIterator<ChangeFeedProcessorState> estimatorIteratorAfter = changeFeedEstimator.GetCurrentStateIterator();
            while (estimatorIteratorAfter.HasMoreResults)
            {
                FeedResponse<ChangeFeedProcessorState> states = await estimatorIteratorAfter.ReadNextAsync();
                foreach (ChangeFeedProcessorState leaseState in states)
                {
                    // Host ownership should be empty as we have already stopped the estimator
                    string host = leaseState.InstanceName == null ? $"not owned by any host currently" : $"owned by host {leaseState.InstanceName}";
                    Console.WriteLine($"Lease [{leaseState.LeaseToken}] {host} reports {leaseState.EstimatedLag} as estimated lag.");
                }
            }



            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
        }

        /// <summary>
        /// Example of a code migration template from Change Feed Processor V2 to SDK V3.
        /// </summary>
        /// <returns></returns>
        public static async Task RunMigrationSample(
            string databaseId,
            CosmosClient client,
            IConfigurationRoot configuration)
        {
            await Program.InitializeAsync(databaseId, client);

            Console.WriteLine("Generating 10 items that will be picked up by the old Change Feed Processor library...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            // This is how you would initialize the processor in V2
            // <ChangeFeedProcessorLibrary>
            ChangeFeedProcessorLibrary.DocumentCollectionInfo monitoredCollectionInfo = new ChangeFeedProcessorLibrary.DocumentCollectionInfo()
            {
                DatabaseName = databaseId,
                CollectionName = Program.monitoredContainer,
                Uri = new Uri(configuration["EndPointUrl"]),
                MasterKey = configuration["AuthorizationKey"]
            };

            ChangeFeedProcessorLibrary.DocumentCollectionInfo leaseCollectionInfo = new ChangeFeedProcessorLibrary.DocumentCollectionInfo()
            {
                DatabaseName = databaseId,
                CollectionName = Program.leasesContainer,
                Uri = new Uri(configuration["EndPointUrl"]),
                MasterKey = configuration["AuthorizationKey"]
            };

            ChangeFeedProcessorLibrary.ChangeFeedProcessorBuilder builder = new ChangeFeedProcessorLibrary.ChangeFeedProcessorBuilder();
            var oldChangeFeedProcessor = await builder
                .WithHostName("consoleHost")
                .WithProcessorOptions(new ChangeFeedProcessorLibrary.ChangeFeedProcessorOptions
                {
                    StartFromBeginning = true,
                    LeasePrefix = "MyLeasePrefix",
                    MaxItemCount = 10,
                    FeedPollDelay = TimeSpan.FromSeconds(1)
                })
                .WithFeedCollection(monitoredCollectionInfo)
                .WithLeaseCollection(leaseCollectionInfo)
                .WithObserver<ChangeFeedObserver>()
                .BuildAsync();
            // </ChangeFeedProcessorLibrary>

            await oldChangeFeedProcessor.StartAsync();

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);
            Console.WriteLine("Now we will stop the V2 Processor and start a V3 with the same parameters to pick up from the same state, press any key to continue...");
            Console.ReadKey();
            await oldChangeFeedProcessor.StopAsync();

            Console.WriteLine("Generating 5 items that will be picked up by the new Change Feed Processor...");
            await Program.GenerateItems(5, client.GetContainer(databaseId, Program.monitoredContainer));

            // This is how you would do the same initialization in V3
            // <ChangeFeedProcessorMigrated>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("MyLeasePrefix", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .WithMaxItems(10)
                    .WithPollInterval(TimeSpan.FromSeconds(1))
                    .WithStartTime(DateTime.MinValue.ToUniversalTime())
                    .Build();
            // </ChangeFeedProcessorMigrated>

            await changeFeedProcessor.StartAsync();

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);
            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// Setup notification APIs for events.
        /// </summary>
        public static async Task RunWithNotifications(
            string databaseId,
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);

            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);

            ManualResetEventSlim manualResetEventSlim = new ManualResetEventSlim();
            Container.ChangeFeedHandler<ToDoItem> handleChanges = (ChangeFeedProcessorContext context, IReadOnlyCollection<ToDoItem> changes, CancellationToken cancellationToken) =>
            {
                Console.WriteLine($"Started handling changes for lease {context.LeaseToken} but throwing an exception to bubble to notifications.");
                manualResetEventSlim.Set();
                throw new Exception("This is an unhandled exception from inside the delegate");
            };

            // <StartWithNotifications>
            Container.ChangeFeedMonitorLeaseAcquireDelegate onLeaseAcquiredAsync = (string leaseToken) =>
            {
                Console.WriteLine($"Lease {leaseToken} is acquired and will start processing");
                return Task.CompletedTask;
            };

            Container.ChangeFeedMonitorLeaseReleaseDelegate onLeaseReleaseAsync = (string leaseToken) =>
            {
                Console.WriteLine($"Lease {leaseToken} is released and processing is stopped");
                return Task.CompletedTask;
            };

            Container.ChangeFeedMonitorErrorDelegate onErrorAsync = (string LeaseToken, Exception exception) =>
            {
                if (exception is ChangeFeedProcessorUserException userException)
                {
                    Console.WriteLine($"Lease {LeaseToken} processing failed with unhandled exception from user delegate {userException.InnerException}");
                }
                else
                {
                    Console.WriteLine($"Lease {LeaseToken} failed with {exception}");
                }

                return Task.CompletedTask;
            };

            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedNotifications", handleChanges)
                    .WithLeaseAcquireNotification(onLeaseAcquiredAsync)
                    .WithLeaseReleaseNotification(onLeaseReleaseAsync)
                    .WithErrorNotification(onErrorAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();
            // </StartWithNotifications>

            Console.WriteLine($"Starting Change Feed Processor with logging enabled...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");
            Console.WriteLine("Generating 10 items that will be picked up by the delegate...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));
            // Wait random time for the delegate to output all messages after initialization is done
            manualResetEventSlim.Wait();
            await Task.Delay(1000);
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// The delegate receives batches of changes as they are generated in the change feed and can process them.
        /// </summary>
        // <Delegate>
        static async Task HandleChangesAsync(ChangeFeedProcessorContext context, IReadOnlyCollection<ToDoItem> changes, CancellationToken cancellationToken)
        {
            Console.WriteLine($"Started handling changes for lease {context.LeaseToken}...");
            Console.WriteLine($"Change Feed request consumed {context.Headers.RequestCharge} RU.");
            // SessionToken if needed to enforce Session consistency on another client instance
            Console.WriteLine($"SessionToken ${context.Headers.Session}");

            // We may want to track any operation's Diagnostics that took longer than some threshold
            if (context.Diagnostics.GetClientElapsedTime() > TimeSpan.FromSeconds(1))
            {
                Console.WriteLine($"Change Feed request took longer than expected. Diagnostics:" + context.Diagnostics.ToString());
            }

            foreach (ToDoItem item in changes)
            {
                Console.WriteLine($"\tDetected operation for item with id {item.id}, created at {item.creationTime}.");
                // Simulate work
                await Task.Delay(1);
            }
        }
        // </Delegate>

        /// <summary>
        /// The delegate for the Estimator receives a numeric representation of items pending to be read.
        /// </summary>
        // <EstimationDelegate>
        static async Task HandleEstimationAsync(long estimation, CancellationToken cancellationToken)
        {
            if (estimation > 0)
            {
                Console.WriteLine($"\tEstimator detected {estimation} items pending to be read by the Processor.");
            }

            await Task.Delay(0);
        }
        // </EstimationDelegate>

        private static async Task GenerateItems(
            int itemsToInsert, 
            Container container)
        {
            await Task.Delay(500);
            for (int i = 0; i < itemsToInsert; i++)
            {
                string id = Guid.NewGuid().ToString();
                await container.CreateItemAsync<ToDoItem>(
                    new ToDoItem()
                    {
                        id = id,
                        creationTime = DateTime.UtcNow
                    },
                    new PartitionKey(id));
            }
        }

        private static async Task InitializeAsync(
            string databaseId,
            CosmosClient client)
        {
            Database database;
            // Recreate database
            try
            {
                database = await client.GetDatabase(databaseId).ReadAsync();
                await database.DeleteAsync();
            }
            catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.NotFound)
            {
            }

            database = await client.CreateDatabaseAsync(databaseId);

            await database.CreateContainerIfNotExistsAsync(new ContainerProperties(Program.monitoredContainer, Program.partitionKeyPath));

            await database.CreateContainerIfNotExistsAsync(new ContainerProperties(Program.leasesContainer, Program.partitionKeyPath));
        }
    }

    // <Model>
    public class ToDoItem
    {
        public string id { get; set; }

        public DateTime creationTime { get; set; }
    }
    // </Model>

    internal class ChangeFeedObserver : IChangeFeedObserver
    {
        public Task CloseAsync(IChangeFeedObserverContext context, ChangeFeedObserverCloseReason reason)
        {
            return Task.CompletedTask;
        }

        public Task OpenAsync(IChangeFeedObserverContext context)
        {
            return Task.CompletedTask;
        }

        public Task ProcessChangesAsync(IChangeFeedObserverContext context, IReadOnlyList<Microsoft.Azure.Documents.Document> docs, CancellationToken cancellationToken)
        {
            foreach (Microsoft.Azure.Documents.Document doc in docs)
            {
                Console.WriteLine($"\t[OLD Processor] Detected operation for item with id {doc.Id}, created at {doc.GetPropertyValue<DateTime>("creationTime")}.");
            }

            return Task.CompletedTask;
        }
    }
}

The change feed processor is initialized for that specific date and time, and it starts to read the changes that happened afterward.

Reading from the beginning

In other scenarios, like in data migrations or if you're analyzing the entire history of a container, you need to read the change feed from the beginning of that container's lifetime. You can use WithStartTime on the builder extension, but pass DateTime.MinValue.ToUniversalTime(), which generates the UTC representation of the minimum DateTime value like in this example:

namespace Cosmos.Samples.ChangeFeed
{
    using System;
    using System.Collections.Generic;
    using System.Net;
    using System.Threading;
    using System.Threading.Tasks;
    using Microsoft.Azure.Cosmos;
    using Microsoft.Azure.Documents.ChangeFeedProcessor.FeedProcessing;
    using Microsoft.Extensions.Configuration;
    using ChangeFeedProcessorLibrary = Microsoft.Azure.Documents.ChangeFeedProcessor;

    // ----------------------------------------------------------------------------------------------------------
    // Prerequisites - 
    // 
    // 1. An Azure Cosmos account - 
    //    https://learn.microsoft.com/azure/cosmos-db/create-cosmosdb-resources-portal
    //
    // 2. Microsoft.Azure.Cosmos NuGet package - 
    //    http://www.nuget.org/packages/Microsoft.Azure.Cosmos/ 
    // ----------------------------------------------------------------------------------------------------------
    // Sample - demonstrates common Change Feed operations
    //
    // 1. Listening for changes that happen after a Change Feed Processor is started.
    //
    // 2. Listening for changes that happen after a certain point in time.
    //
    // 3. Listening for changes that happen since the container was created.
    //
    // 4. Generate Estimator metrics to expose current Change Feed Processor progress as a push notification
    //
    // 5. Generate Estimator metrics to expose current Change Feed Processor progress on demand
    //
    // 6. Code migration template from existing Change Feed Processor library V2
    //
    // 7. Error handling and advanced logging
    //-----------------------------------------------------------------------------------------------------------


    class Program
    {
        private static readonly string monitoredContainer = "monitored";
        private static readonly string leasesContainer = "leases";
        private static readonly string partitionKeyPath = "/id";

        // Async main requires c# 7.1 which is set in the csproj with the LangVersion attribute
        static async Task Main(string[] args)
        {
            try
            {
                IConfigurationRoot configuration = new ConfigurationBuilder()
                    .AddJsonFile("appSettings.json")
                    .Build();

                string endpoint = configuration["EndPointUrl"];
                if (string.IsNullOrEmpty(endpoint))
                {
                    throw new ArgumentNullException("Please specify a valid endpoint in the appSettings.json");
                }

                string authKey = configuration["AuthorizationKey"];
                if (string.IsNullOrEmpty(authKey) || string.Equals(authKey, "Super secret key"))
                {
                    throw new ArgumentException("Please specify a valid AuthorizationKey in the appSettings.json");
                }

                using (CosmosClient client = new CosmosClient(endpoint, authKey))
                {
                    Console.WriteLine($"\n1. Listening for changes that happen after a Change Feed Processor is started.");
                    await Program.RunBasicChangeFeed("changefeed-basic", client);
                    Console.WriteLine($"\n2. Listening for changes that happen after a certain point in time.");
                    await Program.RunStartTimeChangeFeed("changefeed-time", client);
                    Console.WriteLine($"\n3. Listening for changes that happen since the container was created.");
                    await Program.RunStartFromBeginningChangeFeed("changefeed-beginning", client);
                    Console.WriteLine($"\n4. Generate Estimator metrics to expose current Change Feed Processor progress as a push notification.");
                    await Program.RunEstimatorChangeFeed("changefeed-estimator", client);
                    Console.WriteLine($"\n5. Generate Estimator metrics to expose current Change Feed Processor progress on demand.");
                    await Program.RunEstimatorPullChangeFeed("changefeed-estimator-detailed", client);
                    Console.WriteLine($"\n6. Code migration template from existing Change Feed Processor library V2.");
                    await Program.RunMigrationSample("changefeed-migration", client, configuration);
                    Console.WriteLine($"\n7. Error handling and advanced logging.");
                    await Program.RunWithNotifications("changefeed-logging", client);
                }
            }
            finally
            {
                Console.WriteLine("End of demo, press any key to exit.");
                Console.ReadKey();
            }
        }

        /// <summary>
        /// Basic change feed functionality.
        /// </summary>
        /// <remarks>
        /// When StartAsync is called, the Change Feed Processor starts an initialization process that can take several milliseconds, 
        /// in which it starts connections and initializes locks in the leases container.
        /// </remarks>
        public static async Task RunBasicChangeFeed(
            string databaseId, 
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);

            // <BasicInitialization>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedBasic", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();
            // </BasicInitialization>

            Console.WriteLine("Starting Change Feed Processor...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            Console.WriteLine("Generating 10 items that will be picked up by the delegate...");
            await Program.GenerateItems(10, monitoredContainer);

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);
            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// StartTime will make the Change Feed Processor start processing changes at a particular point in time, all previous changes are ignored.
        /// </summary>
        /// <remarks>
        /// StartTime only works if the leaseContainer is empty or contains no leases for the particular processor name.
        /// </remarks>
        public static async Task RunStartTimeChangeFeed(
            string databaseId,
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);
            Console.WriteLine("Generating 5 items that will not be picked up.");
            await Program.GenerateItems(5, client.GetContainer(databaseId, Program.monitoredContainer));
            Console.WriteLine($"Items generated at {DateTime.UtcNow}");
            // Generate a future point in time
            await Task.Delay(2000);
            DateTime particularPointInTime = DateTime.UtcNow;

            Console.WriteLine("Generating 5 items that will be picked up by the delegate...");
            await Program.GenerateItems(5, client.GetContainer(databaseId, Program.monitoredContainer));

            // <TimeInitialization>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedTime", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .WithStartTime(particularPointInTime)
                    .Build();
            // </TimeInitialization>

            Console.WriteLine($"Starting Change Feed Processor with changes after {particularPointInTime}...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);
            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// Reading the Change Feed since the beginning of time.
        /// </summary>
        /// <remarks>
        /// StartTime only works if the leaseContainer is empty or contains no leases for the particular processor name.
        /// </remarks>
        public static async Task RunStartFromBeginningChangeFeed(
            string databaseId,
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);
            Console.WriteLine("Generating 10 items that will be picked up by the delegate...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            // <StartFromBeginningInitialization>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedBeginning", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .WithStartTime(DateTime.MinValue.ToUniversalTime())
                    .Build();
            // </StartFromBeginningInitialization>

            Console.WriteLine($"Starting Change Feed Processor with changes since the beginning...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);
            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// Exposing progress with the Estimator.
        /// </summary>
        /// <remarks>
        /// The Estimator uses the same processorName and the same lease configuration as the existing processor to measure progress.
        /// </remarks>
        public static async Task RunEstimatorChangeFeed(
            string databaseId,
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);

            // <StartProcessorEstimator>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedEstimator", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();
            // </StartProcessorEstimator>

            Console.WriteLine($"Starting Change Feed Processor...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            Console.WriteLine("Generating 10 items that will be picked up by the delegate...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);

            // <StartEstimator>
            ChangeFeedProcessor changeFeedEstimator = monitoredContainer
                .GetChangeFeedEstimatorBuilder("changeFeedEstimator", Program.HandleEstimationAsync, TimeSpan.FromMilliseconds(1000))
                .WithLeaseContainer(leaseContainer)
                .Build();
            // </StartEstimator>

            Console.WriteLine($"Starting Change Feed Estimator...");
            await changeFeedEstimator.StartAsync();
            Console.WriteLine("Change Feed Estimator started.");

            Console.WriteLine("Generating 10 items that will be picked up by the delegate and reported by the Estimator...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
            await changeFeedEstimator.StopAsync();
        }

        /// <summary>
        /// Exposing progress with the Estimator with the detailed iterator.
        /// </summary>
        /// <remarks>
        /// The Estimator uses the same processorName and the same lease configuration as the existing processor to measure progress.
        /// The iterator exposes detailed, per-lease, information on estimation and ownership.
        /// </remarks>
        public static async Task RunEstimatorPullChangeFeed(
            string databaseId,
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);

            // <StartProcessorEstimatorDetailed>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedEstimator", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();
            // </StartProcessorEstimatorDetailed>

            Console.WriteLine($"Starting Change Feed Processor...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");

            // Wait some seconds for instances to acquire leases
            await Task.Delay(5000);

            Console.WriteLine("Generating 10 items that will be picked up by the delegate...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);

            // <StartEstimatorDetailed>
            ChangeFeedEstimator changeFeedEstimator = monitoredContainer
                .GetChangeFeedEstimator("changeFeedEstimator", leaseContainer);
            // </StartEstimatorDetailed>

            // <GetIteratorEstimatorDetailed>
            Console.WriteLine("Checking estimation...");
            using FeedIterator<ChangeFeedProcessorState> estimatorIterator = changeFeedEstimator.GetCurrentStateIterator();
            while (estimatorIterator.HasMoreResults)
            {
                FeedResponse<ChangeFeedProcessorState> states = await estimatorIterator.ReadNextAsync();
                foreach (ChangeFeedProcessorState leaseState in states)
                {
                    string host = leaseState.InstanceName == null ? $"not owned by any host currently" : $"owned by host {leaseState.InstanceName}";
                    Console.WriteLine($"Lease [{leaseState.LeaseToken}] {host} reports {leaseState.EstimatedLag} as estimated lag.");
                }
            }
            // </GetIteratorEstimatorDetailed>

            Console.WriteLine("Stopping processor to show how the lag increases if no processing is happening.");
            await changeFeedProcessor.StopAsync();

            // Wait for processor to shutdown completely so the next items generate lag
            await Task.Delay(5000);

            Console.WriteLine("Generating 10 items that will be seen by the Estimator...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            Console.WriteLine("Checking estimation...");
            using FeedIterator<ChangeFeedProcessorState> estimatorIteratorAfter = changeFeedEstimator.GetCurrentStateIterator();
            while (estimatorIteratorAfter.HasMoreResults)
            {
                FeedResponse<ChangeFeedProcessorState> states = await estimatorIteratorAfter.ReadNextAsync();
                foreach (ChangeFeedProcessorState leaseState in states)
                {
                    // Host ownership should be empty as we have already stopped the estimator
                    string host = leaseState.InstanceName == null ? $"not owned by any host currently" : $"owned by host {leaseState.InstanceName}";
                    Console.WriteLine($"Lease [{leaseState.LeaseToken}] {host} reports {leaseState.EstimatedLag} as estimated lag.");
                }
            }



            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
        }

        /// <summary>
        /// Example of a code migration template from Change Feed Processor V2 to SDK V3.
        /// </summary>
        /// <returns></returns>
        public static async Task RunMigrationSample(
            string databaseId,
            CosmosClient client,
            IConfigurationRoot configuration)
        {
            await Program.InitializeAsync(databaseId, client);

            Console.WriteLine("Generating 10 items that will be picked up by the old Change Feed Processor library...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));

            // This is how you would initialize the processor in V2
            // <ChangeFeedProcessorLibrary>
            ChangeFeedProcessorLibrary.DocumentCollectionInfo monitoredCollectionInfo = new ChangeFeedProcessorLibrary.DocumentCollectionInfo()
            {
                DatabaseName = databaseId,
                CollectionName = Program.monitoredContainer,
                Uri = new Uri(configuration["EndPointUrl"]),
                MasterKey = configuration["AuthorizationKey"]
            };

            ChangeFeedProcessorLibrary.DocumentCollectionInfo leaseCollectionInfo = new ChangeFeedProcessorLibrary.DocumentCollectionInfo()
            {
                DatabaseName = databaseId,
                CollectionName = Program.leasesContainer,
                Uri = new Uri(configuration["EndPointUrl"]),
                MasterKey = configuration["AuthorizationKey"]
            };

            ChangeFeedProcessorLibrary.ChangeFeedProcessorBuilder builder = new ChangeFeedProcessorLibrary.ChangeFeedProcessorBuilder();
            var oldChangeFeedProcessor = await builder
                .WithHostName("consoleHost")
                .WithProcessorOptions(new ChangeFeedProcessorLibrary.ChangeFeedProcessorOptions
                {
                    StartFromBeginning = true,
                    LeasePrefix = "MyLeasePrefix",
                    MaxItemCount = 10,
                    FeedPollDelay = TimeSpan.FromSeconds(1)
                })
                .WithFeedCollection(monitoredCollectionInfo)
                .WithLeaseCollection(leaseCollectionInfo)
                .WithObserver<ChangeFeedObserver>()
                .BuildAsync();
            // </ChangeFeedProcessorLibrary>

            await oldChangeFeedProcessor.StartAsync();

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);
            Console.WriteLine("Now we will stop the V2 Processor and start a V3 with the same parameters to pick up from the same state, press any key to continue...");
            Console.ReadKey();
            await oldChangeFeedProcessor.StopAsync();

            Console.WriteLine("Generating 5 items that will be picked up by the new Change Feed Processor...");
            await Program.GenerateItems(5, client.GetContainer(databaseId, Program.monitoredContainer));

            // This is how you would do the same initialization in V3
            // <ChangeFeedProcessorMigrated>
            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("MyLeasePrefix", Program.HandleChangesAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .WithMaxItems(10)
                    .WithPollInterval(TimeSpan.FromSeconds(1))
                    .WithStartTime(DateTime.MinValue.ToUniversalTime())
                    .Build();
            // </ChangeFeedProcessorMigrated>

            await changeFeedProcessor.StartAsync();

            // Wait random time for the delegate to output all messages after initialization is done
            await Task.Delay(5000);
            Console.WriteLine("Press any key to continue with the next demo...");
            Console.ReadKey();
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// Setup notification APIs for events.
        /// </summary>
        public static async Task RunWithNotifications(
            string databaseId,
            CosmosClient client)
        {
            await Program.InitializeAsync(databaseId, client);

            Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
            Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);

            ManualResetEventSlim manualResetEventSlim = new ManualResetEventSlim();
            Container.ChangeFeedHandler<ToDoItem> handleChanges = (ChangeFeedProcessorContext context, IReadOnlyCollection<ToDoItem> changes, CancellationToken cancellationToken) =>
            {
                Console.WriteLine($"Started handling changes for lease {context.LeaseToken} but throwing an exception to bubble to notifications.");
                manualResetEventSlim.Set();
                throw new Exception("This is an unhandled exception from inside the delegate");
            };

            // <StartWithNotifications>
            Container.ChangeFeedMonitorLeaseAcquireDelegate onLeaseAcquiredAsync = (string leaseToken) =>
            {
                Console.WriteLine($"Lease {leaseToken} is acquired and will start processing");
                return Task.CompletedTask;
            };

            Container.ChangeFeedMonitorLeaseReleaseDelegate onLeaseReleaseAsync = (string leaseToken) =>
            {
                Console.WriteLine($"Lease {leaseToken} is released and processing is stopped");
                return Task.CompletedTask;
            };

            Container.ChangeFeedMonitorErrorDelegate onErrorAsync = (string LeaseToken, Exception exception) =>
            {
                if (exception is ChangeFeedProcessorUserException userException)
                {
                    Console.WriteLine($"Lease {LeaseToken} processing failed with unhandled exception from user delegate {userException.InnerException}");
                }
                else
                {
                    Console.WriteLine($"Lease {LeaseToken} failed with {exception}");
                }

                return Task.CompletedTask;
            };

            ChangeFeedProcessor changeFeedProcessor = monitoredContainer
                .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedNotifications", handleChanges)
                    .WithLeaseAcquireNotification(onLeaseAcquiredAsync)
                    .WithLeaseReleaseNotification(onLeaseReleaseAsync)
                    .WithErrorNotification(onErrorAsync)
                    .WithInstanceName("consoleHost")
                    .WithLeaseContainer(leaseContainer)
                    .Build();
            // </StartWithNotifications>

            Console.WriteLine($"Starting Change Feed Processor with logging enabled...");
            await changeFeedProcessor.StartAsync();
            Console.WriteLine("Change Feed Processor started.");
            Console.WriteLine("Generating 10 items that will be picked up by the delegate...");
            await Program.GenerateItems(10, client.GetContainer(databaseId, Program.monitoredContainer));
            // Wait random time for the delegate to output all messages after initialization is done
            manualResetEventSlim.Wait();
            await Task.Delay(1000);
            await changeFeedProcessor.StopAsync();
        }

        /// <summary>
        /// The delegate receives batches of changes as they are generated in the change feed and can process them.
        /// </summary>
        // <Delegate>
        static async Task HandleChangesAsync(ChangeFeedProcessorContext context, IReadOnlyCollection<ToDoItem> changes, CancellationToken cancellationToken)
        {
            Console.WriteLine($"Started handling changes for lease {context.LeaseToken}...");
            Console.WriteLine($"Change Feed request consumed {context.Headers.RequestCharge} RU.");
            // SessionToken if needed to enforce Session consistency on another client instance
            Console.WriteLine($"SessionToken ${context.Headers.Session}");

            // We may want to track any operation's Diagnostics that took longer than some threshold
            if (context.Diagnostics.GetClientElapsedTime() > TimeSpan.FromSeconds(1))
            {
                Console.WriteLine($"Change Feed request took longer than expected. Diagnostics:" + context.Diagnostics.ToString());
            }

            foreach (ToDoItem item in changes)
            {
                Console.WriteLine($"\tDetected operation for item with id {item.id}, created at {item.creationTime}.");
                // Simulate work
                await Task.Delay(1);
            }
        }
        // </Delegate>

        /// <summary>
        /// The delegate for the Estimator receives a numeric representation of items pending to be read.
        /// </summary>
        // <EstimationDelegate>
        static async Task HandleEstimationAsync(long estimation, CancellationToken cancellationToken)
        {
            if (estimation > 0)
            {
                Console.WriteLine($"\tEstimator detected {estimation} items pending to be read by the Processor.");
            }

            await Task.Delay(0);
        }
        // </EstimationDelegate>

        private static async Task GenerateItems(
            int itemsToInsert, 
            Container container)
        {
            await Task.Delay(500);
            for (int i = 0; i < itemsToInsert; i++)
            {
                string id = Guid.NewGuid().ToString();
                await container.CreateItemAsync<ToDoItem>(
                    new ToDoItem()
                    {
                        id = id,
                        creationTime = DateTime.UtcNow
                    },
                    new PartitionKey(id));
            }
        }

        private static async Task InitializeAsync(
            string databaseId,
            CosmosClient client)
        {
            Database database;
            // Recreate database
            try
            {
                database = await client.GetDatabase(databaseId).ReadAsync();
                await database.DeleteAsync();
            }
            catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.NotFound)
            {
            }

            database = await client.CreateDatabaseAsync(databaseId);

            await database.CreateContainerIfNotExistsAsync(new ContainerProperties(Program.monitoredContainer, Program.partitionKeyPath));

            await database.CreateContainerIfNotExistsAsync(new ContainerProperties(Program.leasesContainer, Program.partitionKeyPath));
        }
    }

    // <Model>
    public class ToDoItem
    {
        public string id { get; set; }

        public DateTime creationTime { get; set; }
    }
    // </Model>

    internal class ChangeFeedObserver : IChangeFeedObserver
    {
        public Task CloseAsync(IChangeFeedObserverContext context, ChangeFeedObserverCloseReason reason)
        {
            return Task.CompletedTask;
        }

        public Task OpenAsync(IChangeFeedObserverContext context)
        {
            return Task.CompletedTask;
        }

        public Task ProcessChangesAsync(IChangeFeedObserverContext context, IReadOnlyList<Microsoft.Azure.Documents.Document> docs, CancellationToken cancellationToken)
        {
            foreach (Microsoft.Azure.Documents.Document doc in docs)
            {
                Console.WriteLine($"\t[OLD Processor] Detected operation for item with id {doc.Id}, created at {doc.GetPropertyValue<DateTime>("creationTime")}.");
            }

            return Task.CompletedTask;
        }
    }
}

The change feed processor is initialized, and it starts reading changes from the beginning of the lifetime of the container.

For full working samples, see here. An example of a delegate implementation when reading the change feed in latest version mode is:

// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License.
package com.azure.cosmos.examples.changefeed;

import com.azure.cosmos.ChangeFeedProcessor;
import com.azure.cosmos.ChangeFeedProcessorBuilder;
import com.azure.cosmos.ConsistencyLevel;
import com.azure.cosmos.CosmosAsyncClient;
import com.azure.cosmos.CosmosAsyncContainer;
import com.azure.cosmos.CosmosAsyncDatabase;
import com.azure.cosmos.CosmosClientBuilder;
import com.azure.cosmos.CosmosException;
import com.azure.cosmos.ThroughputControlGroupConfig;
import com.azure.cosmos.ThroughputControlGroupConfigBuilder;
import com.azure.cosmos.examples.common.AccountSettings;
import com.azure.cosmos.examples.common.CustomPOJO2;
import com.azure.cosmos.implementation.Utils;
import com.azure.cosmos.models.ChangeFeedProcessorOptions;
import com.azure.cosmos.models.CosmosContainerProperties;
import com.azure.cosmos.models.CosmosContainerRequestOptions;
import com.azure.cosmos.models.CosmosContainerResponse;
import com.azure.cosmos.models.CosmosDatabaseResponse;
import com.azure.cosmos.models.PriorityLevel;
import com.azure.cosmos.models.ThroughputProperties;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.List;
import java.util.UUID;
import java.util.function.Consumer;

/**
 * Sample for Change Feed Processor.
 * This sample models an application where documents are being inserted into one container (the "feed container"),
 * and meanwhile another worker thread or worker application is pulling inserted documents from the feed container's Change Feed
 * and operating on them in some way. For one or more workers to process the Change Feed of a container, the workers must first contact the server
 * and "lease" access to monitor one or more partitions of the feed container. The Change Feed Processor Library
 * handles leasing automatically for you, however you must create a separate "lease container" where the Change Feed
 * Processor Library can store and track leases container partitions.
 */
public class SampleChangeFeedProcessor {

    public static int WAIT_FOR_WORK = 60000;
    public static final String DATABASE_NAME = "db_" + UUID.randomUUID();
    public static final String COLLECTION_NAME = "coll_" + UUID.randomUUID();
    private static final ObjectMapper OBJECT_MAPPER = Utils.getSimpleObjectMapper();
    protected static Logger logger = LoggerFactory.getLogger(SampleChangeFeedProcessor.class);

    private static boolean isWorkCompleted = false;

    private static ChangeFeedProcessorOptions options;



    public static void main(String[] args) {
        logger.info("Begin Sample");
        try {

            //By integrating throughput control with Change Feed Processor, users can set a low priority for
            //the change feed workload, ensuring that it does not impact other workloads running in parallel.
            //This prevents heavy throttling during times of high request unit (RU) usage.
            //Learn about throughput control here: https://aka.ms/JavaCosmosSparkThroughputControl
            //Learn about priority based execution here: https://aka.ms/CosmosPriorityBasedExecution
            // <ThroughputControlGroupConfig>
            ThroughputControlGroupConfig throughputControlGroupConfig =
                    new ThroughputControlGroupConfigBuilder()
                            .groupName("cfp")
                            .targetThroughput(300)
                            .priorityLevel(PriorityLevel.LOW)
                            .build();
            // </ThroughputControlGroupConfig>
            // <ChangeFeedProcessorOptions>
            options = new ChangeFeedProcessorOptions();
            options.setStartFromBeginning(false);
            options.setLeasePrefix("myChangeFeedDeploymentUnit");
            options.setFeedPollDelay(Duration.ofSeconds(5));
            options.setFeedPollThroughputControlConfig(throughputControlGroupConfig);
            // </ChangeFeedProcessorOptions>

            //Summary of the next four commands:
            //-Create an asynchronous Azure Cosmos DB client and database so that we can issue async requests to the DB
            //-Create a "feed container" and a "lease container" in the DB
            logger.info("Create CosmosClient");
            CosmosAsyncClient client = getCosmosClient();

            logger.info("Create sample's database: " + DATABASE_NAME);
            CosmosAsyncDatabase cosmosDatabase = createNewDatabase(client, DATABASE_NAME);

            logger.info("Create container for documents: " + COLLECTION_NAME);
            CosmosAsyncContainer feedContainer = createNewCollection(client, DATABASE_NAME, COLLECTION_NAME);

            logger.info("Create container for lease: " + COLLECTION_NAME + "-leases");
            CosmosAsyncContainer leaseContainer = createNewLeaseCollection(client, DATABASE_NAME, COLLECTION_NAME + "-leases");

            //Model of a worker thread or application which leases access to monitor one or more feed container
            //partitions via the Change Feed. In a real-world application you might deploy this code in an Azure function.
            //The next line causes the worker to create and start an instance of the Change Feed Processor. See the implementation of getChangeFeedProcessor() for guidance
            //on creating a handler for Change Feed events. In this stream, we also trigger the insertion of 10 documents on a separate
            //thread.
            // <StartChangeFeedProcessor>
            logger.info("Start Change Feed Processor on worker (handles changes asynchronously)");
            ChangeFeedProcessor changeFeedProcessorInstance = new ChangeFeedProcessorBuilder()
                .hostName("SampleHost_1")
                .feedContainer(feedContainer)
                .leaseContainer(leaseContainer)
                .handleChanges(handleChanges())
                .options(options)
                .buildChangeFeedProcessor();
            changeFeedProcessorInstance.start()
                                       .subscribeOn(Schedulers.boundedElastic())
                                       .subscribe();
            // </StartChangeFeedProcessor>

            //These two lines model an application which is inserting ten documents into the feed container
            logger.info("Start application that inserts documents into feed container");
            createNewDocumentsCustomPOJO(feedContainer, 10, Duration.ofSeconds(3));

            //This loop models the Worker main loop, which spins while its Change Feed Processor instance asynchronously
            //handles incoming Change Feed events from the feed container. Of course in this sample, polling
            //isWorkCompleted is unnecessary because items are being added to the feed container on the same thread, and you
            //can see just above isWorkCompleted is set to true.
            //But conceptually the worker is part of a different thread or application than the one which is inserting
            //into the feed container; so this code illustrates the worker waiting and listening for changes to the feed container
            long remainingWork = WAIT_FOR_WORK;
            while (!isWorkCompleted && remainingWork > 0) {
                Thread.sleep(100);
                remainingWork -= 100;
            }

            //When all documents have been processed, clean up
            if (isWorkCompleted) {
                changeFeedProcessorInstance.stop().subscribe();
            } else {
                throw new RuntimeException("The change feed processor initialization and automatic create document feeding process did not complete in the expected time");
            }

            logger.info("Delete sample's database: " + DATABASE_NAME);
            //deleteDatabase(cosmosDatabase);

            Thread.sleep(500);
        } catch (Exception e) {
            e.printStackTrace();
        }
        logger.info("End Sample");
    }

    // <Delegate>
    private static Consumer<List<JsonNode>> handleChanges() {
        return (List<JsonNode> docs) -> {
            logger.info("Start handleChanges()");

            for (JsonNode document : docs) {
                try {
                    //Change Feed hands the document to you in the form of a JsonNode
                    //As a developer you have two options for handling the JsonNode document provided to you by Change Feed
                    //One option is to operate on the document in the form of a JsonNode, as shown below. This is great
                    //especially if you do not have a single uniform data model for all documents.
                    logger.info("Document received: " + OBJECT_MAPPER.writerWithDefaultPrettyPrinter()
                            .writeValueAsString(document));

                    //You can also transform the JsonNode to a POJO having the same structure as the JsonNode,
                    //as shown below. Then you can operate on the POJO.
                    CustomPOJO2 pojo_doc = OBJECT_MAPPER.treeToValue(document, CustomPOJO2.class);
                    logger.info("id: " + pojo_doc.getId());

                } catch (JsonProcessingException e) {
                    e.printStackTrace();
                }
            }
            isWorkCompleted = true;
            logger.info("End handleChanges()");

        };
    }
    // </Delegate>

    public static CosmosAsyncClient getCosmosClient() {

        return new CosmosClientBuilder()
                .endpoint(AccountSettings.HOST)
                .key(AccountSettings.MASTER_KEY)
                .contentResponseOnWriteEnabled(true)
                .consistencyLevel(ConsistencyLevel.SESSION)
                .buildAsyncClient();
    }

    public static CosmosAsyncDatabase createNewDatabase(CosmosAsyncClient client, String databaseName) {
        CosmosDatabaseResponse databaseResponse = client.createDatabaseIfNotExists(databaseName).block();
        return client.getDatabase(databaseResponse.getProperties().getId());
    }

    public static void deleteDatabase(CosmosAsyncDatabase cosmosDatabase) {
        cosmosDatabase.delete().block();
    }

    public static CosmosAsyncContainer createNewCollection(CosmosAsyncClient client, String databaseName, String collectionName) {
        CosmosAsyncDatabase databaseLink = client.getDatabase(databaseName);
        CosmosAsyncContainer collectionLink = databaseLink.getContainer(collectionName);
        CosmosContainerResponse containerResponse = null;

        try {
            containerResponse = collectionLink.read().block();

            if (containerResponse != null) {
                throw new IllegalArgumentException(String.format("Collection %s already exists in database %s.", collectionName, databaseName));
            }
        } catch (RuntimeException ex) {
            if (ex instanceof CosmosException) {
                CosmosException CosmosException = (CosmosException) ex;

                if (CosmosException.getStatusCode() != 404) {
                    throw ex;
                }
            } else {
                throw ex;
            }
        }

        CosmosContainerProperties containerSettings = new CosmosContainerProperties(collectionName, "/pk");
        CosmosContainerRequestOptions requestOptions = new CosmosContainerRequestOptions();

        ThroughputProperties throughputProperties = ThroughputProperties.createManualThroughput(10000);

        containerResponse = databaseLink.createContainer(containerSettings, throughputProperties, requestOptions).block();

        if (containerResponse == null) {
            throw new RuntimeException(String.format("Failed to create collection %s in database %s.", collectionName, databaseName));
        }

        return databaseLink.getContainer(containerResponse.getProperties().getId());
    }

    public static CosmosAsyncContainer createNewLeaseCollection(CosmosAsyncClient client, String databaseName, String leaseCollectionName) {
        CosmosAsyncDatabase databaseLink = client.getDatabase(databaseName);
        CosmosAsyncContainer leaseCollectionLink = databaseLink.getContainer(leaseCollectionName);
        CosmosContainerResponse leaseContainerResponse = null;

        try {
            leaseContainerResponse = leaseCollectionLink.read().block();

            if (leaseContainerResponse != null) {
                leaseCollectionLink.delete().block();

                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
            }
        } catch (RuntimeException ex) {
            if (ex instanceof CosmosException) {
                CosmosException CosmosException = (CosmosException) ex;

                if (CosmosException.getStatusCode() != 404) {
                    throw ex;
                }
            } else {
                throw ex;
            }
        }

        CosmosContainerProperties containerSettings = new CosmosContainerProperties(leaseCollectionName, "/id");
        CosmosContainerRequestOptions requestOptions = new CosmosContainerRequestOptions();

        ThroughputProperties throughputProperties = ThroughputProperties.createManualThroughput(400);

        leaseContainerResponse = databaseLink.createContainer(containerSettings, throughputProperties, requestOptions).block();

        if (leaseContainerResponse == null) {
            throw new RuntimeException(String.format("Failed to create collection %s in database %s.", leaseCollectionName, databaseName));
        }

        return databaseLink.getContainer(leaseContainerResponse.getProperties().getId());
    }

    public static void createNewDocumentsCustomPOJO(CosmosAsyncContainer containerClient, int count, Duration delay) {
        String suffix = UUID.randomUUID().toString();
        for (int i = 0; i <= count; i++) {
            CustomPOJO2 document = new CustomPOJO2();
            document.setId(String.format("0%d-%s", i, suffix));
            document.setPk(document.getId()); // This is a very simple example, so we'll just have a partition key (/pk) field that we set equal to id

            containerClient.createItem(document).subscribe(doc -> {
                logger.info("Document write: " + doc);
            });

            long remainingWork = delay.toMillis();
            try {
                while (remainingWork > 0) {
                    Thread.sleep(100);
                    remainingWork -= 100;
                }
            } catch (InterruptedException iex) {
                // exception caught
                break;
            }
        }
    }
}

options = new ChangeFeedProcessorOptions();            
options.setStartFromBeginning(false);
options.setLeasePrefix("myChangeFeedDeploymentUnit");

The delegate implementation for reading the change feed in all versions and deletes mode is similar, but instead of calling .handleChanges(), call .handleAllVersionsAndDeletesChanges(). The All versions and deletes mode is in preview and is available in Java SDK version >= 4.42.0.

Here's an example:

// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License.

package com.azure.cosmos.examples.changefeed;

import com.azure.cosmos.ChangeFeedProcessor;
import com.azure.cosmos.ChangeFeedProcessorBuilder;
import com.azure.cosmos.ConsistencyLevel;
import com.azure.cosmos.CosmosAsyncClient;
import com.azure.cosmos.CosmosAsyncContainer;
import com.azure.cosmos.CosmosAsyncDatabase;
import com.azure.cosmos.CosmosClientBuilder;
import com.azure.cosmos.CosmosException;
import com.azure.cosmos.ThroughputControlGroupConfig;
import com.azure.cosmos.ThroughputControlGroupConfigBuilder;
import com.azure.cosmos.examples.common.AccountSettings;
import com.azure.cosmos.examples.common.CustomPOJO2;
import com.azure.cosmos.implementation.Utils;
import com.azure.cosmos.implementation.apachecommons.lang.RandomStringUtils;
import com.azure.cosmos.models.ChangeFeedProcessorItem;
import com.azure.cosmos.models.ChangeFeedProcessorOptions;
import com.azure.cosmos.models.CosmosContainerProperties;
import com.azure.cosmos.models.CosmosContainerRequestOptions;
import com.azure.cosmos.models.CosmosContainerResponse;
import com.azure.cosmos.models.CosmosDatabaseResponse;
import com.azure.cosmos.models.PartitionKey;
import com.azure.cosmos.models.PriorityLevel;
import com.azure.cosmos.models.ThroughputProperties;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

/**
 * Sample for Change Feed Processor.
 * This sample models an application where documents are being inserted into one container (the "feed container"),
 * and meanwhile another worker thread or worker application is pulling inserted documents from the feed container's Change Feed
 * and operating on them in some way. For one or more workers to process the Change Feed of a container, the workers must first contact the server
 * and "lease" access to monitor one or more partitions of the feed container. The Change Feed Processor Library
 * handles leasing automatically for you, however you must create a separate "lease container" where the Change Feed
 * Processor Library can store and track leases container partitions.
 */
public class SampleChangeFeedProcessorForAllVersionsAndDeletesMode {

    public static int WAIT_FOR_WORK = 60000;
    public static final String DATABASE_NAME = "db_" + RandomStringUtils.randomAlphabetic(7);
    public static final String COLLECTION_NAME = "coll_" + RandomStringUtils.randomAlphabetic(7);
    private static final ObjectMapper OBJECT_MAPPER = Utils.getSimpleObjectMapper();
    protected static Logger logger = LoggerFactory.getLogger(SampleChangeFeedProcessorForAllVersionsAndDeletesMode.class);


    private static ChangeFeedProcessor changeFeedProcessorInstance;
    private static boolean isWorkCompleted = false;

    private static ChangeFeedProcessorOptions options;
    private static List<CustomPOJO2> documentList = new ArrayList<>();


    public static void main(String[] args) {
        logger.info("BEGIN Sample");

        try {

            //By integrating throughput control with Change Feed Processor, users can set a low priority for
            //the change feed workload, ensuring that it does not impact other workloads running in parallel.
            //This prevents heavy throttling during times of high request unit (RU) usage.
            //Learn about throughput control here: https://aka.ms/JavaCosmosSparkThroughputControl
            //Learn about priority based execution here: https://aka.ms/CosmosPriorityBasedExecution
            // <ThroughputControlGroupConfig>
            ThroughputControlGroupConfig throughputControlGroupConfig =
                    new ThroughputControlGroupConfigBuilder()
                            .groupName("cfp")
                            .targetThroughput(300)
                            .priorityLevel(PriorityLevel.LOW)
                            .build();
            // </ThroughputControlGroupConfig>
            // <ChangeFeedProcessorOptions>
            options = new ChangeFeedProcessorOptions();            
            options.setStartFromBeginning(false);
            options.setLeasePrefix("myChangeFeedDeploymentUnit");
            options.setFeedPollDelay(Duration.ofSeconds(5));
            options.setFeedPollThroughputControlConfig(throughputControlGroupConfig);
            // </ChangeFeedProcessorOptions>

            //Summary of the next four commands:
            //-Create an asynchronous Azure Cosmos DB client and database so that we can issue async requests to the DB
            //-Create a "feed container" and a "lease container" in the DB
            logger.info("-->CREATE DocumentClient");
            CosmosAsyncClient client = getCosmosClient();

            logger.info("-->CREATE sample's database: " + DATABASE_NAME);
            CosmosAsyncDatabase cosmosDatabase = createNewDatabase(client, DATABASE_NAME);

            logger.info("-->CREATE container for documents: " + COLLECTION_NAME);
            CosmosAsyncContainer feedContainer = createNewCollection(client, DATABASE_NAME, COLLECTION_NAME);

            logger.info("-->CREATE container for lease: " + COLLECTION_NAME + "-leases");
            CosmosAsyncContainer leaseContainer = createNewLeaseCollection(client, DATABASE_NAME, COLLECTION_NAME + "-leases");

            //Model of a worker thread or application which leases access to monitor one or more feed container
            //partitions via the Change Feed. In a real-world application you might deploy this code in an Azure function.
            //The next line causes the worker to create and start an instance of the Change Feed Processor. See the implementation of getChangeFeedProcessorForAllVersionsAndDeletesMode() for guidance
            //on creating a handler for Change Feed events. In this stream, we also trigger the insertion of 10 documents on a separate
            //thread.

            // <StartChangeFeedProcessor>
            logger.info("-->START Change Feed Processor on worker (handles changes asynchronously)");
            changeFeedProcessorInstance = getChangeFeedProcessorForAllVersionsAndDeletesMode("SampleHost_1", feedContainer, leaseContainer);
            changeFeedProcessorInstance.start()
                    .subscribeOn(Schedulers.boundedElastic())
                    .doOnSuccess(aVoid -> {
                        //pass
                        
                    })
                    .subscribe();
            // </StartChangeFeedProcessor>

            //These two lines model an application which is inserting ten documents into the feed container
            logger.info("-->START application that inserts documents into feed container");
            createNewDocumentsCustomPOJO(feedContainer, 10, Duration.ofSeconds(3));
            upsertDocumentsCustomPOJO(feedContainer, 10, Duration.ofSeconds(3));
            deleteDocumentsCustomPOJO(feedContainer, 10, Duration.ofSeconds(3));
            isWorkCompleted = true;

            //This loop models the Worker main loop, which spins while its Change Feed Processor instance asynchronously
            //handles incoming Change Feed events from the feed container. Of course in this sample, polling
            //isWorkCompleted is unnecessary because items are being added to the feed container on the same thread, and you
            //can see just above isWorkCompleted is set to true.
            //But conceptually the worker is part of a different thread or application than the one which is inserting
            //into the feed container; so this code illustrates the worker waiting and listening for changes to the feed container
            long remainingWork = WAIT_FOR_WORK;
            while (!isWorkCompleted && remainingWork > 0) {
                Thread.sleep(100);
                remainingWork -= 100;
            }

            //When all documents have been processed, clean up
            if (isWorkCompleted) {
                if (changeFeedProcessorInstance != null) {
                    changeFeedProcessorInstance.stop().subscribe();
                }
            } else {
                throw new RuntimeException("The change feed processor initialization and automatic create document feeding process did not complete in the expected time");
            }

            logger.info("-->DELETE sample's database: " + DATABASE_NAME);
            //deleteDatabase(cosmosDatabase);

            Thread.sleep(500);

        } catch (Exception e) {
            e.printStackTrace();
        }

        logger.info("END Sample");
    }

    // <Delegate>
    public static ChangeFeedProcessor getChangeFeedProcessorForAllVersionsAndDeletesMode(String hostName, CosmosAsyncContainer feedContainer, CosmosAsyncContainer leaseContainer) {
        return new ChangeFeedProcessorBuilder()
                .hostName(hostName)
                .options(options)
                .feedContainer(feedContainer)
                .leaseContainer(leaseContainer)
                .handleAllVersionsAndDeletesChanges((List<ChangeFeedProcessorItem> changeFeedProcessorItems) -> {
                    logger.info("--->handleAllVersionsAndDeletesChanges() START");

                    for (ChangeFeedProcessorItem item : changeFeedProcessorItems) {
                        try {
                            // AllVersionsAndDeletes Change Feed hands the document to you in the form of ChangeFeedProcessorItem
                            // As a developer you have two options for handling the ChangeFeedProcessorItem provided to you by Change Feed
                            // One option is to operate on the item as it is and call the different getters for different states, as shown below.
                            logger.info("---->DOCUMENT RECEIVED: {}", OBJECT_MAPPER.writerWithDefaultPrettyPrinter()
                                    .writeValueAsString(item));

                            logger.info("---->CURRENT RECEIVED: {}", item.getCurrent());
                            logger.info("---->PREVIOUS RECEIVED: {}", item.getPrevious());
                            logger.info("---->METADATA RECEIVED: {}", item.getChangeFeedMetaData());

                            // You can also transform the ChangeFeedProcessorItem to JsonNode and work on the generic json structure.
                            // This is great especially if you do not have a single uniform data model for all documents.
                            logger.info("----=>JsonNode received: " + item.toJsonNode());

                        } catch (JsonProcessingException e) {
                            e.printStackTrace();
                        }
                    }
                    logger.info("--->handleAllVersionsAndDeletesChanges() END");

                })
                .buildChangeFeedProcessor();
    }
    // </Delegate>

    public static CosmosAsyncClient getCosmosClient() {

        return new CosmosClientBuilder()
                .endpoint(AccountSettings.HOST)
                .key(AccountSettings.MASTER_KEY)
                .contentResponseOnWriteEnabled(true)
                .consistencyLevel(ConsistencyLevel.SESSION)
                .buildAsyncClient();
    }

    public static CosmosAsyncDatabase createNewDatabase(CosmosAsyncClient client, String databaseName) {
        CosmosDatabaseResponse databaseResponse = client.createDatabaseIfNotExists(databaseName).block();
        return client.getDatabase(databaseResponse.getProperties().getId());
    }

    public static void deleteDatabase(CosmosAsyncDatabase cosmosDatabase) {
        cosmosDatabase.delete().block();
    }

    public static CosmosAsyncContainer createNewCollection(CosmosAsyncClient client, String databaseName, String collectionName) {
        CosmosAsyncDatabase databaseLink = client.getDatabase(databaseName);
        CosmosAsyncContainer collectionLink = databaseLink.getContainer(collectionName);
        CosmosContainerResponse containerResponse = null;

        try {
            containerResponse = collectionLink.read().block();

            if (containerResponse != null) {
                throw new IllegalArgumentException(String.format("Collection %s already exists in database %s.", collectionName, databaseName));
            }
        } catch (RuntimeException ex) {
            if (ex instanceof CosmosException) {
                CosmosException CosmosException = (CosmosException) ex;

                if (CosmosException.getStatusCode() != 404) {
                    throw ex;
                }
            } else {
                throw ex;
            }
        }

        CosmosContainerProperties containerSettings = new CosmosContainerProperties(collectionName, "/pk");
        CosmosContainerRequestOptions requestOptions = new CosmosContainerRequestOptions();

        ThroughputProperties throughputProperties = ThroughputProperties.createManualThroughput(10000);

        containerResponse = databaseLink.createContainer(containerSettings, throughputProperties, requestOptions).block();

        if (containerResponse == null) {
            throw new RuntimeException(String.format("Failed to create collection %s in database %s.", collectionName, databaseName));
        }

        return databaseLink.getContainer(containerResponse.getProperties().getId());
    }

    public static CosmosAsyncContainer createNewLeaseCollection(CosmosAsyncClient client, String databaseName, String leaseCollectionName) {
        CosmosAsyncDatabase databaseLink = client.getDatabase(databaseName);
        CosmosAsyncContainer leaseCollectionLink = databaseLink.getContainer(leaseCollectionName);
        CosmosContainerResponse leaseContainerResponse = null;

        try {
            leaseContainerResponse = leaseCollectionLink.read().block();

            if (leaseContainerResponse != null) {
                leaseCollectionLink.delete().block();

                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
            }
        } catch (RuntimeException ex) {
            if (ex instanceof CosmosException) {
                CosmosException CosmosException = (CosmosException) ex;

                if (CosmosException.getStatusCode() != 404) {
                    throw ex;
                }
            } else {
                throw ex;
            }
        }

        CosmosContainerProperties containerSettings = new CosmosContainerProperties(leaseCollectionName, "/id");
        CosmosContainerRequestOptions requestOptions = new CosmosContainerRequestOptions();

        ThroughputProperties throughputProperties = ThroughputProperties.createManualThroughput(400);

        leaseContainerResponse = databaseLink.createContainer(containerSettings, throughputProperties, requestOptions).block();

        if (leaseContainerResponse == null) {
            throw new RuntimeException(String.format("Failed to create collection %s in database %s.", leaseCollectionName, databaseName));
        }

        return databaseLink.getContainer(leaseContainerResponse.getProperties().getId());
    }

    public static void createNewDocumentsCustomPOJO(CosmosAsyncContainer containerClient, int count, Duration delay) {
        String suffix = RandomStringUtils.randomAlphabetic(10);
        for (int i = 0; i <= count; i++) {
            CustomPOJO2 document = new CustomPOJO2();
            document.setId(String.format("0%d-%s", i, suffix));
            document.setPk(document.getId()); // This is a very simple example, so we'll just have a partition key (/pk) field that we set equal to id

            containerClient.createItem(document).subscribe(doc -> {
                logger.info("---->DOCUMENT INSERT: " + doc);
                documentList.add(document);
            });

            long remainingWork = delay.toMillis();
            try {
                while (remainingWork > 0) {
                    Thread.sleep(100);
                    remainingWork -= 100;
                }
            } catch (InterruptedException iex) {
                // exception caught
                break;
            }
        }
    }

    public static void upsertDocumentsCustomPOJO(CosmosAsyncContainer containerClient, int count, Duration delay) {
        for (CustomPOJO2 document : documentList){
            document.setId(document.getId());
            document.setPk(document.getId()); // This is a very simple example, so we'll just have a partition key (/pk) field that we set equal to id

            containerClient.upsertItem(document).subscribe(doc -> {
                logger.info("---->DOCUMENT UPSERT: " + doc);
            });

            long remainingWork = delay.toMillis();
            try {
                while (remainingWork > 0) {
                    Thread.sleep(100);
                    remainingWork -= 100;
                }
            } catch (InterruptedException iex) {
                // exception caught
                break;
            }
        }
    }

    public static void deleteDocumentsCustomPOJO(CosmosAsyncContainer containerClient, int count, Duration delay) {
        for (int i = 0; i <= count; i++) {
            CustomPOJO2 document = documentList.get(i);
            containerClient.deleteItem(document.getId(), new PartitionKey(document.getPk())).subscribe(doc -> {
                logger.info("---->DOCUMENT DELETE: " + doc);
            });

            long remainingWork = delay.toMillis();
            try {
                while (remainingWork > 0) {
                    Thread.sleep(100);
                    remainingWork -= 100;
                }
            } catch (InterruptedException iex) {
                // exception caught
                break;
            }
        }
    }
}

In either change feed mode, you can assign it to changeFeedProcessorInstance and pass the parameters of the compute instance name (hostName), the monitored container (here called feedContainer), and the leaseContainer. Then start the change feed processor:

// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License.
package com.azure.cosmos.examples.changefeed;

import com.azure.cosmos.ChangeFeedProcessor;
import com.azure.cosmos.ChangeFeedProcessorBuilder;
import com.azure.cosmos.ConsistencyLevel;
import com.azure.cosmos.CosmosAsyncClient;
import com.azure.cosmos.CosmosAsyncContainer;
import com.azure.cosmos.CosmosAsyncDatabase;
import com.azure.cosmos.CosmosClientBuilder;
import com.azure.cosmos.CosmosException;
import com.azure.cosmos.ThroughputControlGroupConfig;
import com.azure.cosmos.ThroughputControlGroupConfigBuilder;
import com.azure.cosmos.examples.common.AccountSettings;
import com.azure.cosmos.examples.common.CustomPOJO2;
import com.azure.cosmos.implementation.Utils;
import com.azure.cosmos.models.ChangeFeedProcessorOptions;
import com.azure.cosmos.models.CosmosContainerProperties;
import com.azure.cosmos.models.CosmosContainerRequestOptions;
import com.azure.cosmos.models.CosmosContainerResponse;
import com.azure.cosmos.models.CosmosDatabaseResponse;
import com.azure.cosmos.models.PriorityLevel;
import com.azure.cosmos.models.ThroughputProperties;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.List;
import java.util.UUID;
import java.util.function.Consumer;

/**
 * Sample for Change Feed Processor.
 * This sample models an application where documents are being inserted into one container (the "feed container"),
 * and meanwhile another worker thread or worker application is pulling inserted documents from the feed container's Change Feed
 * and operating on them in some way. For one or more workers to process the Change Feed of a container, the workers must first contact the server
 * and "lease" access to monitor one or more partitions of the feed container. The Change Feed Processor Library
 * handles leasing automatically for you, however you must create a separate "lease container" where the Change Feed
 * Processor Library can store and track leases container partitions.
 */
public class SampleChangeFeedProcessor {

    public static int WAIT_FOR_WORK = 60000;
    public static final String DATABASE_NAME = "db_" + UUID.randomUUID();
    public static final String COLLECTION_NAME = "coll_" + UUID.randomUUID();
    private static final ObjectMapper OBJECT_MAPPER = Utils.getSimpleObjectMapper();
    protected static Logger logger = LoggerFactory.getLogger(SampleChangeFeedProcessor.class);

    private static boolean isWorkCompleted = false;

    private static ChangeFeedProcessorOptions options;



    public static void main(String[] args) {
        logger.info("Begin Sample");
        try {

            //By integrating throughput control with Change Feed Processor, users can set a low priority for
            //the change feed workload, ensuring that it does not impact other workloads running in parallel.
            //This prevents heavy throttling during times of high request unit (RU) usage.
            //Learn about throughput control here: https://aka.ms/JavaCosmosSparkThroughputControl
            //Learn about priority based execution here: https://aka.ms/CosmosPriorityBasedExecution
            // <ThroughputControlGroupConfig>
            ThroughputControlGroupConfig throughputControlGroupConfig =
                    new ThroughputControlGroupConfigBuilder()
                            .groupName("cfp")
                            .targetThroughput(300)
                            .priorityLevel(PriorityLevel.LOW)
                            .build();
            // </ThroughputControlGroupConfig>
            // <ChangeFeedProcessorOptions>
            options = new ChangeFeedProcessorOptions();
            options.setStartFromBeginning(false);
            options.setLeasePrefix("myChangeFeedDeploymentUnit");
            options.setFeedPollDelay(Duration.ofSeconds(5));
            options.setFeedPollThroughputControlConfig(throughputControlGroupConfig);
            // </ChangeFeedProcessorOptions>

            //Summary of the next four commands:
            //-Create an asynchronous Azure Cosmos DB client and database so that we can issue async requests to the DB
            //-Create a "feed container" and a "lease container" in the DB
            logger.info("Create CosmosClient");
            CosmosAsyncClient client = getCosmosClient();

            logger.info("Create sample's database: " + DATABASE_NAME);
            CosmosAsyncDatabase cosmosDatabase = createNewDatabase(client, DATABASE_NAME);

            logger.info("Create container for documents: " + COLLECTION_NAME);
            CosmosAsyncContainer feedContainer = createNewCollection(client, DATABASE_NAME, COLLECTION_NAME);

            logger.info("Create container for lease: " + COLLECTION_NAME + "-leases");
            CosmosAsyncContainer leaseContainer = createNewLeaseCollection(client, DATABASE_NAME, COLLECTION_NAME + "-leases");

            //Model of a worker thread or application which leases access to monitor one or more feed container
            //partitions via the Change Feed. In a real-world application you might deploy this code in an Azure function.
            //The next line causes the worker to create and start an instance of the Change Feed Processor. See the implementation of getChangeFeedProcessor() for guidance
            //on creating a handler for Change Feed events. In this stream, we also trigger the insertion of 10 documents on a separate
            //thread.
            // <StartChangeFeedProcessor>
            logger.info("Start Change Feed Processor on worker (handles changes asynchronously)");
            ChangeFeedProcessor changeFeedProcessorInstance = new ChangeFeedProcessorBuilder()
                .hostName("SampleHost_1")
                .feedContainer(feedContainer)
                .leaseContainer(leaseContainer)
                .handleChanges(handleChanges())
                .options(options)
                .buildChangeFeedProcessor();
            changeFeedProcessorInstance.start()
                                       .subscribeOn(Schedulers.boundedElastic())
                                       .subscribe();
            // </StartChangeFeedProcessor>

            //These two lines model an application which is inserting ten documents into the feed container
            logger.info("Start application that inserts documents into feed container");
            createNewDocumentsCustomPOJO(feedContainer, 10, Duration.ofSeconds(3));

            //This loop models the Worker main loop, which spins while its Change Feed Processor instance asynchronously
            //handles incoming Change Feed events from the feed container. Of course in this sample, polling
            //isWorkCompleted is unnecessary because items are being added to the feed container on the same thread, and you
            //can see just above isWorkCompleted is set to true.
            //But conceptually the worker is part of a different thread or application than the one which is inserting
            //into the feed container; so this code illustrates the worker waiting and listening for changes to the feed container
            long remainingWork = WAIT_FOR_WORK;
            while (!isWorkCompleted && remainingWork > 0) {
                Thread.sleep(100);
                remainingWork -= 100;
            }

            //When all documents have been processed, clean up
            if (isWorkCompleted) {
                changeFeedProcessorInstance.stop().subscribe();
            } else {
                throw new RuntimeException("The change feed processor initialization and automatic create document feeding process did not complete in the expected time");
            }

            logger.info("Delete sample's database: " + DATABASE_NAME);
            //deleteDatabase(cosmosDatabase);

            Thread.sleep(500);
        } catch (Exception e) {
            e.printStackTrace();
        }
        logger.info("End Sample");
    }

    // <Delegate>
    private static Consumer<List<JsonNode>> handleChanges() {
        return (List<JsonNode> docs) -> {
            logger.info("Start handleChanges()");

            for (JsonNode document : docs) {
                try {
                    //Change Feed hands the document to you in the form of a JsonNode
                    //As a developer you have two options for handling the JsonNode document provided to you by Change Feed
                    //One option is to operate on the document in the form of a JsonNode, as shown below. This is great
                    //especially if you do not have a single uniform data model for all documents.
                    logger.info("Document received: " + OBJECT_MAPPER.writerWithDefaultPrettyPrinter()
                            .writeValueAsString(document));

                    //You can also transform the JsonNode to a POJO having the same structure as the JsonNode,
                    //as shown below. Then you can operate on the POJO.
                    CustomPOJO2 pojo_doc = OBJECT_MAPPER.treeToValue(document, CustomPOJO2.class);
                    logger.info("id: " + pojo_doc.getId());

                } catch (JsonProcessingException e) {
                    e.printStackTrace();
                }
            }
            isWorkCompleted = true;
            logger.info("End handleChanges()");

        };
    }
    // </Delegate>

    public static CosmosAsyncClient getCosmosClient() {

        return new CosmosClientBuilder()
                .endpoint(AccountSettings.HOST)
                .key(AccountSettings.MASTER_KEY)
                .contentResponseOnWriteEnabled(true)
                .consistencyLevel(ConsistencyLevel.SESSION)
                .buildAsyncClient();
    }

    public static CosmosAsyncDatabase createNewDatabase(CosmosAsyncClient client, String databaseName) {
        CosmosDatabaseResponse databaseResponse = client.createDatabaseIfNotExists(databaseName).block();
        return client.getDatabase(databaseResponse.getProperties().getId());
    }

    public static void deleteDatabase(CosmosAsyncDatabase cosmosDatabase) {
        cosmosDatabase.delete().block();
    }

    public static CosmosAsyncContainer createNewCollection(CosmosAsyncClient client, String databaseName, String collectionName) {
        CosmosAsyncDatabase databaseLink = client.getDatabase(databaseName);
        CosmosAsyncContainer collectionLink = databaseLink.getContainer(collectionName);
        CosmosContainerResponse containerResponse = null;

        try {
            containerResponse = collectionLink.read().block();

            if (containerResponse != null) {
                throw new IllegalArgumentException(String.format("Collection %s already exists in database %s.", collectionName, databaseName));
            }
        } catch (RuntimeException ex) {
            if (ex instanceof CosmosException) {
                CosmosException CosmosException = (CosmosException) ex;

                if (CosmosException.getStatusCode() != 404) {
                    throw ex;
                }
            } else {
                throw ex;
            }
        }

        CosmosContainerProperties containerSettings = new CosmosContainerProperties(collectionName, "/pk");
        CosmosContainerRequestOptions requestOptions = new CosmosContainerRequestOptions();

        ThroughputProperties throughputProperties = ThroughputProperties.createManualThroughput(10000);

        containerResponse = databaseLink.createContainer(containerSettings, throughputProperties, requestOptions).block();

        if (containerResponse == null) {
            throw new RuntimeException(String.format("Failed to create collection %s in database %s.", collectionName, databaseName));
        }

        return databaseLink.getContainer(containerResponse.getProperties().getId());
    }

    public static CosmosAsyncContainer createNewLeaseCollection(CosmosAsyncClient client, String databaseName, String leaseCollectionName) {
        CosmosAsyncDatabase databaseLink = client.getDatabase(databaseName);
        CosmosAsyncContainer leaseCollectionLink = databaseLink.getContainer(leaseCollectionName);
        CosmosContainerResponse leaseContainerResponse = null;

        try {
            leaseContainerResponse = leaseCollectionLink.read().block();

            if (leaseContainerResponse != null) {
                leaseCollectionLink.delete().block();

                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
            }
        } catch (RuntimeException ex) {
            if (ex instanceof CosmosException) {
                CosmosException CosmosException = (CosmosException) ex;

                if (CosmosException.getStatusCode() != 404) {
                    throw ex;
                }
            } else {
                throw ex;
            }
        }

        CosmosContainerProperties containerSettings = new CosmosContainerProperties(leaseCollectionName, "/id");
        CosmosContainerRequestOptions requestOptions = new CosmosContainerRequestOptions();

        ThroughputProperties throughputProperties = ThroughputProperties.createManualThroughput(400);

        leaseContainerResponse = databaseLink.createContainer(containerSettings, throughputProperties, requestOptions).block();

        if (leaseContainerResponse == null) {
            throw new RuntimeException(String.format("Failed to create collection %s in database %s.", leaseCollectionName, databaseName));
        }

        return databaseLink.getContainer(leaseContainerResponse.getProperties().getId());
    }

    public static void createNewDocumentsCustomPOJO(CosmosAsyncContainer containerClient, int count, Duration delay) {
        String suffix = UUID.randomUUID().toString();
        for (int i = 0; i <= count; i++) {
            CustomPOJO2 document = new CustomPOJO2();
            document.setId(String.format("0%d-%s", i, suffix));
            document.setPk(document.getId()); // This is a very simple example, so we'll just have a partition key (/pk) field that we set equal to id

            containerClient.createItem(document).subscribe(doc -> {
                logger.info("Document write: " + doc);
            });

            long remainingWork = delay.toMillis();
            try {
                while (remainingWork > 0) {
                    Thread.sleep(100);
                    remainingWork -= 100;
                }
            } catch (InterruptedException iex) {
                // exception caught
                break;
            }
        }
    }
}

Processing life cycle

The normal life cycle of a host instance is:

1. Read the change feed.
2. If there are no changes, sleep for a predefined amount of time (customizable with options.setFeedPollDelay in the builder) and go to #1.
3. If there are changes, send them to the delegate.
4. When the delegate finishes processing the changes successfully, update the lease store by using the latest processed point in time and go to #1.

Error handling

The change feed processor is resilient to user code errors. If your delegate implementation has an unhandled exception (step #4), the thread that's processing that particular batch of changes is stopped, and a new thread is created. The new thread checks the latest point in time that the lease store has saved for that range of partition key values, and it restarts from there, effectively sending the same batch of changes to the delegate. This behavior continues until your delegate processes the changes correctly. It's the reason why the change feed processor has an "at least once" guarantee.

To prevent your change feed processor from getting "stuck" continuously retrying the same batch of changes, you should add logic in your delegate code to write documents, upon exception, to an errored-message. This design ensures that you can keep track of unprocessed changes while still being able to continue to process future changes. The errored-message might be another Azure Cosmos DB container. The exact data store doesn't matter. You simply want the unprocessed changes to be persisted.

You also can use the change feed estimator to monitor the progress of your change feed processor instances as they read the change feed.

Deployment unit

A single change feed processor deployment unit consists of one or more compute instances that have the same lease container configuration and the same leasePrefix, but different hostName values. You can have many deployment units in which each one has a different business flow for the changes, and each deployment unit consists of one or more instances.

For example, you might have one deployment unit that triggers an external API each time there's a change in your container. Another deployment unit might move data in real time each time there's a change. When a change happens in your monitored container, all your deployment units are notified.

Dynamic scaling

As mentioned earlier, within a deployment unit, you can have one or more compute instances. To take advantage of the compute distribution within the deployment unit, the only key requirements are that:

• All instances should have the same lease container configuration.
• All instances should have the same value set in options.setLeasePrefix (or none set at all).
• Each instance needs to have a different hostName.

If these three conditions apply, then the change feed processor distributes all the leases in the lease container across all running instances of that deployment unit, and it parallelizes compute by using an equal-distribution algorithm. A lease is owned by one instance at any time, so the number of instances shouldn't be greater than the number of leases.

The number of instances can grow and shrink. The change feed processor dynamically adjusts the load by redistributing it accordingly. Deployment units can share the same lease container, but they should each have a different leasePrefix value.

Moreover, the change feed processor can dynamically adjust a container's scale if the container's throughput or storage increases. When your container grows, the change feed processor transparently handles the scenario by dynamically increasing the leases and distributing the new leases among existing instances.

Starting time

By default, when a change feed processor starts for the first time, it initializes the lease container and starts its processing life cycle. Any changes that happened in the monitored container before the change feed processor was initialized for the first time aren't detected.

Reading from a previous date and time

It's possible to initialize the change feed processor to read changes starting at a specific date and time by setting setStartTime in options. The change feed processor is initialized for that specific date and time, and it starts reading the changes that happened afterward.

Reading from the beginning

In the sample, setStartFromBeginning is set to false, which is the same as the default value. In other scenarios, like in data migrations or if you're analyzing the entire history of a container, you need to read the change feed from the beginning of that container's lifetime. To do that, you can set setStartFromBeginning to true. The change feed processor is initialized, and it starts reading changes from the beginning of the lifetime of the container.