Change Feed / Processor AVAD: Fixes timeToLiveExpired missing from me…

…tadata (#4523) * longrunning cfp avad test for ttl * add more to comment * internal set * other tests fail with higher ms * run updatecontracts * Update Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.EmulatorTests/ChangeFeed/GetChangeFeedProcessorBuilderWithAllVersionsAndDeletesTests.cs Co-authored-by: Matias Quaranta <[email protected]> * Update Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.EmulatorTests/ChangeFeed/GetChangeFeedProcessorBuilderWithAllVersionsAndDeletesTests.cs Co-authored-by: Matias Quaranta <[email protected]> * Update Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.EmulatorTests/ChangeFeed/GetChangeFeedProcessorBuilderWithAllVersionsAndDeletesTests.cs Co-authored-by: Matias Quaranta <[email protected]> * Update Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.EmulatorTests/ChangeFeed/GetChangeFeedProcessorBuilderWithAllVersionsAndDeletesTests.cs Co-authored-by: Matias Quaranta <[email protected]> * using Logger.LogLine * change back over to ManualResetEvent. fixed assert to look at Previous. keeping stopwatch just for logging. timeout at 5 minutes. * Update GetChangeFeedProcessorBuilderWithAllVersionsAndDeletesTests.cs Co-authored-by: Matias Quaranta <[email protected]> * try/finally * removing ctor. making all set internal to address serialization issue. later PRs to test STJ de/serialization. * fixing de/serialziation issue * Update GetChangeFeedProcessorBuilderWithAllVersionsAndDeletesTests.cs change timeout * Update ReEncryption.csproj 1.11.4 * internal set change to CanWrite:True; --------- Co-authored-by: Matias Quaranta <[email protected]>
Azure · Jun 19, 2024 · aab02e5 · aab02e5
1 parent c5ac103
commit aab02e5
Show file tree

Hide file tree

Showing 3 changed files with 125 additions and 31 deletions.
diff --git a/Microsoft.Azure.Cosmos/src/Resource/FullFidelity/ChangeFeedMetadata.cs b/Microsoft.Azure.Cosmos/src/Resource/FullFidelity/ChangeFeedMetadata.cs
@@ -19,55 +19,36 @@ namespace Microsoft.Azure.Cosmos
 #endif 
         class ChangeFeedMetadata
     {
-        /// <summary>
-        /// New instance of meta data for <see cref="ChangeFeedItem{T}"/> created.
-        /// </summary>
-        /// <param name="conflictResolutionTimestamp"></param>
-        /// <param name="lsn"></param>
-        /// <param name="operationType"></param>
-        /// <param name="previousLsn"></param>
-        public ChangeFeedMetadata(
-            DateTime conflictResolutionTimestamp,
-            long lsn,
-            ChangeFeedOperationType operationType,
-            long previousLsn)
-        {
-            this.ConflictResolutionTimestamp = conflictResolutionTimestamp;
-            this.Lsn = lsn;
-            this.OperationType = operationType;
-            this.PreviousLsn = previousLsn;
-        }
-
         /// <summary>
         /// The conflict resolution timestamp.
         /// </summary>
         [JsonProperty(PropertyName = "crts", NullValueHandling = NullValueHandling.Ignore)]
         [JsonConverter(typeof(UnixDateTimeConverter))]
-        public DateTime ConflictResolutionTimestamp { get; }
+        public DateTime ConflictResolutionTimestamp { get; internal set; }
 
         /// <summary>
         /// The current logical sequence number.
         /// </summary>
         [JsonProperty(PropertyName = "lsn", NullValueHandling = NullValueHandling.Ignore)]
-        public long Lsn { get; }
+        public long Lsn { get; internal set;  }
 
         /// <summary>
         /// The change feed operation type.
         /// </summary>
         [JsonProperty(PropertyName = "operationType")]
         [JsonConverter(typeof(StringEnumConverter))]
-        public ChangeFeedOperationType OperationType { get; }
+        public ChangeFeedOperationType OperationType { get; internal set; }
 
         /// <summary>
         /// The previous logical sequence number.
         /// </summary>
         [JsonProperty(PropertyName = "previousImageLSN", NullValueHandling = NullValueHandling.Ignore)]
-        public long PreviousLsn { get; }
+        public long PreviousLsn { get; internal set; }
 
         /// <summary>
         /// Used to distinquish explicit deletes (e.g. via DeleteItem) from deletes caused by TTL expiration (a collection may define time-to-live policy for documents).
         /// </summary>
         [JsonProperty(PropertyName = "timeToLiveExpired", NullValueHandling= NullValueHandling.Ignore)]
-        public bool IsTimeToLiveExpired { get; }
+        public bool IsTimeToLiveExpired { get; internal set; }
     }
 }
