Fixed netty leak on DELETE operations in GATEWAY mode (#14727)

Azure · Sep 2, 2020 · 37c78d7 · 37c78d7
1 parent 73a8ec3
commit 37c78d7
Show file tree

Hide file tree

Showing 4 changed files with 7 additions and 30 deletions.
diff --git a/...osmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxGatewayStoreModel.java b/...osmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxGatewayStoreModel.java
@@ -246,17 +246,9 @@ private Mono<RxDocumentServiceResponse> toDocumentServiceResponse(Mono<HttpRespo
             HttpHeaders httpResponseHeaders = httpResponse.headers();
             int httpResponseStatus = httpResponse.statusCode();
 
-            Mono<byte[]> contentObservable;
-
-            if (request.getOperationType() == OperationType.Delete) {
-                // for delete we don't expect any body
-                contentObservable = Mono.just(EMPTY_BYTE_ARRAY);
-            } else {
-                // transforms the ByteBufFlux to Flux<String>
-                contentObservable = httpResponse
-                                        .bodyAsByteArray()
-                                        .switchIfEmpty(Mono.just(EMPTY_BYTE_ARRAY));
-            }
+            Mono<byte[]> contentObservable = httpResponse
+                .bodyAsByteArray()
+                .switchIfEmpty(Mono.just(EMPTY_BYTE_ARRAY));
 
             return contentObservable
                        .map(content -> {

diff --git a/...e-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ErrorUtils.java b/...e-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ErrorUtils.java
@@ -17,11 +17,7 @@ public class ErrorUtils {
     private static final Logger logger = LoggerFactory.getLogger(ErrorUtils.class);
 
     static Mono<String> getErrorResponseAsync(HttpResponse responseMessage, HttpRequest request) {
-        Mono<String> responseAsString = responseMessage.bodyAsString().switchIfEmpty(Mono.just(StringUtils.EMPTY));
-        if (request.httpMethod() == HttpMethod.DELETE) {
-            return Mono.just(StringUtils.EMPTY);
-        }
-        return responseAsString;
+        return responseMessage.bodyAsString().switchIfEmpty(Mono.just(StringUtils.EMPTY));
     }
 
     static void logGoneException(URI physicalAddress, String activityId) {

diff --git a/...osmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ResponseUtils.java b/...osmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/ResponseUtils.java
@@ -10,20 +10,13 @@
 import reactor.core.publisher.Mono;
 
 class ResponseUtils {
-    private static byte[] EMPTY_BYTE_ARRAY = {};
+    private static final byte[] EMPTY_BYTE_ARRAY = {};
 
     static Mono<StoreResponse> toStoreResponse(HttpResponse httpClientResponse, HttpRequest httpRequest) {
 
         HttpHeaders httpResponseHeaders = httpClientResponse.headers();
 
-        Mono<byte[]> contentObservable;
-
-        if (httpRequest.httpMethod() == HttpMethod.DELETE) {
-            // for delete we don't expect any body
-            contentObservable = Mono.just(EMPTY_BYTE_ARRAY);
-        } else {
-            contentObservable = httpClientResponse.bodyAsByteArray().switchIfEmpty(Mono.just(EMPTY_BYTE_ARRAY));
-        }
+        Mono<byte[]> contentObservable = httpClientResponse.bodyAsByteArray().switchIfEmpty(Mono.just(EMPTY_BYTE_ARRAY));
 
         return contentObservable.map(byteArrayContent -> {
             // transforms to Mono<StoreResponse>

diff --git a/...s/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java b/...s/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java
@@ -197,11 +197,7 @@ public HttpHeaders headers() {
         @Override
         public Flux<ByteBuf> body() {
             return bodyIntern()
-                .doOnSubscribe(this::updateSubscriptionState)
-                .map(byteBuf -> {
-                    byteBuf.retain();
-                    return byteBuf;
-                });
+                .doOnSubscribe(this::updateSubscriptionState);
         }
 
         @Override