ChangeFeedProcessor: Fixes when ChangeFeedMode is switched should throw an exception.

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/ChangeFeedEstimatorIterator.cs

@@ -334,7 +334,8 @@ private async Task InitializeLeaseStoreAsync(ITrace trace, CancellationToken can
                         monitoredContainer: this.monitoredContainer,
                         leaseContainer: this.leaseContainer,
                         leaseContainerPrefix: leasePrefix,
-                        instanceName: ChangeFeedEstimatorIterator.EstimatorDefaultHostName);
+                        instanceName: ChangeFeedEstimatorIterator.EstimatorDefaultHostName,
+                        changeFeedMode: ChangeFeedMode.LatestVersion);
 
                     this.documentServiceLeaseContainer = documentServiceLeaseStoreManager.LeaseContainer;
                 }

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/ChangeFeedEstimatorRunner.cs

@@ -139,11 +139,14 @@ private async Task InitializeLeaseStoreAsync()
             {
                 string monitoredContainerAndDatabaseRid = await this.monitoredContainer.GetMonitoredDatabaseAndContainerRidAsync(default);
                 string leasePrefix = this.monitoredContainer.GetLeasePrefix(this.changeFeedLeaseOptions.LeasePrefix, monitoredContainerAndDatabaseRid);
-                DocumentServiceLeaseStoreManager documentServiceLeaseStoreManager = await DocumentServiceLeaseStoreManagerBuilder.InitializeAsync(
-                    monitoredContainer: this.monitoredContainer,
-                    leaseContainer: this.leaseContainer,
-                    leaseContainerPrefix: leasePrefix,
-                    instanceName: ChangeFeedEstimatorRunner.EstimatorDefaultHostName);
+                DocumentServiceLeaseStoreManager documentServiceLeaseStoreManager = await DocumentServiceLeaseStoreManagerBuilder
+                    .InitializeAsync(
+                        monitoredContainer: this.monitoredContainer,
+                        leaseContainer: this.leaseContainer,
+                        leaseContainerPrefix: leasePrefix,
+                        instanceName: ChangeFeedEstimatorRunner.EstimatorDefaultHostName,
+                        changeFeedMode: ChangeFeedMode.LatestVersion)
+                    .ConfigureAwait(false);
 
                 this.documentServiceLeaseContainer = documentServiceLeaseStoreManager.LeaseContainer;
             }

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/ChangeFeedProcessorBuilder.cs

@@ -70,13 +70,15 @@ public ChangeFeedProcessorBuilder WithInstanceName(string instanceName)
         }
 
         /// <summary>
-        /// Sets the mode for the change freed processor.
+        /// Sets the mode for the change feed processor.
         /// </summary>
         /// <param name="changeFeedMode"></param>
         /// <returns>The instance of <see cref="ChangeFeedProcessorBuilder"/> to use.</returns>
         internal ChangeFeedProcessorBuilder WithChangeFeedMode(ChangeFeedMode changeFeedMode)
         {
             this.changeFeedProcessorOptions.Mode = changeFeedMode;
+            this.changeFeedLeaseOptions.Mode = changeFeedMode;
+
             return this;
         }
 

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/ChangeFeedProcessorCore.cs

