[Preview] Change Feed Processor: Adds notification APIs

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

@@ -22,6 +22,7 @@ internal sealed class ChangeFeedEstimatorRunner : ChangeFeedProcessor
         private const string EstimatorDefaultHostName = "Estimator";
         private readonly ChangesEstimationHandler initialEstimateDelegate;
         private readonly TimeSpan? estimatorPeriod;
+        private ChangeFeedProcessorHealthMonitor healthMonitor;
         private CancellationTokenSource shutdownCts;
         private ContainerInternal leaseContainer;
         private ContainerInternal monitoredContainer;
@@ -75,6 +76,7 @@ public void ApplyBuildConfiguration(
             this.monitoredContainer = monitoredContainer ?? throw new ArgumentNullException(nameof(monitoredContainer));
             this.changeFeedLeaseOptions = changeFeedLeaseOptions;
             this.documentServiceLeaseContainer = customDocumentServiceLeaseStoreManager?.LeaseContainer;
+            this.healthMonitor = changeFeedProcessorOptions.HealthMonitor;
         }
 
         public override async Task StartAsync()
@@ -128,7 +130,7 @@ private FeedEstimatorRunner BuildFeedEstimatorRunner()
                    this.documentServiceLeaseContainer);
             }
 
-            return new FeedEstimatorRunner(this.initialEstimateDelegate, this.remainingWorkEstimator, this.estimatorPeriod);
+            return new FeedEstimatorRunner(this.initialEstimateDelegate, this.remainingWorkEstimator, this.healthMonitor, this.estimatorPeriod);
         }
 
         private async Task InitializeLeaseStoreAsync()

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

@@ -7,6 +7,11 @@ namespace Microsoft.Azure.Cosmos
     using System;
     using Microsoft.Azure.Cosmos.ChangeFeed.Configuration;
     using Microsoft.Azure.Cosmos.ChangeFeed.LeaseManagement;
+#if PREVIEW
+    using static Microsoft.Azure.Cosmos.Container;
+#else
+    using static Microsoft.Azure.Cosmos.ContainerInternal;
+#endif
 
     /// <summary>
     /// Provides a flexible way to create an instance of <see cref="ChangeFeedProcessor"/> with custom set of parameters.
@@ -25,7 +30,7 @@ public class ChangeFeedProcessorBuilder
                 ChangeFeedProcessorOptions,
                 ContainerInternal> applyBuilderConfiguration;
 
-        private ChangeFeedProcessorOptions changeFeedProcessorOptions;
+        private readonly ChangeFeedProcessorOptions changeFeedProcessorOptions = new ChangeFeedProcessorOptions();
 
         private ContainerInternal leaseContainer;
         private string InstanceName;
@@ -96,7 +101,6 @@ public ChangeFeedProcessorBuilder WithPollInterval(TimeSpan pollInterval)
                 throw new ArgumentNullException(nameof(pollInterval));
             }
 
-            this.changeFeedProcessorOptions = this.changeFeedProcessorOptions ?? new ChangeFeedProcessorOptions();
             this.changeFeedProcessorOptions.FeedPollDelay = pollInterval;
             return this;
         }
@@ -114,7 +118,6 @@ public ChangeFeedProcessorBuilder WithPollInterval(TimeSpan pollInterval)
         /// <returns>The instance of <see cref="ChangeFeedProcessorBuilder"/> to use.</returns>
         internal virtual ChangeFeedProcessorBuilder WithStartFromBeginning()
         {
-            this.changeFeedProcessorOptions = this.changeFeedProcessorOptions ?? new ChangeFeedProcessorOptions();
             this.changeFeedProcessorOptions.StartFromBeginning = true;
             return this;
         }
@@ -137,7 +140,6 @@ public ChangeFeedProcessorBuilder WithStartTime(DateTime startTime)
                 throw new ArgumentNullException(nameof(startTime));
             }
 
-            this.changeFeedProcessorOptions = this.changeFeedProcessorOptions ?? new ChangeFeedProcessorOptions();
             this.changeFeedProcessorOptions.StartTime = startTime;
             return this;
         }
@@ -154,7 +156,6 @@ public ChangeFeedProcessorBuilder WithMaxItems(int maxItemCount)
                 throw new ArgumentOutOfRangeException(nameof(maxItemCount));
             }
 
-            this.changeFeedProcessorOptions = this.changeFeedProcessorOptions ?? new ChangeFeedProcessorOptions();
             this.changeFeedProcessorOptions.MaxItemCount = maxItemCount;
             return this;
         }
@@ -213,6 +214,69 @@ internal virtual ChangeFeedProcessorBuilder WithInMemoryLeaseContainer()
             return this;
         }
 
