diff --git a/core/src/main/java/feign/AsynchronousMethodHandler.java b/core/src/main/java/feign/AsynchronousMethodHandler.java index 4ffc663c6..87873f883 100644 --- a/core/src/main/java/feign/AsynchronousMethodHandler.java +++ b/core/src/main/java/feign/AsynchronousMethodHandler.java @@ -29,6 +29,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.TimeUnit; + import java.util.function.BiConsumer; import java.util.stream.Stream; @@ -114,18 +115,31 @@ private CompletableFuture executeAndDecode( } private static class CancellableFuture extends CompletableFuture { - private CompletableFuture inner = null; - + // volatile provides the same JMM happens-before guarantees as AtomicReference + // since we only ever read/write (never CAS), with less indirection. + private volatile CompletableFuture inner; + + /** + * Registers {@code value} as the active inner future and pipes its result into this future. + * + *

Side-effect: if this future has already been cancelled before {@code setInner} is called, + * the cancellation is immediately forwarded to {@code value} so that the in-flight async work + * is also cancelled rather than completing silently. + */ public void setInner(CompletableFuture value) { inner = value; - inner.whenComplete(pipeTo(this)); + value.whenComplete(pipeTo(this)); + if (isCancelled()) { + value.cancel(true); + } } @Override public boolean cancel(boolean mayInterruptIfRunning) { final boolean result = super.cancel(mayInterruptIfRunning); - if (inner != null) { - inner.cancel(mayInterruptIfRunning); + CompletableFuture current = inner; + if (current != null) { + current.cancel(mayInterruptIfRunning); } return result; } diff --git a/core/src/main/java/feign/Request.java b/core/src/main/java/feign/Request.java index a3627a164..235700df6 100644 --- a/core/src/main/java/feign/Request.java +++ b/core/src/main/java/feign/Request.java @@ -26,7 +26,6 @@ import java.util.Arrays; import java.util.Collection; import java.util.Collections; -import java.util.HashMap; import java.util.Map; import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; @@ -327,6 +326,15 @@ public static class Options { private final boolean followRedirects; private final Map> threadToMethodOptions; + /** + * Returns the identifier used to bucket method-level options by calling context. Defaults to + * the current thread's identity. Subclasses may override this to provide a fixed identifier, + * which is useful in tests to force concurrent threads to contend on the same outer map key. + */ + protected String threadIdentifier() { + return getThreadIdentifier(); + } + /** * Get an Options by methodName * @@ -335,9 +343,12 @@ public static class Options { */ @Experimental public Options getMethodOptions(String methodName) { - Map methodOptions = - threadToMethodOptions.getOrDefault(getThreadIdentifier(), new HashMap<>()); - return methodOptions.getOrDefault(methodName, this); + Map methodOptions = threadToMethodOptions.get(threadIdentifier()); + if (methodOptions == null) { + return this; + } + Options options = methodOptions.get(methodName); + return options != null ? options : this; } /** @@ -348,11 +359,9 @@ public Options getMethodOptions(String methodName) { */ @Experimental public void setMethodOptions(String methodName, Options options) { - String threadIdentifier = getThreadIdentifier(); - Map methodOptions = - threadToMethodOptions.getOrDefault(threadIdentifier, new HashMap<>()); - threadToMethodOptions.put(threadIdentifier, methodOptions); - methodOptions.put(methodName, options); + threadToMethodOptions + .computeIfAbsent(threadIdentifier(), key -> new ConcurrentHashMap<>()) + .put(methodName, options); } /** @@ -517,10 +526,10 @@ public static class Body implements Serializable { private transient Charset encoding; - private byte[] data; + private final byte[] data; private Body() { - super(); + this(null); } private Body(byte[] data) { diff --git a/core/src/test/java/feign/CancellableFutureTest.java b/core/src/test/java/feign/CancellableFutureTest.java new file mode 100644 index 000000000..9738d554b --- /dev/null +++ b/core/src/test/java/feign/CancellableFutureTest.java @@ -0,0 +1,148 @@ +/* + * Copyright © 2012 The Feign Authors (feign@commonhaus.dev) + * + * 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 feign; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.util.Collections; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; +import org.junit.jupiter.api.Test; + +class CancellableFutureTest { + + interface Api { + @RequestLine("GET /") + CompletableFuture get(); + } + + /** + * cancel() arrives BEFORE setInner() is called. + * + *

The AsyncClient returns immediately with a pending CompletableFuture so that api.get() + * returns the CancellableFuture to the caller without blocking. The caller then cancels it before + * the client future is completed. When the client future eventually completes, setInner() must + * detect isCancelled() and immediately forward cancellation to the newly registered inner future. + */ + @Test + void cancelBeforeSetInnerRacesCorrectly() throws Exception { + // execute() returns this immediately — no blocking inside execute() + CompletableFuture clientFuture = new CompletableFuture<>(); + + AsyncClient client = (request, options, ctx) -> clientFuture; + + Api api = AsyncFeign.builder().client(client).target(Api.class, "http://localhost:0"); + + // api.get() returns immediately because execute() returns immediately + CompletableFuture result = api.get(); + + // Cancel BEFORE clientFuture resolves — inner is not yet set on CancellableFuture + result.cancel(true); + + // Complete the client future now. This triggers the whenComplete → setInner() path. + // setInner() must see isCancelled() == true and cancel the newly registered inner future. + clientFuture.complete( + Response.builder() + .status(200) + .reason("OK") + .request( + Request.create( + Request.HttpMethod.GET, + "http://localhost:0", + Collections.emptyMap(), + Request.Body.empty(), + null)) + .build()); + + assertThat(result).isCancelled(); + } + + /** + * cancel() arrives AFTER setInner() has already been called (the retry path). + * + *

The first execute() fails immediately to trigger a retry. The retry execute() returns a + * pending CompletableFuture immediately (no blocking inside execute()) and signals a latch so + * the caller knows setInner() has been called. The caller then cancels — cancel() must read + * inner and propagate to the retry future. + */ + @Test + void cancelAfterSetInnerRacesCorrectly() throws Exception { + AtomicInteger callCount = new AtomicInteger(); + CountDownLatch retryStarted = new CountDownLatch(1); + // Holds the raw client future from the retry execute() call + CompletableFuture[] retryFutureHolder = new CompletableFuture[1]; + + AsyncClient client = + (request, options, ctx) -> { + int n = callCount.incrementAndGet(); + if (n == 1) { + // First call: fail immediately to trigger the retryer + CompletableFuture failed = new CompletableFuture<>(); + failed.completeExceptionally(new IOException("transient")); + return failed; + } + // Retry call: return a pending future immediately — execute() does NOT block. + // The latch is used only to tell the caller that setInner() has been called. + CompletableFuture retryFuture = new CompletableFuture<>(); + retryFutureHolder[0] = retryFuture; + retryStarted.countDown(); + return retryFuture; + }; + + Api api = + AsyncFeign.builder() + .client(client) + .retryer(new Retryer.Default(0, 0, 2)) + .target(Api.class, "http://localhost:0"); + + CompletableFuture result = api.get(); + + // Wait until the retry execute() returned and setInner() has been called + assertThat(retryStarted.await(2, TimeUnit.SECONDS)).isTrue(); + + // cancel() now arrives after setInner() — inner is already set to retryFuture + result.cancel(true); + + assertThat(result).isCancelled(); + + // Verify the pipeTo guard: even if the raw retry client future eventually completes, + // it must NOT overwrite the cancellation on result. setInner() registered a whenComplete + // that calls pipeTo(result), which checks isDone() before completing — so result must + // remain cancelled after the raw client future resolves. + CompletableFuture retryFuture = retryFutureHolder[0]; + assertThat(retryFuture).isNotNull(); + retryFuture.cancel(false); // let the retry future give up + assertThat(result).isCancelled(); // must still be cancelled, not overwritten + } + + /** Normal completion (no cancellation) must not be disrupted by the volatile field change. */ + @Test + void normalCompletionIsNotAffected() throws Exception { + MockWebServer server = new MockWebServer(); + server.enqueue(new MockResponse().setBody("hello")); + + Api api = AsyncFeign.builder().target(Api.class, server.url("/").toString()); + + assertThat(api.get().get(2, TimeUnit.SECONDS)).isEqualTo("hello"); + server.shutdown(); + } +} + diff --git a/core/src/test/java/feign/OptionsTest.java b/core/src/test/java/feign/OptionsTest.java index 904c6b255..ec41ada1e 100644 --- a/core/src/test/java/feign/OptionsTest.java +++ b/core/src/test/java/feign/OptionsTest.java @@ -38,6 +38,22 @@ public ChildOptions(int connectTimeoutMillis, int readTimeoutMillis) { } } + /** + * Options subclass that overrides threadIdentifier() to return a fixed constant, forcing all + * threads to contend on the same outer key in threadToMethodOptions. This is the only way to + * exercise the pre-fix check-then-act race without changing thread identity. + */ + static class SharedKeyOptions extends Request.Options { + public SharedKeyOptions(int connectTimeoutMillis, int readTimeoutMillis) { + super(connectTimeoutMillis, readTimeoutMillis); + } + + @Override + protected String threadIdentifier() { + return "shared-key"; + } + } + interface OptionsInterface { @RequestLine("GET /") String get(Request.Options options); @@ -135,4 +151,50 @@ void normalResponseWithMethodOptionsTest() throws Exception { thread.start(); thread.join(); } + + /** + * Forces multiple threads to contend on the SAME outer key in threadToMethodOptions by using + * SharedKeyOptions, which returns a fixed "shared-key" from threadIdentifier(). + * + *

Before the fix (getOrDefault + put), two threads racing with the same key could both + * observe the key absent, both create a new inner map, and one thread's put would overwrite the + * other's — silently losing entries. With computeIfAbsent + ConcurrentHashMap, creation is + * atomic and all entries must be present after all threads complete. + */ + @Test + void concurrentSetMethodOptionsOnSameKeyDoesNotLoseEntries() throws Exception { + SharedKeyOptions options = new SharedKeyOptions(1000, 1000); + int threadCount = 20; + CountDownLatch start = new CountDownLatch(1); + CountDownLatch done = new CountDownLatch(threadCount); + AtomicReference error = new AtomicReference<>(); + + for (int i = 0; i < threadCount; i++) { + final String method = "method" + i; + new Thread( + () -> { + try { + start.await(); + options.setMethodOptions(method, new Request.Options(1000, 2000)); + } catch (Throwable t) { + error.set(t); + } finally { + done.countDown(); + } + }) + .start(); + } + + start.countDown(); + assertThat(done.await(5, TimeUnit.SECONDS)).isTrue(); + + // No exception must have been thrown + assertThat(error.get()).isNull(); + // All 20 entries must be present — proves no lost updates due to the race + for (int i = 0; i < threadCount; i++) { + assertThat(options.getMethodOptions("method" + i)) + .as("entry for method%d must not have been lost", i) + .isNotSameAs(options); + } + } }