@@ -5,15 +5,20 @@
 namespace Microsoft.Azure.Cosmos.ChangeFeed
 {
     using System;
+    using System.Collections.Generic;
+    using System.Linq;
     using System.Threading.Tasks;
+    using global::Azure;
     using Microsoft.Azure.Cosmos.ChangeFeed.Bootstrapping;
     using Microsoft.Azure.Cosmos.ChangeFeed.Configuration;
     using Microsoft.Azure.Cosmos.ChangeFeed.FeedManagement;
     using Microsoft.Azure.Cosmos.ChangeFeed.FeedProcessing;
     using Microsoft.Azure.Cosmos.ChangeFeed.LeaseManagement;
     using Microsoft.Azure.Cosmos.ChangeFeed.Utils;
     using Microsoft.Azure.Cosmos.Core.Trace;
+    using Microsoft.Azure.Cosmos.Resource.CosmosExceptions;
     using Microsoft.Azure.Cosmos.Tracing;
+    using Microsoft.Azure.Documents;
 
     internal sealed class ChangeFeedProcessorCore : ChangeFeedProcessor
     {
@@ -75,23 +80,99 @@ public override async Task StopAsync()
         private async Task InitializeAsync()
         {
             string containerRid = await this.monitoredContainer.GetCachedRIDAsync(
-                forceRefresh: false, 
-                NoOpTrace.Singleton, 
+                forceRefresh: false,
+                NoOpTrace.Singleton,
                 default);
+
             string monitoredDatabaseAndContainerRid = await this.monitoredContainer.GetMonitoredDatabaseAndContainerRidAsync();
             string leaseContainerPrefix = this.monitoredContainer.GetLeasePrefix(this.changeFeedLeaseOptions.LeasePrefix, monitoredDatabaseAndContainerRid);
             Routing.PartitionKeyRangeCache partitionKeyRangeCache = await this.monitoredContainer.ClientContext.DocumentClient.GetPartitionKeyRangeCacheAsync(NoOpTrace.Singleton);
             if (this.documentServiceLeaseStoreManager == null)
             {
-                this.documentServiceLeaseStoreManager = await DocumentServiceLeaseStoreManagerBuilder.InitializeAsync(this.monitoredContainer, this.leaseContainer, leaseContainerPrefix, this.instanceName).ConfigureAwait(false);
+                this.documentServiceLeaseStoreManager = await DocumentServiceLeaseStoreManagerBuilder
+                    .InitializeAsync(
+                        this.monitoredContainer,
+                        this.leaseContainer,
+                        leaseContainerPrefix,
+                        this.instanceName,
+                        changeFeedMode: this.changeFeedProcessorOptions.Mode)
+                    .ConfigureAwait(false);
             }
 
+            await this.ChangeFeedModeSwitchingCheckAsync().ConfigureAwait(false);
+
             this.partitionManager = this.BuildPartitionManager(
-                containerRid, 
+                containerRid,
                 partitionKeyRangeCache);
             this.initialized = true;
         }
 
+        /// <summary>
+        /// If the lease container's lease document is found, this method checks for lease 
+        /// document's ChangeFeedMode and if the new ChangeFeedMode is different
+        /// from the current ChangeFeedMode, a CosmosException is thrown.
+        /// This is based on an issue located at <see href="https://github.com/Azure/azure-cosmos-dotnet-v3/issues/4308"/>.
+        /// </summary>
+        private async Task ChangeFeedModeSwitchingCheckAsync()
+        {
+            IReadOnlyList<DocumentServiceLease> documentServiceLeases = await this.documentServiceLeaseStoreManager
+                .LeaseContainer
+                .GetAllLeasesAsync()
+                .ConfigureAwait(false);
+
+            // No lease documents. Return.
+
+            if (documentServiceLeases.Count == 0)
+            {
+                return;
+            }
+
+            DocumentServiceLease documentServiceLease = documentServiceLeases.FirstOrDefault();
+
+            // No lease documents that match the Id.
+
+            if (documentServiceLease == default)
+            {
+                return;
+            }
+
+            // Mode attribute exists on lease document, but it is not set. legacy is always LatestVersion because
+            // AllVersionsAndDeletes does not exist. There should not be any legacy lease documents that are
+            // AllVersionsAndDeletes. If the ChangeFeedProcessor's mode is not legacy, a CosmosException should thrown.
+            // If the ChangeFeedProcessor mode is not the mode in the lease document, a CosmosException should be thrown.
+
+            bool shouldThrowException = string.IsNullOrEmpty(documentServiceLease.Mode)
+                ? this.VerifyChangeFeedProcessorMode(
+                    changeFeedMode: ChangeFeedMode.LatestVersion,
+                    leaseChangeFeedMode: documentServiceLease.Mode,
+                    normalizedProcessorChangeFeedMode: out string normalizedProcessorChangeFeedMode)
+                : this.VerifyChangeFeedProcessorMode(
+                    changeFeedMode: this.changeFeedLeaseOptions.Mode,
+                    leaseChangeFeedMode: documentServiceLease.Mode,
+                    normalizedProcessorChangeFeedMode: out normalizedProcessorChangeFeedMode);
+
+            // If shouldThrowException is true, throw the CosmosException.
+
+            if (shouldThrowException)
+            {
+                throw CosmosExceptionFactory.CreateBadRequestException(
+                    message: $"Switching {nameof(ChangeFeedMode)} {documentServiceLease.Mode} to {normalizedProcessorChangeFeedMode} is not allowed.",
+                    headers: default);
+            }
+        }
+
+        private bool VerifyChangeFeedProcessorMode(
+            ChangeFeedMode changeFeedMode, 
+            string leaseChangeFeedMode,
+            out string normalizedProcessorChangeFeedMode)
+        {
+            normalizedProcessorChangeFeedMode = changeFeedMode == ChangeFeedMode.AllVersionsAndDeletes
+                ? HttpConstants.A_IMHeaderValues.FullFidelityFeed
+                : HttpConstants.A_IMHeaderValues.IncrementalFeed;
+
+            return string.Compare(leaseChangeFeedMode, normalizedProcessorChangeFeedMode, StringComparison.OrdinalIgnoreCase) != 0;
+        }
+
         private PartitionManager BuildPartitionManager(
             string containerRid,
             Routing.PartitionKeyRangeCache partitionKeyRangeCache)

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/Configuration/ChangeFeedLeaseOptions.cs

@@ -44,5 +44,10 @@ public ChangeFeedLeaseOptions()
         /// instances pointing at the same feed while using the same auxiliary collection.
         /// </summary>
         public string LeasePrefix { get; set; }
+
+        /// <summary>
+        /// Gets or sets the <see cref="ChangeFeedMode"/>.
+        /// </summary>
+        public ChangeFeedMode Mode { get; set; }
     }
 }

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLease.cs