+        /// <summary>
+        /// Defines a delegate to receive notifications on errors that occur during change feed processor execution.
+        /// </summary>
+        /// <param name="errorDelegate">A delegate to receive notifications for change feed processor related errors.</param>
+        /// <returns>The instance of <see cref="ChangeFeedProcessorBuilder"/> to use.</returns>
+#if PREVIEW
+        public
+#else
+        internal
+#endif
+        ChangeFeedProcessorBuilder WithErrorNotification(ChangeFeedMonitorErrorDelegate errorDelegate)
+        {
+            if (errorDelegate == null)
+            {
+                throw new ArgumentNullException(nameof(errorDelegate));
+            }
+
+            this.changeFeedProcessorOptions.HealthMonitor.SetErrorDelegate(errorDelegate);
+            return this;
+        }
+
+        /// <summary>
+        /// Defines a delegate to receive notifications on lease acquires that occur during change feed processor execution.
+        /// </summary>
+        /// <param name="acquireDelegate">A delegate to receive notifications when a change feed processor acquires a lease.</param>
+        /// <returns>The instance of <see cref="ChangeFeedProcessorBuilder"/> to use.</returns>
+#if PREVIEW
+        public
+#else
+        internal
+#endif
+        ChangeFeedProcessorBuilder WithLeaseAcquireNotification(ChangeFeedMonitorLeaseAcquireDelegate acquireDelegate)
+        {
+            if (acquireDelegate == null)
+            {
+                throw new ArgumentNullException(nameof(acquireDelegate));
+            }
+
+            this.changeFeedProcessorOptions.HealthMonitor.SetLeaseAcquireDelegate(acquireDelegate);
+            return this;
+        }
+
+        /// <summary>
+        /// Defines a delegate to receive notifications on lease releases that occur during change feed processor execution.
+        /// </summary>
+        /// <param name="releaseDelegate">A delegate to receive notifications when a change feed processor releases a lease.</param>
+        /// <returns>The instance of <see cref="ChangeFeedProcessorBuilder"/> to use.</returns>
+#if PREVIEW
+        public
+#else
+        internal
+#endif
+        ChangeFeedProcessorBuilder WithLeaseReleaseNotification(ChangeFeedMonitorLeaseReleaseDelegate releaseDelegate)
+        {
+            if (releaseDelegate == null)
+            {
+                throw new ArgumentNullException(nameof(releaseDelegate));
+            }
+
+            this.changeFeedProcessorOptions.HealthMonitor.SetLeaseReleaseDelegate(releaseDelegate);
+            return this;
+        }
+
         /// <summary>
         /// Builds a new instance of the <see cref="ChangeFeedProcessor"/> with the specified configuration.
         /// </summary>
@@ -239,16 +303,10 @@ public ChangeFeedProcessor Build()
                 throw new InvalidOperationException("Processor name not specified during creation.");
             }
 
-            this.InitializeDefaultOptions();
             this.applyBuilderConfiguration(this.LeaseStoreManager, this.leaseContainer, this.InstanceName, this.changeFeedLeaseOptions, this.changeFeedProcessorOptions, this.monitoredContainer);
 
             this.isBuilt = true;
             return this.changeFeedProcessor;
         }
-
-        private void InitializeDefaultOptions()
-        {
-            this.changeFeedProcessorOptions = this.changeFeedProcessorOptions ?? new ChangeFeedProcessorOptions();
-        }
     }
 }

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

@@ -11,7 +11,6 @@ namespace Microsoft.Azure.Cosmos.ChangeFeed
     using Microsoft.Azure.Cosmos.ChangeFeed.FeedManagement;
     using Microsoft.Azure.Cosmos.ChangeFeed.FeedProcessing;
     using Microsoft.Azure.Cosmos.ChangeFeed.LeaseManagement;
-    using Microsoft.Azure.Cosmos.ChangeFeed.Monitoring;
     using Microsoft.Azure.Cosmos.ChangeFeed.Utils;
     using Microsoft.Azure.Cosmos.Core.Trace;
     using Microsoft.Azure.Cosmos.Tracing;
@@ -117,9 +116,13 @@ private PartitionManager BuildPartitionManager(
                     EqualPartitionsBalancingStrategy.DefaultMaxLeaseCount,
                     this.changeFeedLeaseOptions.LeaseExpirationInterval);
 
-            PartitionController partitionController = new PartitionControllerCore(this.documentServiceLeaseStoreManager.LeaseContainer, this.documentServiceLeaseStoreManager.LeaseManager, partitionSuperviserFactory, synchronizer);
+            PartitionController partitionController = new PartitionControllerCore(
+                this.documentServiceLeaseStoreManager.LeaseContainer, 
+                this.documentServiceLeaseStoreManager.LeaseManager, 
+                partitionSuperviserFactory, 
+                synchronizer,
+                this.changeFeedProcessorOptions.HealthMonitor);
 
