ChangeFeedProcessor: Fixes blocking review comments for lease export

1. Eager MemoryStream validation: WithInMemoryLeaseContainer now
   checks that the stream is expandable at config time via
   SetLength(same length). A MemoryStream(byte[]) is writable but
   not expandable - SetLength/Write throws NotSupportedException
   when serialized data exceeds original capacity. Failing fast at
   config time prevents silent data loss at shutdown.

2. StopAsync propagates ShutdownAsync exceptions: Removed try/finally
   and catch-swallow pattern. ShutdownAsync runs sequentially after
   partitionManager.StopAsync() so persistence failures propagate
   to the caller. partitionManager.StopAsync() rarely fails, and
   when it does the caller already receives that exception.
   Silently swallowing persistence failures defeats the feature
   guarantee.

3. Log timing: Moved 'Processor stopped.' trace after ShutdownAsync
   for consistency with StartAsync pattern.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: kirankumarkolli

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

@@ -266,6 +266,21 @@ public virtual ChangeFeedProcessorBuilder WithInMemoryLeaseContainer(MemoryStrea
                 throw new ArgumentException("The lease state stream must be writable so that state can be persisted on shutdown.", nameof(leaseState));
             }
 
+            // Verify the stream is expandable. A MemoryStream created via
+            // new MemoryStream(byte[]) is writable but not expandable and will
+            // throw NotSupportedException when ShutdownAsync tries to resize it.
+            try
+            {
+                leaseState.SetLength(leaseState.Length);
+            }
+            catch (NotSupportedException)
+            {
+                throw new ArgumentException(
+                    "The lease state stream must be resizable. Use 'new MemoryStream()' and write bytes into it "
+                    + "instead of 'new MemoryStream(byte[])' to create a resizable stream.",
+                    nameof(leaseState));
+            }
+
             this.ValidateNoLeaseContainerConfigured();
 
             if (string.IsNullOrEmpty(this.InstanceName))

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

@@ -78,28 +78,18 @@ public override async Task StartAsync()
         public override async Task StopAsync()
         {
             DefaultTrace.TraceInformation("Stopping processor...");
-            try
-            {
-                await this.partitionManager.StopAsync().ConfigureAwait(false);
-            }
-            finally
-            {
-                // Persist in-memory lease state even if partitionManager.StopAsync() throws,
-                // so that progress is not silently lost.
-                try
-                {
-                    await this.documentServiceLeaseStoreManager
-                        .LeaseContainer
-                        .ShutdownAsync()
-                        .ConfigureAwait(false);
-                }
-                catch (Exception shutdownException)
-                {
-                    Cosmos.Extensions.TraceException(shutdownException);
-                    DefaultTrace.TraceWarning(
-                        "Failed to persist in-memory lease state during shutdown.");
-                }
-            }
+            await this.partitionManager.StopAsync().ConfigureAwait(false);
+
+            // ShutdownAsync persists in-memory lease state. Runs outside finally
+            // because: (a) partitionManager.StopAsync() rarely fails, and when it
+            // does the caller already receives that exception; (b) letting
+            // ShutdownAsync exceptions propagate ensures callers know if
+            // persistence failed — silently swallowing defeats the feature's
+            // guarantee.
+            await this.documentServiceLeaseStoreManager
+                .LeaseContainer
+                .ShutdownAsync()
+                .ConfigureAwait(false);
 
             DefaultTrace.TraceInformation("Processor stopped.");
         }