GatewayStoreClient: Fixes stream consumption bug in GatewayStoreClient.CreateDocumentClientExceptionAsync (#5291)

## Problem

During write region failover, Gateway returns a 403 response with
`application/json` content type but invalid JSON content. This triggers
a bug in `GatewayStoreClient.CreateDocumentClientExceptionAsync` where
the HTTP response stream is consumed twice:

1. First consumption happens when attempting to deserialize the response
as JSON (line 176)
2. When deserialization fails and the catch block is entered, execution
continues to the fallback logic (line 195) which tries to read the same
stream again
3. This results in an unhandled `InvalidOperationException: The stream
was already consumed. It cannot be read again.`

The original `DocumentClientException` with proper diagnostics is lost,
making debugging difficult.

## Root Cause

The issue was introduced when an `else` clause was removed and an empty
`catch` block was added to the JSON deserialization logic, causing the
same stream to be processed twice if deserialization fails.

## Solution

This PR implements a minimal fix that:

1. **Buffers the HTTP response content once** using
`ReadAsStringAsync()` before attempting JSON deserialization
2. **Creates a new MemoryStream** from the buffered content for the JSON
deserialization attempt
3. **Reuses the buffered content** in the fallback logic instead of
trying to read from the response stream again
4. **Fixes a typo** in the generic type parameter from `<e>` to
`<Error>`

## Changes Made

- Modified `CreateDocumentClientExceptionAsync` method to buffer content
once and reuse it
- Added explanatory comments for the stream consumption fix
- Added test case
`TestStreamConsumptionBugFixWhenJsonDeserializationFails` to verify the
fix
- All changes are surgical and preserve existing functionality

## Code Example

**Before (buggy):**
```csharp
// First read - consumes the stream
Stream contentAsStream = await responseMessage.Content.ReadAsStreamAsync();
Error error = JsonSerializable.LoadFrom<Error>(stream: contentAsStream);
// ... if this fails and throws, execution continues to:

// Second read - fails with "stream already consumed"
contextBuilder.AppendLine(await responseMessage.Content.ReadAsStringAsync());
```

**After (fixed):**
```csharp
// Buffer content once
contentString = await responseMessage.Content.ReadAsStringAsync();
using (MemoryStream contentStream = new MemoryStream(Encoding.UTF8.GetBytes(contentString)))
{
    Error error = JsonSerializable.LoadFrom<Error>(stream: contentStream);
    // ... 
}
// ... if this fails and throws, execution continues to:

// Reuse buffered content - no stream re-reading
contextBuilder.AppendLine(contentString ?? await responseMessage.Content.ReadAsStringAsync());
```

This ensures that when Gateway returns invalid JSON during failover
scenarios, clients get proper `DocumentClientException` instances with
diagnostics instead of unhandled `InvalidOperationException` errors.

Fixes #5243.

<!-- START COPILOT CODING AGENT TIPS -->
---

💡 You can make Copilot smarter by setting up custom instructions,
customizing its development environment and configuring Model Context
Protocol (MCP) servers. Learn more [Copilot coding agent
tips](https://gh.io/copilot-coding-agent-tips) in the docs.

---------

Co-authored-by: copilot-swe-agent[bot] <[email protected]>
Co-authored-by: kirankumarkolli <[email protected]>
Co-authored-by: Kiran Kumar Kolli <[email protected]>

Author: Copilot

Microsoft.Azure.Cosmos/src/GatewayStoreClient.cs

@@ -168,31 +168,37 @@ internal static async Task<DocumentClientException> CreateDocumentClientExceptio
             }
 
             // If service rejects the initial payload like header is to large it will return an HTML error instead of JSON.
+            string contentString = null;
             if (string.Equals(responseMessage.Content?.Headers?.ContentType?.MediaType, "application/json", StringComparison.OrdinalIgnoreCase) &&
                 responseMessage.Content?.Headers.ContentLength > 0)
             {
                 try
                 {
-                    Stream contentAsStream = await responseMessage.Content.ReadAsStreamAsync();
-                    Error error = JsonSerializable.LoadFrom<Error>(stream: contentAsStream);
-
-                    return new DocumentClientException(
-                        errorResource: error,
-                        responseHeaders: responseMessage.Headers,
-                        statusCode: responseMessage.StatusCode)
+                    // Buffer the content once to avoid "stream already consumed" issue
+                    contentString = await responseMessage.Content.ReadAsStringAsync();
+                    using (MemoryStream contentStream = new MemoryStream(Encoding.UTF8.GetBytes(contentString)))
                     {
-                        StatusDescription = responseMessage.ReasonPhrase,
-                        ResourceAddress = resourceIdOrFullName,
-                        RequestStatistics = requestStatistics
-                    };
+                        Error error = JsonSerializable.LoadFrom<Error>(stream: contentStream);
+
+                        return new DocumentClientException(
+                            errorResource: error,
+                            responseHeaders: responseMessage.Headers,
+                            statusCode: responseMessage.StatusCode)
+                        {
+                            StatusDescription = responseMessage.ReasonPhrase,
+                            ResourceAddress = resourceIdOrFullName,
+                            RequestStatistics = requestStatistics
+                        };
+                    }
                 }
                 catch
                 {
                 }
             }
 
             StringBuilder contextBuilder = new StringBuilder();