-            partitionController = new HealthMonitoringPartitionControllerDecorator(partitionController, new TraceHealthMonitor());
             PartitionLoadBalancerCore partitionLoadBalancer = new PartitionLoadBalancerCore(
                 partitionController,
                 this.documentServiceLeaseStoreManager.LeaseContainer,

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

@@ -6,6 +6,7 @@ namespace Microsoft.Azure.Cosmos.ChangeFeed.Configuration
 {
     using System;
     using Microsoft.Azure.Cosmos;
+    using Microsoft.Azure.Cosmos.ChangeFeed.Monitoring;
 
     /// <summary>
     /// Options to control various aspects of partition distribution happening within <see cref="ChangeFeedProcessorCore"/> instance.
@@ -81,5 +82,7 @@ public DateTime? StartTime
         /// </remarks>
         /// <seealso cref="ChangeFeedOptions.StartFromBeginning"/>
         public bool StartFromBeginning { get; set; }
+
+        public ChangeFeedProcessorHealthMonitorCore HealthMonitor { get; set; } = new ChangeFeedProcessorHealthMonitorCore();
     }
 }

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/DocDBErrors/DocDbError.cs

@@ -7,8 +7,6 @@ namespace Microsoft.Azure.Cosmos.ChangeFeed.DocDBErrors
     internal enum DocDbError
     {
         Undefined,
-        PartitionSplit,
-        PartitionNotFound,
-        ReadSessionNotAvailable
+        PartitionSplit
     }
 }

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/DocDBErrors/ExceptionClassifier.cs

@@ -18,11 +18,6 @@ public static DocDbError ClassifyStatusCodes(
                 return DocDbError.PartitionSplit;
             }
 
-            if (statusCode == HttpStatusCode.NotFound)
-            {
-                return subStatusCode == (int)SubStatusCodes.ReadSessionNotAvailable ? DocDbError.ReadSessionNotAvailable : DocDbError.PartitionNotFound;
-            }
-
             return DocDbError.Undefined;
         }
     }

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/Exceptions/ObserverException.cs → Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/Exceptions/ChangeFeedProcessorUserException.cs

@@ -2,7 +2,7 @@
 // Copyright (c) Microsoft Corporation.  All rights reserved.
 //------------------------------------------------------------
 
-namespace Microsoft.Azure.Cosmos.ChangeFeed.Exceptions
+namespace Microsoft.Azure.Cosmos
 {
     using System;
     using System.Runtime.Serialization;
@@ -11,29 +11,44 @@ namespace Microsoft.Azure.Cosmos.ChangeFeed.Exceptions
     /// Exception occurred when an operation in an IChangeFeedObserver is running and throws by user code
     /// </summary>
     [Serializable]
-    internal class ObserverException : Exception
+
+#if PREVIEW
+    public
+#else
+    internal
+#endif
+    class ChangeFeedProcessorUserException : Exception
     {
-        private static readonly string DefaultMessage = "Exception has been thrown by the Observer.";
+        private static readonly string DefaultMessage = "Exception has been thrown by the change feed processor delegate.";
 
         /// <summary>
-        /// Initializes a new instance of the <see cref="ObserverException" /> class using the specified internal exception.
+        /// Initializes a new instance of the <see cref="ChangeFeedProcessorUserException " /> class using the specified internal exception.
         /// </summary>
         /// <param name="originalException"><see cref="Exception"/> thrown by the user code.</param>
-        public ObserverException(Exception originalException)
-            : base(ObserverException.DefaultMessage, originalException)
+        /// <param name="context">Context under which the exception occurred.</param>
+        public ChangeFeedProcessorUserException(
+            Exception originalException,
+            ChangeFeedProcessorContext context)
+            : base(ChangeFeedProcessorUserException.DefaultMessage, originalException)
         {
+            this.ExceptionContext = context;
         }
 
         /// <summary>
-        /// Initializes a new instance of the <see cref="ObserverException" /> class using default values.
+        /// Initializes a new instance of the <see cref="ChangeFeedProcessorUserException " /> for serialization purposes.
         /// </summary>
         /// <param name="info">The SerializationInfo object that holds serialized object data for the exception being thrown.</param>
         /// <param name="context">The StreamingContext that contains contextual information about the source or destination.</param>
-        protected ObserverException(SerializationInfo info, StreamingContext context)
-            : this((Exception)info.GetValue("InnerException", typeof(Exception)))
+        protected ChangeFeedProcessorUserException(SerializationInfo info, StreamingContext context)
+            : this((Exception)info.GetValue("InnerException", typeof(Exception)), null)
         {
         }
 
+        /// <summary>
+        /// Contextual information that identifies which was the payload that was delivered to the delegate when this error occurred.
+        /// </summary>
+        public ChangeFeedProcessorContext ExceptionContext { get; private set; }
+
         /// <summary>
         /// Sets the System.Runtime.Serialization.SerializationInfo with information about the exception.
         /// </summary>