Fix netty ByteBuf leak (#1822)

* Fix netty ByteBuf leak

* Consume responses everywhere

Author: trask

agent/agent-profiler/agent-service-profiler/src/main/java/com/microsoft/applicationinsights/serviceprofilerapi/client/ProfilerFrontendClientV2.java

@@ -85,6 +85,9 @@ public Mono<BlobAccessPass> getUploadAccess(UUID profileId) {
                 // this shouldn't happen, the mono should complete with a response or a failure
                 throw new AssertionError("http response mono returned empty");
               }
+              // response body is not consumed below
+              consumeResponseBody(response);
+
               if (response.getStatusCode() >= 300) {
                 throw new HttpResponseException(response);
               }
@@ -97,6 +100,13 @@ public Mono<BlobAccessPass> getUploadAccess(UUID profileId) {
             });
   }
 
+  // need to consume response, otherwise get netty ByteBuf leak warnings:
+  // io.netty.util.ResourceLeakDetector - LEAK: ByteBuf.release() was not called before
+  // it's garbage-collected (see https://github.com/Azure/azure-sdk-for-java/issues/10467)
+  private static void consumeResponseBody(HttpResponse response) {
+    response.getBody().subscribe();
+  }
+
   public Mono<HttpResponse> executePostWithRedirect(URL requestUrl) {
 
     HttpRequest request = new HttpRequest(HttpMethod.POST, requestUrl);
@@ -124,6 +134,7 @@ public Mono<ArtifactAcceptedResponse> reportUploadFinish(UUID profileId, String
               int statusCode = response.getStatusCode();
               if (statusCode != 201 && statusCode != 202) {
                 LOGGER.error("Trace upload failed: {}", statusCode);
+                consumeResponseBody(response);
                 return Mono.error(new AssertionError("http request failed"));
               }
 
@@ -168,9 +179,7 @@ public Mono<String> getSettings(Date oldTimeStamp) {
                 return Mono.error(new AssertionError("http response mono returned empty"));
               }
               if (response.getStatusCode() >= 300) {
-                // FIXME (trask) does azure http client throw HttpResponseException already on >=
-                // 300 response
-                // above?
+                consumeResponseBody(response);
                 return Mono.error(new HttpResponseException(response));
               }
               return response.getBodyAsString();

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/httpclient/LazyHttpClient.java

@@ -147,6 +147,13 @@ public Mono<HttpResponse> send(HttpRequest request, Context context) {
     return getDelegate().send(request, context);
   }
 
+  // need to consume response, otherwise get netty ByteBuf leak warnings:
+  // io.netty.util.ResourceLeakDetector - LEAK: ByteBuf.release() was not called before
+  // it's garbage-collected (see https://github.com/Azure/azure-sdk-for-java/issues/10467)
+  public static void consumeResponseBody(HttpResponse response) {
+    response.getBody().subscribe();
+  }
+
   private static HttpPipelinePolicy getAuthenticationPolicy(
       Configuration.AadAuthentication configuration) {
     switch (configuration.type) {

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/quickpulse/QuickPulseDataSender.java

@@ -24,6 +24,7 @@
 import com.azure.core.http.HttpPipeline;
 import com.azure.core.http.HttpRequest;
 import com.azure.core.http.HttpResponse;
+import com.microsoft.applicationinsights.agent.internal.httpclient.LazyHttpClient;
 import java.util.concurrent.ArrayBlockingQueue;
 
 class QuickPulseDataSender implements Runnable {
@@ -57,7 +58,14 @@ public void run() {
 
       long sendTime = System.nanoTime();
       try (HttpResponse response = httpPipeline.send(post).block()) {
-        if (response != null && networkHelper.isSuccess(response)) {
+        if (response == null) {
+          // this shouldn't happen, the mono should complete with a response or a failure
+          throw new AssertionError("http response mono returned empty");
+        }
+        // response body is not consumed below
+        LazyHttpClient.consumeResponseBody(response);
+
+        if (networkHelper.isSuccess(response)) {
           QuickPulseHeaderInfo quickPulseHeaderInfo =
               networkHelper.getQuickPulseHeaderInfo(response);
           switch (quickPulseHeaderInfo.getQuickPulseStatus()) {

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/quickpulse/QuickPulsePingSender.java

@@ -27,6 +27,7 @@
 import com.microsoft.applicationinsights.agent.internal.common.ExceptionStats;
 import com.microsoft.applicationinsights.agent.internal.common.ExceptionUtils;
 import com.microsoft.applicationinsights.agent.internal.common.Strings;
+import com.microsoft.applicationinsights.agent.internal.httpclient.LazyHttpClient;
 import com.microsoft.applicationinsights.agent.internal.telemetry.TelemetryClient;
 import java.util.Date;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -91,11 +92,15 @@ public QuickPulseHeaderInfo ping(String redirectedEndpoint) {
     request.setBody(buildPingEntity(currentDate.getTime()));
 
     long sendTime = System.nanoTime();
-    HttpResponse response = null;
-    try {
+    try (HttpResponse response = httpPipeline.send(request).block()) {
+      if (response == null) {
+        // this shouldn't happen, the mono should complete with a response or a failure
+        throw new AssertionError("http response mono returned empty");
+      }
+      // response body is not consumed below
+      LazyHttpClient.consumeResponseBody(response);
 
-      response = httpPipeline.send(request).block();
-      if (response != null && networkHelper.isSuccess(response)) {
+      if (networkHelper.isSuccess(response)) {
         QuickPulseHeaderInfo quickPulseHeaderInfo = networkHelper.getQuickPulseHeaderInfo(response);
         switch (quickPulseHeaderInfo.getQuickPulseStatus()) {
           case QP_IS_OFF:
@@ -111,10 +116,6 @@ public QuickPulseHeaderInfo ping(String redirectedEndpoint) {
     } catch (Throwable t) {
       exceptionStats.recordFailure(t.getMessage(), t);
       ExceptionUtils.parseError(t, getQuickPulseEndpoint(), friendlyExceptionThrown, logger);
-    } finally {
-      if (response != null) {
-        response.close();
-      }
     }
     return onPingError(sendTime);
   }

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/telemetry/TelemetryChannel.java

@@ -202,6 +202,7 @@ private CompletableResultCode internalSend(
         .subscribe(
             response -> {
               parseResponseCode(response.getStatusCode(), byteBuffers, byteBuffers);
+              LazyHttpClient.consumeResponseBody(response);
             },
             error -> {
               StatsbeatModule.get().getNetworkStatsbeat().incrementRequestFailureCount();