diff --git a/...s.EmulatorTests/ChangeFeed/GetChangeFeedProcessorBuilderWithAllVersionsAndDeletesTests.cs b/...s.EmulatorTests/ChangeFeed/GetChangeFeedProcessorBuilderWithAllVersionsAndDeletesTests.cs
@@ -6,11 +6,14 @@ namespace Microsoft.Azure.Cosmos.SDK.EmulatorTests.ChangeFeed
 {
     using System;
     using System.Collections.Generic;
+    using System.Diagnostics;
     using System.Linq;
     using System.Threading;
     using System.Threading.Tasks;
     using Microsoft.Azure.Cosmos.ChangeFeed.Utils;
+    using Microsoft.Azure.Cosmos.Services.Management.Tests;
     using Microsoft.VisualStudio.TestTools.UnitTesting;
+    using Newtonsoft.Json;
     using Newtonsoft.Json.Linq;
 
     [TestClass]
@@ -29,6 +32,115 @@ public async Task Cleanup()
             await base.TestCleanup();
         }
 
+        [TestMethod]
+        [Timeout(300000)]
+        [TestCategory("LongRunning")]
+        [Owner("philipthomas-MSFT")]
+        [Description("Scenario: When a document is created with ttl set, there should be 1 create and 1 delete that will appear for that " +
+            "document when using ChangeFeedProcessor with AllVersionsAndDeletes set as the ChangeFeedMode.")]
+        public async Task WhenADocumentIsCreatedWithTtlSetThenTheDocumentIsDeletedTestsAsync()
+        {
+            ContainerInternal monitoredContainer = await this.CreateMonitoredContainer(ChangeFeedMode.AllVersionsAndDeletes);
+            Exception exception = default;
+            int ttlInSeconds = 5;
+            Stopwatch stopwatch = new();
+            ManualResetEvent allDocsProcessed = new ManualResetEvent(false);
+
+            ChangeFeedProcessor processor = monitoredContainer
+                .GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes(processorName: "processor", onChangesDelegate: (ChangeFeedProcessorContext context, IReadOnlyCollection<ChangeFeedItem<dynamic>> docs, CancellationToken token) =>
+                {
+                    // NOTE(philipthomas-MSFT): Please allow these Logger.LogLine because TTL on items will purge at random times so I am using this to test when ran locally using emulator.
+
+                    Logger.LogLine($"@ {DateTime.Now}, {nameof(stopwatch)} -> CFP AVAD took '{stopwatch.ElapsedMilliseconds}' to read document CRUD in feed.");
+                    Logger.LogLine($"@ {DateTime.Now}, {nameof(docs)} -> {JsonConvert.SerializeObject(docs)}");
+
+                    foreach (ChangeFeedItem<dynamic> change in docs)
+                    {
+                        if (change.Metadata.OperationType == ChangeFeedOperationType.Create)
+                        {
+                            // current
+                            Assert.AreEqual(expected: "1", actual: change.Current.id.ToString());
+                            Assert.AreEqual(expected: "1", actual: change.Current.pk.ToString());
+                            Assert.AreEqual(expected: "Testing TTL on CFP.", actual: change.Current.description.ToString());
+                            Assert.AreEqual(expected: ttlInSeconds, actual: change.Current.ttl.ToObject<int>());
+
+                            // metadata
+                            Assert.IsTrue(DateTime.TryParse(s: change.Metadata.ConflictResolutionTimestamp.ToString(), out _), message: "Invalid csrt must be a datetime value.");
+                            Assert.IsTrue(change.Metadata.Lsn > 0, message: "Invalid lsn must be a long value.");
+                            Assert.IsFalse(change.Metadata.IsTimeToLiveExpired);
+
+                            // previous
+                            Assert.IsNull(change.Previous);
+                        }
+                        else if (change.Metadata.OperationType == ChangeFeedOperationType.Delete)
+                        {
+                            // current
+                            Assert.IsNull(change.Current.id);
+
+                            // metadata
+                            Assert.IsTrue(DateTime.TryParse(s: change.Metadata.ConflictResolutionTimestamp.ToString(), out _), message: "Invalid csrt must be a datetime value.");
+                            Assert.IsTrue(change.Metadata.Lsn > 0, message: "Invalid lsn must be a long value.");
+                            Assert.IsTrue(change.Metadata.IsTimeToLiveExpired);
+
+                            // previous
+                            Assert.AreEqual(expected: "1", actual: change.Previous.id.ToString());
+                            Assert.AreEqual(expected: "1", actual: change.Previous.pk.ToString());
+                            Assert.AreEqual(expected: "Testing TTL on CFP.", actual: change.Previous.description.ToString());
+                            Assert.AreEqual(expected: ttlInSeconds, actual: change.Previous.ttl.ToObject<int>());
+
+                            // stop after reading delete since it is the last document in feed.
+                            stopwatch.Stop();
+                            allDocsProcessed.Set();
+                        }
+                        else
+                        {
+                            Assert.Fail("Invalid operation.");
+                        }
+                    }
+
+                    return Task.CompletedTask;
+                })
+                .WithInstanceName(Guid.NewGuid().ToString())
+                .WithLeaseContainer(this.LeaseContainer)
+                .WithErrorNotification((leaseToken, error) =>
+                {
+                    exception = error.InnerException;
+
+                    return Task.CompletedTask;
+                })
+                .Build();
+
+            stopwatch.Start();
+
+            // NOTE(philipthomas-MSFT): Please allow these Logger.LogLine because TTL on items will purge at random times so I am using this to test when ran locally using emulator.
+
+            Logger.LogLine($"@ {DateTime.Now}, CFProcessor starting...");
+
+            await processor.StartAsync();
+
+            try
+            {
+                await Task.Delay(GetChangeFeedProcessorBuilderWithAllVersionsAndDeletesTests.ChangeFeedSetupTime);
+                await monitoredContainer.CreateItemAsync<dynamic>(new { id = "1", pk = "1", description = "Testing TTL on CFP.", ttl = ttlInSeconds }, partitionKey: new PartitionKey("1"));
+
+                // NOTE(philipthomas-MSFT): Please allow these Logger.LogLine because TTL on items will purge at random times so I am using this to test when ran locally using emulator.
+
+                Logger.LogLine($"@ {DateTime.Now}, Document created.");
+
+                bool receivedDelete = allDocsProcessed.WaitOne(250000);
+                Assert.IsTrue(receivedDelete, "Timed out waiting for docs to process");
+
+                if (exception != default)
+                {
+                    Assert.Fail(exception.ToString());
+                }
+            }
+            finally
+            {
+                await processor.StopAsync();
+            }
+        }
+
         [TestMethod]
         [Owner("philipthomas-MSFT")]
         [Description("Scenario: When a document is created, then updated, and finally deleted, there should be 3 changes that will appear for that " +
@@ -467,6 +579,7 @@ private async Task<ContainerInternal> CreateMonitoredContainer(ChangeFeedMode ch
             if (changeFeedMode == ChangeFeedMode.AllVersionsAndDeletes)
             {
                 properties.ChangeFeedPolicy.FullFidelityRetention = TimeSpan.FromMinutes(5);
+                properties.DefaultTimeToLive = -1;
             }
 
             ContainerResponse response = await this.database.CreateContainerAsync(properties,

diff --git a/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Contracts/DotNetPreviewSDKAPI.json b/Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/Contracts/DotNetPreviewSDKAPI.json
@@ -89,7 +89,7 @@
           "Attributes": [
             "JsonPropertyAttribute"
           ],
-          "MethodInfo": "Boolean IsTimeToLiveExpired;CanRead:True;CanWrite:False;Boolean get_IsTimeToLiveExpired();IsAbstract:False;IsStatic:False;IsVirtual:False;IsGenericMethod:False;IsConstructor:False;IsFinal:False;"
+          "MethodInfo": "Boolean IsTimeToLiveExpired;CanRead:True;CanWrite:True;Boolean get_IsTimeToLiveExpired();IsAbstract:False;IsStatic:False;IsVirtual:False;IsGenericMethod:False;IsConstructor:False;IsFinal:False;"
         },
         "Int64 get_Lsn()[System.Runtime.CompilerServices.CompilerGeneratedAttribute()]": {
           "Type": "Method",
@@ -110,14 +110,14 @@
           "Attributes": [
             "JsonPropertyAttribute"
           ],
-          "MethodInfo": "Int64 Lsn;CanRead:True;CanWrite:False;Int64 get_Lsn();IsAbstract:False;IsStatic:False;IsVirtual:False;IsGenericMethod:False;IsConstructor:False;IsFinal:False;"
+          "MethodInfo": "Int64 Lsn;CanRead:True;CanWrite:True;Int64 get_Lsn();IsAbstract:False;IsStatic:False;IsVirtual:False;IsGenericMethod:False;IsConstructor:False;IsFinal:False;"
         },
         "Int64 PreviousLsn[Newtonsoft.Json.JsonPropertyAttribute(NullValueHandling = 1, PropertyName = \"previousImageLSN\")]": {
           "Type": "Property",
           "Attributes": [
             "JsonPropertyAttribute"
           ],
-          "MethodInfo": "Int64 PreviousLsn;CanRead:True;CanWrite:False;Int64 get_PreviousLsn();IsAbstract:False;IsStatic:False;IsVirtual:False;IsGenericMethod:False;IsConstructor:False;IsFinal:False;"
+          "MethodInfo": "Int64 PreviousLsn;CanRead:True;CanWrite:True;Int64 get_PreviousLsn();IsAbstract:False;IsStatic:False;IsVirtual:False;IsGenericMethod:False;IsConstructor:False;IsFinal:False;"
         },
         "Microsoft.Azure.Cosmos.ChangeFeedOperationType get_OperationType()[System.Runtime.CompilerServices.CompilerGeneratedAttribute()]": {
           "Type": "Method",
@@ -132,15 +132,15 @@
             "JsonConverterAttribute",
             "JsonPropertyAttribute"
           ],
-          "MethodInfo": "Microsoft.Azure.Cosmos.ChangeFeedOperationType OperationType;CanRead:True;CanWrite:False;Microsoft.Azure.Cosmos.ChangeFeedOperationType get_OperationType();IsAbstract:False;IsStatic:False;IsVirtual:False;IsGenericMethod:False;IsConstructor:False;IsFinal:False;"
+          "MethodInfo": "Microsoft.Azure.Cosmos.ChangeFeedOperationType OperationType;CanRead:True;CanWrite:True;Microsoft.Azure.Cosmos.ChangeFeedOperationType get_OperationType();IsAbstract:False;IsStatic:False;IsVirtual:False;IsGenericMethod:False;IsConstructor:False;IsFinal:False;"
         },
         "System.DateTime ConflictResolutionTimestamp[Newtonsoft.Json.JsonPropertyAttribute(NullValueHandling = 1, PropertyName = \"crts\")]-[Newtonsoft.Json.JsonConverterAttribute(typeof(Microsoft.Azure.Documents.UnixDateTimeConverter))]": {
           "Type": "Property",
           "Attributes": [
             "JsonConverterAttribute",
             "JsonPropertyAttribute"
           ],
-          "MethodInfo": "System.DateTime ConflictResolutionTimestamp;CanRead:True;CanWrite:False;System.DateTime get_ConflictResolutionTimestamp();IsAbstract:False;IsStatic:False;IsVirtual:False;IsGenericMethod:False;IsConstructor:False;IsFinal:False;"
+          "MethodInfo": "System.DateTime ConflictResolutionTimestamp;CanRead:True;CanWrite:True;System.DateTime get_ConflictResolutionTimestamp();IsAbstract:False;IsStatic:False;IsVirtual:False;IsGenericMethod:False;IsConstructor:False;IsFinal:False;"
         },
         "System.DateTime get_ConflictResolutionTimestamp()[System.Runtime.CompilerServices.CompilerGeneratedAttribute()]": {
           "Type": "Method",
@@ -149,10 +149,10 @@
           ],
           "MethodInfo": "System.DateTime get_ConflictResolutionTimestamp();IsAbstract:False;IsStatic:False;IsVirtual:False;IsGenericMethod:False;IsConstructor:False;IsFinal:False;"
         },
-        "Void .ctor(System.DateTime, Int64, Microsoft.Azure.Cosmos.ChangeFeedOperationType, Int64)": {
+        "Void .ctor()": {
           "Type": "Constructor",
           "Attributes": [],
-          "MethodInfo": "[Void .ctor(System.DateTime, Int64, Microsoft.Azure.Cosmos.ChangeFeedOperationType, Int64), Void .ctor(System.DateTime, Int64, Microsoft.Azure.Cosmos.ChangeFeedOperationType, Int64)]"
+          "MethodInfo": "[Void .ctor(), Void .ctor()]"
         }
       },
       "NestedTypes": {}