Fix Netty buffer leak in ReactorNettyClient response handling

Copilot · FabianMeiswinkel · Copilot · commit ff57b2a07624 · 2025-11-14T23:19:34.000Z
Co-authored-by: FabianMeiswinkel &lt;19165014+FabianMeiswinkel@users.noreply.github.com&gt;
diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java
@@ -357,8 +357,8 @@ public Mono<ByteBuf> body() {
                             if (leakDetectionDebuggingEnabled && buf.refCnt() > 0) {
                                 buf.touch("ReactorNettyHttpResponse.body - onDiscard - refCnt: " + buf.refCnt());
                                 logger.info("ReactorNettyHttpResponse.body - onDiscard - refCnt: {}", buf.refCnt());
-                                ReferenceCountUtil.safeRelease(buf);
                             }
+                            ReferenceCountUtil.safeRelease(buf);
                         })
                 )
                 .aggregate()
@@ -417,14 +417,12 @@ private void releaseOnNotSubscribedResponse(ReactorNettyResponseState reactorNet
                 }
 
                 body()
-                    .map(buf -> {
+                    .doOnNext(buf -> {
                         if (leakDetectionDebuggingEnabled && buf.refCnt() > 0) {
                             buf.touch("ReactorNettyHttpResponse.releaseOnNotSubscribedResponse - refCnt: " + buf.refCnt());
                             logger.info("ReactorNettyHttpResponse.releaseOnNotSubscribedResponse - refCnt: {}", buf.refCnt());
-                            ReferenceCountUtil.safeRelease(buf);
                         }
-
-                        return buf;
+                        ReferenceCountUtil.safeRelease(buf);
                     })
                     .subscribe(v -> {}, ex -> {}, () -> {});
             }

Original file line number	Diff line number	Diff line change
`@@ -357,8 +357,8 @@ public Mono<ByteBuf> body() {`
`357`	`357`	`if (leakDetectionDebuggingEnabled && buf.refCnt() > 0) {`
`358`	`358`	`buf.touch("ReactorNettyHttpResponse.body - onDiscard - refCnt: " + buf.refCnt());`
`359`	`359`	`logger.info("ReactorNettyHttpResponse.body - onDiscard - refCnt: {}", buf.refCnt());`
`360`		`- ReferenceCountUtil.safeRelease(buf);`
`361`	`360`	`}`
	`361`	`+ ReferenceCountUtil.safeRelease(buf);`
`362`	`362`	`})`
`363`	`363`	`)`
`364`	`364`	`.aggregate()`
`@@ -417,14 +417,12 @@ private void releaseOnNotSubscribedResponse(ReactorNettyResponseState reactorNet`
`417`	`417`	`}`
`418`	`418`
`419`	`419`	`body()`
`420`		`- .map(buf -> {`
	`420`	`+ .doOnNext(buf -> {`
`421`	`421`	`if (leakDetectionDebuggingEnabled && buf.refCnt() > 0) {`
`422`	`422`	`buf.touch("ReactorNettyHttpResponse.releaseOnNotSubscribedResponse - refCnt: " + buf.refCnt());`
`423`	`423`	`logger.info("ReactorNettyHttpResponse.releaseOnNotSubscribedResponse - refCnt: {}", buf.refCnt());`
`424`		`- ReferenceCountUtil.safeRelease(buf);`
`425`	`424`	`}`
`426`		`-`
`427`		`- return buf;`
	`425`	`+ ReferenceCountUtil.safeRelease(buf);`
`428`	`426`	`})`
`429`	`427`	`.subscribe(v -> {}, ex -> {}, () -> {});`
`430`	`428`	`}`