ChangeFeedProcessor: Fixes when ChangeFeedMode is switched should throw an exception. (#4334)

philipthomas-MSFT · web-flow · commit dbdf8064f538 · 2024-03-22T14:39:33.000Z
* test to prove no exception, no documents when change feed more is changed.

* renaming

* throwing exception when changeFeedMode change on CFP with same lease container

* removed PREVIEW tags under advisement. Well create a PR later

* removed usings

* revert to old summary

* defaulting change feed mode to latestversion

* change feed mode exception on switching and tests based on last recommendation.

* more changes based on discussions

* some refactoring

* fixed some tests

* more changes based on recommendations

* another refactor

* change from CosmosException to ArgumentException

* removed CosmosException from comments

* fixin tests to account for extra GetAllLeasesAsync call

* some recommendation changes
diff --git a/Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/ChangeFeedEstimatorIterator.cs b/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;
                 }
diff --git a/Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/ChangeFeedEstimatorRunner.cs b/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;
             }
diff --git a/Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/ChangeFeedProcessorBuilder.cs b/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;
         }
 
diff --git a/Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/ChangeFeedProcessorCore.cs b/Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/ChangeFeedProcessorCore.cs
@@ -5,6 +5,7 @@
 namespace Microsoft.Azure.Cosmos.ChangeFeed
 {
     using System;
+    using System.Collections.Generic;
     using System.Threading.Tasks;
     using Microsoft.Azure.Cosmos.ChangeFeed.Bootstrapping;
     using Microsoft.Azure.Cosmos.ChangeFeed.Configuration;
@@ -14,6 +15,7 @@ namespace Microsoft.Azure.Cosmos.ChangeFeed
     using Microsoft.Azure.Cosmos.ChangeFeed.Utils;
     using Microsoft.Azure.Cosmos.Core.Trace;
     using Microsoft.Azure.Cosmos.Tracing;
+    using Microsoft.Azure.Documents;
 
     internal sealed class ChangeFeedProcessorCore : ChangeFeedProcessor
     {
@@ -75,19 +77,36 @@ 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);
             }
 
+            this.documentServiceLeaseStoreManager
+                .LeaseManager
+                .ChangeFeedModeSwitchingCheck(
+                    documentServiceLeases: await this.documentServiceLeaseStoreManager
+                        .LeaseContainer
+                            .GetAllLeasesAsync()
+                            .ConfigureAwait(false),
+                    changeFeedLeaseOptionsMode: this.changeFeedLeaseOptions.Mode);
+
             this.partitionManager = this.BuildPartitionManager(
-                containerRid, 
+                containerRid,
                 partitionKeyRangeCache);
             this.initialized = true;
         }
diff --git a/Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/Configuration/ChangeFeedLeaseOptions.cs b/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; }
     }
 }
diff --git a/Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLease.cs b/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; }
     }
 }
diff --git a/Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLeaseCore.cs b/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(
diff --git a/Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLeaseCoreEpk.cs b/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; }
 
diff --git a/Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLeaseManager.cs b/Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLeaseManager.cs
@@ -4,6 +4,8 @@
 
 namespace Microsoft.Azure.Cosmos.ChangeFeed.LeaseManagement
 {
+    using System;
+    using System.Collections.Generic;
     using System.Threading.Tasks;
     using Microsoft.Azure.Cosmos.ChangeFeed.Exceptions;
     using Microsoft.Azure.Documents;
@@ -63,5 +65,57 @@ internal abstract class DocumentServiceLeaseManager
         /// <returns>Updated lease.</returns>
         /// <exception cref="LeaseLostException">Thrown if other host acquired the lease</exception>
         public abstract Task<DocumentServiceLease> UpdatePropertiesAsync(DocumentServiceLease leaseToUpdatePropertiesFrom);
+
+        /// <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, an exception is thrown.
+        /// This is based on an issue located at <see href="https://github.com/Azure/azure-cosmos-dotnet-v3/issues/4308"/>.
+        /// </summary>
+        public void ChangeFeedModeSwitchingCheck(
+            IReadOnlyList<DocumentServiceLease> documentServiceLeases, 
+            ChangeFeedMode changeFeedLeaseOptionsMode)
+        {
+            // No lease documents. Return.
+
+            if (documentServiceLeases.Count == 0)
+            {
+                return;
+            }
+
+            DocumentServiceLease documentServiceLease = documentServiceLeases[0];
+
+            // 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, an exception should thrown.
+            // If the ChangeFeedProcessor mode is not the mode in the lease document, an exception should be thrown.
+
+            bool shouldThrowException = this.VerifyChangeFeedProcessorMode(
+                changeFeedMode:
+                    string.IsNullOrEmpty(documentServiceLease.Mode)
+                        ? ChangeFeedMode.LatestVersion
+                        : changeFeedLeaseOptionsMode,
+                leaseChangeFeedMode: documentServiceLease.Mode,
+                normalizedProcessorChangeFeedMode: out string normalizedProcessorChangeFeedMode);
+
+            // If shouldThrowException is true, throw the exception.
+
+            if (shouldThrowException)
+            {
+                throw new ArgumentException(message: $"Switching {nameof(ChangeFeedMode)} {documentServiceLease.Mode} to {normalizedProcessorChangeFeedMode} is not allowed.");
+            }
+        }
+
+        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;
+        }
     }
 }
