fix #1385: Fix a response leak when requests are canceled (#1415)

Fix a response leak when requests are canceled. This is a workaround for an upstream bug in okhttp.

changelog/@unreleased/pr-1415.v2.yml

@@ -0,0 +1,6 @@
+type: fix
+fix:
+  description: Fix a response leak when requests are canceled. This is a workaround
+    for an upstream bug in okhttp.
+  links:
+  - https://github.com/palantir/conjure-java-runtime/pull/1415

okhttp-clients/src/main/java/com/palantir/conjure/java/okhttp/OkHttpClients.java

@@ -212,6 +212,9 @@ private static RemotingOkHttpClient createInternal(
 
         client.addInterceptor(CatchThrowableInterceptor.INSTANCE);
         client.addInterceptor(SpanTerminatingInterceptor.INSTANCE);
+        // Order is important, this interceptor must be applied prior to ConcurrencyLimitingInterceptor
+        // in order to prevent concurrency limiters from leaking.
+        client.addInterceptor(ResponseCapturingInterceptor.INSTANCE);
 
         // Routing
         UrlSelectorImpl urlSelector = UrlSelectorImpl.createWithFailedUrlCooldown(

okhttp-clients/src/main/java/com/palantir/conjure/java/okhttp/RemotingOkHttpCall.java

@@ -242,7 +242,7 @@ public void onResponse(Call call, Response response) throws IOException {
     }
 
     private void enqueueInternal(Callback callback) {
-        super.enqueue(new Callback() {
+        super.enqueue(new LeakedResponseClosingCallback(new Callback() {
             @Override
             public void onFailure(Call call, IOException exception) {
                 if (isCanceled()) {
@@ -352,7 +352,29 @@ public void onResponse(Call call, Response response) throws IOException {
                         new SafeIoException(
                                 "Failed to handle request, this is an conjure-java-runtime bug."));
             }
-        });
+        }));
+    }
+
+    private static final class LeakedResponseClosingCallback implements Callback {
+
+        private final Callback delegate;
+
+        LeakedResponseClosingCallback(Callback delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public void onFailure(Call call, IOException exception) {
+            ResponseCapturingInterceptor.getResponse().ifPresent(RemotingOkHttpCall::close);
+            ResponseCapturingInterceptor.clearThreadState();
+            delegate.onFailure(call, exception);
+        }
+
+        @Override
+        public void onResponse(Call call, Response response) throws IOException {
+            ResponseCapturingInterceptor.clearThreadState();
+            delegate.onResponse(call, response);
+        }
     }
 
     private boolean shouldRetry(IOException exception, Optional<Duration> backoff) {

okhttp-clients/src/main/java/com/palantir/conjure/java/okhttp/ResponseCapturingInterceptor.java

@@ -0,0 +1,47 @@
+/*
+ * (c) Copyright 2020 Palantir Technologies Inc. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.palantir.conjure.java.okhttp;
+
+import java.io.IOException;
+import java.util.Optional;
+import okhttp3.Interceptor;
+import okhttp3.Response;
+
+/**
+ * Workaround a bug in okhttp where cancellation in flight results in leaked responses.
+ * https://github.com/square/okhttp/blob/d28d2cec21641b61f3d34e05dd52f43a717c2d32/okhttp/src/main/java/okhttp3/RealCall.java#L210-L213
+ */
+enum ResponseCapturingInterceptor implements Interceptor {
+    INSTANCE;
+
+    private static final ThreadLocal<Response> currentResponse = new ThreadLocal<>();
+
+    @Override
+    public Response intercept(Chain chain) throws IOException {
+        Response response = chain.proceed(chain.request());
+        currentResponse.set(response);
+        return response;
+    }
+
+    static Optional<Response> getResponse() {
+        return Optional.ofNullable(currentResponse.get());
+    }
+
+    static void clearThreadState() {
+        currentResponse.remove();
+    }
+}