Change Feed Processor: Fixes behavior with StartTime on Local (#3645)

* To UTC

* Test

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

@@ -66,7 +66,7 @@ public DateTime? StartTime
                     throw new ArgumentException("StartTime cannot have DateTimeKind.Unspecified", nameof(value));
                 }
 
-                this.startTime = value;
+                this.startTime = value.HasValue && value.Value.Kind == DateTimeKind.Local ? value.Value.ToUniversalTime() : value;
             }
         }
 

Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/ChangeFeed/ChangeFeedProcessorBuilderTests.cs

@@ -220,6 +220,41 @@ public void CanBuildWithInMemoryContainer()
             Assert.IsInstanceOfType(builder.Build(), typeof(ChangeFeedProcessor));
         }
 
+        [TestMethod]
+        public void ConvertsToUTC()
+        {
+            DateTime localTime = DateTime.Now;
+
+            Assert.AreEqual(DateTimeKind.Local, localTime.Kind);
+
+            Action<DocumentServiceLeaseStoreManager,
+                Container,
+                string,
+                ChangeFeedLeaseOptions,
+                ChangeFeedProcessorOptions,
+                Container> verifier = (DocumentServiceLeaseStoreManager leaseStoreManager,
+                Container leaseContainer,
+                string instanceName,
+                ChangeFeedLeaseOptions changeFeedLeaseOptions,
+                ChangeFeedProcessorOptions changeFeedProcessorOptions,
+                Container monitoredContainer) =>
+                {
+                    Assert.AreEqual(DateTimeKind.Utc, changeFeedProcessorOptions.StartTime.Value.Kind);
+                    Assert.AreEqual(localTime.ToUniversalTime(), changeFeedProcessorOptions.StartTime.Value);
+                };
+
+            ChangeFeedProcessorBuilder builder = new ChangeFeedProcessorBuilder("workflowName",
+                ChangeFeedProcessorBuilderTests.GetMockedContainer(),
+                ChangeFeedProcessorBuilderTests.GetMockedProcessor(),
+                verifier);
+
+            builder.WithLeaseContainer(ChangeFeedProcessorBuilderTests.GetMockedContainer());
+
+            builder.WithStartTime(localTime);
+
+            Assert.IsInstanceOfType(builder.Build(), typeof(ChangeFeedProcessor));
+        }
+
         private static ContainerInternal GetMockedContainer(string containerName = null)
         {
             Mock<ContainerInternal> mockedContainer = MockCosmosUtil.CreateMockContainer(containerName: containerName);