[Internal] Binary Encoding: Adds Client Option to Enable/Disable Request and Response Stream Conversation (#5123)

# Pull Request Template

## Description

Create a new internal cosmos client options: `EnableStreamPassThrough`
to enable or disable the request/ response stream conversation. When
enabled, the request and/or response stream would not be converted to
the desired target serialization type and would act just like a
pass-through. The default value for this parameter is `False`.

**Usage:**

```
      CosmosClientOptions clientOptions = new CosmosClientOptions()
      {
          EnableStreamPassThrough = true,
      };

      CosmosClient cosmosClient = new (
          endpoint,
          authKey,
          clientOptions);
```

## Type of change

Please delete options that are not relevant.

- [x] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)

## Closing issues

To automatically close an issue: closes #5122

Author: kundadebdatta

Microsoft.Azure.Cosmos/src/CosmosClientOptions.cs

@@ -452,6 +452,14 @@ public System.Text.Json.JsonSerializerOptions UseSystemTextJsonSerializerWithOpt
         /// The default value is 'true'.
         /// </summary>
         internal bool EnableAsyncCacheExceptionNoSharing { get; set; } = true;
+
+        /// <summary>
+        /// Gets or sets the boolean flag to skip converting a text stream to binary and vice versa. When enabled, the request and response stream
+        /// would not be converted to the desired target serialization type and will act just like a pass through. This client option will
+        /// remain internal only since the consumer of this flag will be the internal components of the cosmos db ecosystem.
+        /// The default value for this parameter is 'false'.
+        /// </summary>
+        internal bool EnableStreamPassThrough { get; set; } = false;
 
         /// <summary>
         /// (Direct/TCP) Controls the amount of idle time after which unused connections are closed.

Microsoft.Azure.Cosmos/src/Resource/Container/ContainerCore.Items.cs

