Fix Netty ByteBuf leaks in StoreResponse and RetryContextOnDiagnosticTest (#47266)

* Initial plan

* Improve logging for ByteBufInputStream close failures

Change log level from debug to warn and catch Throwable instead of just IOException to make potential ByteBuf leak issues more visible.

Co-authored-by: FabianMeiswinkel <19165014+FabianMeiswinkel@users.noreply.github.com>

* Fix ByteBuf leak in RetryContextOnDiagnosticTest

Changed from Mono.just() to Mono.fromCallable() to defer StoreResponse creation, ensuring ByteBuf lifecycle is properly managed within each subscription rather than eagerly at mock setup time.

Co-authored-by: FabianMeiswinkel <19165014+FabianMeiswinkel@users.noreply.github.com>

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: FabianMeiswinkel <19165014+FabianMeiswinkel@users.noreply.github.com>

Author: Copilot

sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/RetryContextOnDiagnosticTest.java

@@ -105,15 +105,12 @@ public void backoffRetryUtilityExecuteRetry() throws Exception {
         addressSelector = Mockito.mock(AddressSelector.class);
         CosmosException exception = new CosmosException(410, exceptionText);
         String rawJson = "{\"id\":\"" + responseText + "\"}";
-        ByteBuf buffer = getUTF8BytesOrNull(rawJson);
         Mockito.when(callbackMethod.call()).thenThrow(exception, exception, exception, exception, exception)
 
-            .thenReturn(Mono.just(new StoreResponse(
-                null,
-                200,
-                new HashMap<>(),
-                new ByteBufInputStream(buffer, true),
-                buffer.readableBytes())));
+            .thenReturn(Mono.fromCallable(() -> StoreResponseBuilder.create()
+                .withContent(rawJson)
+                .withStatus(200)
+                .build()));
         Mono<StoreResponse> monoResponse = BackoffRetryUtility.executeRetry(callbackMethod, retryPolicy);
         StoreResponse response = validateSuccess(monoResponse);
 
@@ -156,14 +153,11 @@ public void backoffRetryUtilityExecuteAsync() {
         CosmosException exception = new CosmosException(410, exceptionText);
         Mono<StoreResponse> exceptionMono = Mono.error(exception);
         String rawJson = "{\"id\":\"" + responseText + "\"}";
-        ByteBuf buffer = getUTF8BytesOrNull(rawJson);
         Mockito.when(parameterizedCallbackMethod.apply(ArgumentMatchers.any())).thenReturn(exceptionMono, exceptionMono, exceptionMono, exceptionMono, exceptionMono)
-            .thenReturn(Mono.just(new StoreResponse(
-                null,
-                200,
-                new HashMap<>(),
-                new ByteBufInputStream(buffer, true),
-                buffer.readableBytes())));
+            .thenReturn(Mono.fromCallable(() -> StoreResponseBuilder.create()
+                .withContent(rawJson)
+                .withStatus(200)
+                .build()));
         Mono<StoreResponse> monoResponse = BackoffRetryUtility.executeAsync(
             parameterizedCallbackMethod,
             retryPolicy,

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/StoreResponse.java

@@ -71,12 +71,12 @@ public StoreResponse(
         if (contentStream != null) {
             try {
                 this.responsePayload = new JsonNodeStorePayload(contentStream, responsePayloadLength, headerMap);
-            }
-            finally {
+            } finally {
                 try {
                     contentStream.close();
-                } catch (IOException e) {
-                    logger.debug("Could not successfully close content stream.", e);
+                } catch (Throwable e) {
+                    // Log as warning instead of debug to make ByteBuf leak issues more visible
+                    logger.warn("Failed to close content stream. This may cause a Netty ByteBuf leak.", e);
                 }
             }
         } else {