-            contextBuilder.AppendLine(await responseMessage.Content.ReadAsStringAsync());
+            // Reuse the already buffered content if available, otherwise read it now
+            contextBuilder.AppendLine(contentString ?? await responseMessage.Content.ReadAsStringAsync());
 
             HttpRequestMessage requestMessage = responseMessage.RequestMessage;
 

Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.Tests/GatewayStoreClientTests.cs

@@ -246,6 +246,38 @@ public async Task TestCreateDocumentClientExceptionWhenMediaTypeIsApplicationJso
             Assert.IsNotNull(value: documentClientException.Error.Message);
         }
 
+        /// <summary>
+        /// Test to verify the fix for the stream consumption issue when JSON deserialization fails.
+        /// This reproduces the scenario where a 403 response has application/json content type 
+        /// but invalid JSON content, which would previously cause "stream already consumed" exception.
+        /// Fixes issue #5243.
+        /// </summary>
+        [TestMethod]
+        [Owner("copilot")]
+        public async Task TestStreamConsumptionBugFixWhenJsonDeserializationFails()
+        {
+            // Create invalid JSON content that will fail deserialization but has application/json content type
+            string invalidJson = "{ \"error\": invalid json content that will fail parsing }";
+            
+            HttpResponseMessage responseMessage = new HttpResponseMessage(HttpStatusCode.Forbidden)
+            {
+                RequestMessage = new HttpRequestMessage(HttpMethod.Get, "https://test.com/dbs/db1/colls/coll1/docs/doc1"),
+                Content = new StringContent(invalidJson, Encoding.UTF8, "application/json")
+            };
+
+            IClientSideRequestStatistics requestStatistics = GatewayStoreClientTests.CreateClientSideRequestStatistics();
+
+            // This should NOT throw an InvalidOperationException about stream being consumed
+            DocumentClientException exception = await GatewayStoreClient.CreateDocumentClientExceptionAsync(
+                responseMessage: responseMessage,
+                requestStatistics: requestStatistics);
+
+            // Verify the exception was created successfully with fallback logic
+            Assert.IsNotNull(exception);
+            Assert.AreEqual(HttpStatusCode.Forbidden, exception.StatusCode);
+            Assert.IsTrue(exception.Message.Contains(invalidJson), "Exception message should contain the original invalid JSON content");
+        }
+
         private static IClientSideRequestStatistics CreateClientSideRequestStatistics()
         {
             return new ClientSideRequestStatisticsTraceDatum(