diff --git a/Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLeaseManagerCosmos.cs b/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)
diff --git a/Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLeaseStoreManagerBuilder.cs b/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)
diff --git a/Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLeaseStoreManagerOptions.cs b/Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/LeaseManagement/DocumentServiceLeaseStoreManagerOptions.cs
@@ -16,5 +16,7 @@ internal string GetPartitionLeasePrefix()
         {
             return this.ContainerNamePrefix + PartitionLeasePrefixSeparator;
         }
+
+        internal ChangeFeedMode Mode { get; set; }
     }
 }
diff --git a/Microsoft.Azure.Cosmos/src/Resource/Container/Container.cs b/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
     }
 }
diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.EmulatorTests/ChangeFeed/GetChangeFeedProcessorBuilderWithAllVersionsAndDeletesTests.cs b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.EmulatorTests/ChangeFeed/GetChangeFeedProcessorBuilderWithAllVersionsAndDeletesTests.cs
diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/ChangeFeed/ChangeFeedProcessorCoreTests.cs b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/ChangeFeed/ChangeFeedProcessorCoreTests.cs
diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/ChangeFeed/DocumentServiceLeaseStoreManagerBuilderTests.cs b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/ChangeFeed/DocumentServiceLeaseStoreManagerBuilderTests.cs

Original file line number	Diff line number	Diff line change
`@@ -334,7 +334,8 @@ private async Task InitializeLeaseStoreAsync(ITrace trace, CancellationToken can`
`334`	`334`	`monitoredContainer: this.monitoredContainer,`
`335`	`335`	`leaseContainer: this.leaseContainer,`
`336`	`336`	`leaseContainerPrefix: leasePrefix,`
`337`		`- instanceName: ChangeFeedEstimatorIterator.EstimatorDefaultHostName);`
	`337`	`+ instanceName: ChangeFeedEstimatorIterator.EstimatorDefaultHostName,`
	`338`	`+ changeFeedMode: ChangeFeedMode.LatestVersion);`
`338`	`339`
`339`	`340`	`this.documentServiceLeaseContainer = documentServiceLeaseStoreManager.LeaseContainer;`
`340`	`341`	`}`
Original file line number	Diff line number	Diff line change
`@@ -70,13 +70,15 @@ public ChangeFeedProcessorBuilder WithInstanceName(string instanceName)`
`70`	`70`	`}`
`71`	`71`
`72`	`72`	`/// <summary>`
`73`		`- /// Sets the mode for the change freed processor.`
	`73`	`+ /// Sets the mode for the change feed processor.`
`74`	`74`	`/// </summary>`
`75`	`75`	`/// <param name="changeFeedMode"></param>`
`76`	`76`	`/// <returns>The instance of <see cref="ChangeFeedProcessorBuilder"/> to use.</returns>`
`77`	`77`	`internal ChangeFeedProcessorBuilder WithChangeFeedMode(ChangeFeedMode changeFeedMode)`
`78`	`78`	`{`
`79`	`79`	`this.changeFeedProcessorOptions.Mode = changeFeedMode;`
	`80`	`+ this.changeFeedLeaseOptions.Mode = changeFeedMode;`
	`81`	`+`
`80`	`82`	`return this;`
`81`	`83`	`}`
`82`	`84`
Original file line number	Diff line number	Diff line change
`@@ -44,5 +44,10 @@ public ChangeFeedLeaseOptions()`
`44`	`44`	`/// instances pointing at the same feed while using the same auxiliary collection.`
`45`	`45`	`/// </summary>`
`46`	`46`	`public string LeasePrefix { get; set; }`
	`47`	`+`
	`48`	`+ /// <summary>`
	`49`	`+ /// Gets or sets the <see cref="ChangeFeedMode"/>.`
	`50`	`+ /// </summary>`
	`51`	`+ public ChangeFeedMode Mode { get; set; }`
`47`	`52`	`}`
`48`	`53`	`}`
Original file line number	Diff line number	Diff line change
`@@ -70,5 +70,10 @@ internal abstract class DocumentServiceLease`
`70`	`70`	`/// Gets or sets custom lease properties which can be managed from <see cref="LoadBalancingStrategy"/>.`
`71`	`71`	`/// </summary>`
`72`	`72`	`public abstract Dictionary<string, string> Properties { get; set; }`
	`73`	`+`
	`74`	`+ /// <summary>`
	`75`	`+ /// Gets or sets the ChangeFeedMode.`
	`76`	`+ /// </summary>`
	`77`	`+ public abstract string Mode { get; set; }`
`73`	`78`	`}`
`74`	`79`	`}`
Original file line number	Diff line number	Diff line change
`@@ -95,6 +95,9 @@ public override DateTime Timestamp`
`95`	`95`	`[JsonProperty("_ts")]`
`96`	`96`	`private long TS { get; set; }`
`97`	`97`
	`98`	`+ [JsonProperty("Mode", NullValueHandling = NullValueHandling.Ignore)]`
	`99`	`+ public override string Mode { get; set; }`
	`100`	`+`
`98`	`101`	`public override string ToString()`
`99`	`102`	`{`
`100`	`103`	`return string.Format(`