@@ -934,7 +934,8 @@ private async Task<ResponseMessage> ProcessItemStreamAsync(
             // engine, which does not support binary encoded content at the moment. For long term, since trigger operations won't
             // be supported in the backend, avoiding the binary encoding in such cases, will be the ideal approach.
             if (ConfigurationManager.IsBinaryEncodingEnabled()
-                && !ContainerCore.IsTriggerPresentInRequestOptions(requestOptions))
+                && !ContainerCore.IsTriggerPresentInRequestOptions(requestOptions)
+                && !this.ClientContext.ClientOptions.EnableStreamPassThrough)
             {
                 streamPayload = CosmosSerializationUtil.TrySerializeStreamToTargetFormat(
                     targetSerializationFormat: ContainerCore.GetTargetRequestSerializationFormat(),
@@ -957,7 +958,8 @@ private async Task<ResponseMessage> ProcessItemStreamAsync(
                 cancellationToken: cancellationToken);
 
             // Convert Binary Stream to Text.
-            if (targetResponseSerializationFormat.HasValue
+            if (!this.ClientContext.ClientOptions.EnableStreamPassThrough
+                && targetResponseSerializationFormat.HasValue
                 && (requestOptions == null || !requestOptions.EnableBinaryResponseOnPointOperations)
                 && responseMessage?.Content is CloneableStream outputCloneableStream)
             {

Microsoft.Azure.Cosmos/tests/Microsoft.Azure.Cosmos.EmulatorTests/CosmosItemTests.cs

@@ -768,6 +768,138 @@ public async Task CreateDropItemStreamTest(bool binaryEncodingEnabledInClient, b
             {
                 Environment.SetEnvironmentVariable(ConfigurationManager.BinaryEncodingEnabled, null);
             }
+        }
+
+        [TestMethod]
+        [Owner("dkunda")]
+        [DataRow(true, true, DisplayName = "Test scenario when binary encoding is enabled at client level and stream conversation for binary encoding is skipped.")]
+        [DataRow(true, false, DisplayName = "Test scenario when binary encoding is enabled at client level and stream conversation for binary encoding is enabled.")]
+        [DataRow(false, true, DisplayName = "Test scenario when binary encoding is disabled at client level and stream conversation for binary encoding is skipped.")]
+        [DataRow(false, false, DisplayName = "Test scenario when binary encoding is disabled at client level and stream conversation for binary encoding is enabled.")]
+        public async Task CreateItemStream_WithEnableBinaryResponseOptions_ShouldSkipStreamConversation(
+            bool binaryEncodingEnabledInClient,
+            bool enableStreamPassThrough)
+        {
+            Cosmos.Database database = null;
+            Container container = null;
+            try
+            {
+                string databaseName = "binary-encoding-db";
+                string containerName = "binary-encoding-container";
+                if (binaryEncodingEnabledInClient)
+                {
+                    Environment.SetEnvironmentVariable(ConfigurationManager.BinaryEncodingEnabled, "True");
+                }
+
+                (string endpoint, string authKey) = TestCommon.GetAccountInfo();
+                CosmosClientOptions clientOptions = new CosmosClientOptions()
+                {
+                    EnableStreamPassThrough = enableStreamPassThrough,
+                };
+
+                CosmosClient cosmosClient = new (
+                    endpoint,
+                    authKey,
+                    clientOptions);
+
+                DatabaseResponse dbResponse = await cosmosClient.CreateDatabaseIfNotExistsAsync(databaseName);
+                database = dbResponse.Database;
+
+                ContainerProperties properties = new (id: containerName, partitionKeyPath: "/pk");
+                container = await database.CreateContainerIfNotExistsAsync(properties);
+
+                ToDoActivity testItem = ToDoActivity.CreateRandomToDoActivity();
+                CosmosSerializerCore cosmosSerializer = new CosmosSerializerCore();
+                using (Stream stream = cosmosSerializer.ToStream<ToDoActivity>(
+                    testItem,
+                    canUseBinaryEncodingForPointOperations: binaryEncodingEnabledInClient))
+                {
+                    if (binaryEncodingEnabledInClient)
+                    {
+                        // Asserting the input stream is in binary format.
+                        AssertOnResponseSerializationBinaryType(stream);
+                    }
+                    else
+                    {
+                        // Asserting the input stream is in text format.
+                        AssertOnResponseSerializationTextType(stream);
+                    }
+
+                    using (ResponseMessage response = await container.CreateItemStreamAsync(
+                        streamPayload: stream,
+                        partitionKey: new Cosmos.PartitionKey(testItem.pk)))
+                    {
+                        Assert.IsNotNull(response);
+                        Assert.AreEqual(HttpStatusCode.Created, response.StatusCode);
+                        Assert.IsTrue(response.Headers.RequestCharge > 0);
+                        Assert.IsNotNull(response.Headers.ActivityId);
+                        Assert.IsNotNull(response.Headers.ETag);
+                        Assert.IsNotNull(response.Diagnostics);
+                        Assert.IsTrue(!string.IsNullOrEmpty(response.Diagnostics.ToString()));
+                        Assert.IsTrue(response.Diagnostics.GetClientElapsedTime() > TimeSpan.Zero);
+
+                        if (!enableStreamPassThrough)
+                        {
+                            AssertOnResponseSerializationTextType(response.Content);
+                        }
+                        else
+                        {
+                            if (binaryEncodingEnabledInClient)
+                            {
+                                AssertOnResponseSerializationBinaryType(response.Content);
+                            }
+                            else
+                            {
+                                AssertOnResponseSerializationTextType(response.Content);
+                            }
+                        }
+                    }
+                }
+
+                using (ResponseMessage response = await container.ReadItemStreamAsync(
+                    id: testItem.id,
+                    partitionKey: new Cosmos.PartitionKey(testItem.pk)))
+                {
+                    Assert.IsNotNull(response);
+                    Assert.AreEqual(HttpStatusCode.OK, response.StatusCode);
+                    Assert.IsTrue(response.Headers.RequestCharge > 0);
+                    Assert.IsNotNull(response.Headers.ActivityId);
+                    Assert.IsNotNull(response.Headers.ETag);
+                    Assert.IsNotNull(response.Diagnostics);
+                    Assert.IsTrue(!string.IsNullOrEmpty(response.Diagnostics.ToString()));
+                    Assert.IsTrue(response.Diagnostics.GetClientElapsedTime() > TimeSpan.Zero);
+
+                    if (!enableStreamPassThrough)
+                    {
+                        AssertOnResponseSerializationTextType(response.Content);
+                    }
+                    else
+                    {
+                        if (binaryEncodingEnabledInClient)
+                        {
+                            AssertOnResponseSerializationBinaryType(response.Content);
+                        }
+                        else
+                        {
+                            AssertOnResponseSerializationTextType(response.Content);
+                        }
+                    }
+                }
+            }
+            finally
+            {
+                Environment.SetEnvironmentVariable(ConfigurationManager.BinaryEncodingEnabled, null);
+
+                if (container != null)
+                {
+                    await container.DeleteContainerStreamAsync();
+                }
+
+                if (database != null)
+                {
+                    await database.DeleteAsync();
+                }
+            }
         }
 
         [TestMethod]