Change Feed Processor: Fixes LeaseLostException on Notifications API for Renewer

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/FeedManagement/PartitionControllerCore.cs

@@ -161,6 +161,17 @@ private async Task ProcessPartitionAsync(DocumentServiceLease lease)
             {
                 DefaultTrace.TraceVerbose("Lease with token {0}: processing canceled", lease.CurrentLeaseToken);
             }
+            catch (LeaseLostException leaseLostException)
+            {
+                // LeaseLostException by itself is not loggable, unless it contains a related inner exception
+                // For cases when the lease or container has been deleted or the lease has been stolen
+                if (leaseLostException.InnerException != null)
+                {
+                    await this.monitor.NotifyErrorAsync(lease.CurrentLeaseToken, leaseLostException.InnerException);
+                }
+
+                DefaultTrace.TraceVerbose("Lease with token {0}: lease was lost", lease.CurrentLeaseToken);
+            }
             catch (Exception ex)
             {
                 await this.monitor.NotifyErrorAsync(lease.CurrentLeaseToken, ex);

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

@@ -254,7 +254,7 @@ public async Task Controller_ShouldNotify_IfProcessingFails()
                     manualResetEvent.Set();
                     throw exception;
                 });
-
+            
             Mock.Get(this.partitionSupervisorFactory)
                 .Setup(f => f.Create(this.lease))
                 .Returns(supervisor.Object);
@@ -532,6 +532,89 @@ public async Task Shutdown_ShouldNotify_Monitor()
                 .Verify(m => m.NotifyLeaseReleaseAsync(this.lease.CurrentLeaseToken), Times.Once);
         }
 
+        [TestMethod]
+        public async Task Renewer_ShouldNotNotify_IfLeaseAcquireFailsWithLeaseLost_WithoutInnerException()
+        {
+            Mock.Get(this.partitionProcessor)
+                .Reset();
+
+            Mock<PartitionSupervisor> supervisor = new Mock<PartitionSupervisor>();
+
+            LeaseLostException exception = new LeaseLostException();
+
+            ManualResetEvent manualResetEvent = new ManualResetEvent(false);
+
+            supervisor
+                .Setup(s => s.RunAsync(It.IsAny<CancellationToken>()))
+                .ThrowsAsync(exception)
+                .Callback((CancellationToken ct) =>
+                {
+                    manualResetEvent.Set();
+                    throw exception;
+                });
+
+            Mock.Get(this.partitionSupervisorFactory)
+                .Setup(f => f.Create(this.lease))
+                .Returns(supervisor.Object);
+
+            await this.sut.AddOrUpdateLeaseAsync(this.lease);
+
+            bool timeout = manualResetEvent.WaitOne(100);
+            Assert.IsTrue(timeout, "Partition supervisor not started");
+
+            this.healthMonitor
+                .Verify(m => m.NotifyErrorAsync(this.lease.CurrentLeaseToken, exception), Times.Never);
+
+            Mock.Get(this.leaseManager)
+                .Verify(manager => manager.ReleaseAsync(this.lease), Times.Once);
+
+            this.healthMonitor
+                .Verify(m => m.NotifyLeaseReleaseAsync(this.lease.CurrentLeaseToken), Times.Once);
+        }
+
+        [TestMethod]
+        public async Task Renewer_ShouldNotify_IfLeaseAcquireFailsWithLeaseLost_WithInnerException()
+        {
+            Mock.Get(this.partitionProcessor)
+                .Reset();
+
+            Mock<PartitionSupervisor> supervisor = new Mock<PartitionSupervisor>();
+
+            Exception internalException = new Exception();
+
+            LeaseLostException exception = new LeaseLostException("some error", internalException);
+
+            ManualResetEvent manualResetEvent = new ManualResetEvent(false);
+
+            supervisor
+                .Setup(s => s.RunAsync(It.IsAny<CancellationToken>()))
+                .Callback((CancellationToken ct) =>
+                {
+                    manualResetEvent.Set();
+                    throw exception;
+                });
+
+            Mock.Get(this.partitionSupervisorFactory)
+                .Setup(f => f.Create(this.lease))
+                .Returns(supervisor.Object);
+
+            await this.sut.AddOrUpdateLeaseAsync(this.lease).ConfigureAwait(false);
+
+            bool timeout = manualResetEvent.WaitOne(100);
+            Assert.IsTrue(timeout, "Partition supervisor not started");
+
+            this.healthMonitor
+                .Verify(m => m.NotifyErrorAsync(this.lease.CurrentLeaseToken, internalException), Times.Once);
+
+            Mock.Get(this.leaseManager)
+                .Verify(manager => manager.ReleaseAsync(this.lease), Times.Once);
+
+            this.healthMonitor
+                .Verify(m => m.NotifyLeaseReleaseAsync(this.lease.CurrentLeaseToken), Times.Once);
+        }
+
+
+
         public Task InitializeAsync()
         {
             return Task.FromResult(false);