@@ -70,5 +70,10 @@ internal abstract class DocumentServiceLease
         /// Gets or sets custom lease properties which can be managed from <see cref="LoadBalancingStrategy"/>.
         /// </summary>
         public abstract Dictionary<string, string> Properties { get; set; }
+
+        /// <summary>
+        /// Gets or sets the ChangeFeedMode.
+        /// </summary>
+        public abstract string Mode { get; set; }
     }
 }

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLeaseCore.cs

@@ -95,6 +95,9 @@ public override DateTime Timestamp
         [JsonProperty("_ts")]
         private long TS { get; set; }
 
+        [JsonProperty("Mode", NullValueHandling = NullValueHandling.Ignore)]
+        public override string Mode { get; set; }
+
         public override string ToString()
         {
             return string.Format(

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLeaseCoreEpk.cs

@@ -66,6 +66,9 @@ public override DateTime Timestamp
         [JsonProperty("properties")]
         public override Dictionary<string, string> Properties { get; set; } = new Dictionary<string, string>();
 
+        [JsonProperty("Mode", NullValueHandling = NullValueHandling.Ignore)]
+        public override string Mode { get; set; }
+
         [JsonProperty("timestamp")]
         private DateTime? ExplicitTimestamp { get; set; }
 

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLeaseManagerCosmos.cs

@@ -124,7 +124,8 @@ public override Task<DocumentServiceLease> CreateLeaseIfNotExistAsync(
                 LeaseId = leaseDocId,
                 LeaseToken = leaseToken,
                 ContinuationToken = continuationToken,
-                FeedRange = new FeedRangeEpk(partitionKeyRange.ToRange())
+                FeedRange = new FeedRangeEpk(partitionKeyRange.ToRange()),
+                Mode = this.GetChangeFeedMode()
             };
 
             this.requestOptionsFactory.AddPartitionKeyIfNeeded((string pk) => documentServiceLease.LeasePartitionKey = pk, Guid.NewGuid().ToString());
@@ -148,14 +149,22 @@ public override Task<DocumentServiceLease> CreateLeaseIfNotExistAsync(
                 LeaseId = leaseDocId,
                 LeaseToken = leaseToken,
                 ContinuationToken = continuationToken,
-                FeedRange = feedRange
+                FeedRange = feedRange,
+                Mode = this.GetChangeFeedMode()
             };
 
             this.requestOptionsFactory.AddPartitionKeyIfNeeded((string pk) => documentServiceLease.LeasePartitionKey = pk, Guid.NewGuid().ToString());
 
             return this.TryCreateDocumentServiceLeaseAsync(documentServiceLease);
         }
 
+        private string GetChangeFeedMode()
+        {
+            return this.options.Mode == ChangeFeedMode.AllVersionsAndDeletes
+                ? HttpConstants.A_IMHeaderValues.FullFidelityFeed
+                : HttpConstants.A_IMHeaderValues.IncrementalFeed;
+        }
+
         public override async Task ReleaseAsync(DocumentServiceLease lease)
         {
             if (lease == null)

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLeaseStoreManagerBuilder.cs

@@ -22,7 +22,8 @@ public static async Task<DocumentServiceLeaseStoreManager> InitializeAsync(
             ContainerInternal monitoredContainer,
             ContainerInternal leaseContainer,
             string leaseContainerPrefix,
-            string instanceName)
+            string instanceName,
+            ChangeFeedMode changeFeedMode)
         {
             ContainerProperties containerProperties = await leaseContainer.GetCachedContainerPropertiesAsync(forceRefresh: false, NoOpTrace.Singleton, cancellationToken: default);
 
@@ -58,7 +59,8 @@ public static async Task<DocumentServiceLeaseStoreManager> InitializeAsync(
                 .WithMonitoredContainer(monitoredContainer)
                 .WithLeaseContainer(leaseContainer)
                 .WithRequestOptionsFactory(requestOptionsFactory)
-                .WithHostName(instanceName);
+                .WithHostName(instanceName)
+                .WithChangeFeedMode(changeFeedMode);
 
             return leaseStoreManagerBuilder.Build();
         }
@@ -70,7 +72,7 @@ public static async Task<DocumentServiceLeaseStoreManager> InitializeAsync(
 
         private DocumentServiceLeaseStoreManagerBuilder WithMonitoredContainer(ContainerInternal monitoredContainer)
         {
-            this.monitoredContainer = monitoredContainer ?? throw new ArgumentNullException(nameof(leaseContainer));
+            this.monitoredContainer = monitoredContainer ?? throw new ArgumentNullException(nameof(monitoredContainer));
             return this;
         }
 
@@ -98,6 +100,12 @@ private DocumentServiceLeaseStoreManagerBuilder WithHostName(string hostName)
             return this;
         }
 
+        private DocumentServiceLeaseStoreManagerBuilder WithChangeFeedMode(ChangeFeedMode changeFeedMode)
+        {
+            this.options.Mode = changeFeedMode ?? throw new ArgumentNullException(nameof(changeFeedMode));
+            return this;
+        }
+
         private DocumentServiceLeaseStoreManager Build()
         {
             if (this.monitoredContainer == null)

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLeaseStoreManagerOptions.cs

@@ -16,5 +16,7 @@ internal string GetPartitionLeasePrefix()
         {
             return this.ContainerNamePrefix + PartitionLeasePrefixSeparator;
         }
+
+        internal ChangeFeedMode Mode { get; set; }
     }
 }

Microsoft.Azure.Cosmos/src/Resource/Container/Container.cs

@@ -1681,7 +1681,6 @@ public abstract Task<ResponseMessage> DeleteAllItemsByPartitionKeyStreamAsync(
         public abstract Task<IEnumerable<string>> GetPartitionKeyRangesAsync(
             FeedRange feedRange,
             CancellationToken cancellationToken = default);
-
 #endif
     }
 }