fix: cancel the provider call when its reply times out

A reply that times out or is cancelled now cancels the task behind it, so the provider call is interrupted and its pool thread is handed back instead of finishing work that nobody reads. A new test times out a blocked provider on a single-thread executor and checks that the provider was interrupted and that the thread answers a fresh call afterwards. The class javadoc, the README snippet and the resource usage trade-off follow.

Author: ylcn91

scatter-gather/README.md

@@ -106,9 +106,21 @@ public interface Aggregator<R> {
 public List<PendingReply> scatter(RateRequest request, List<RateProvider> providers) {
   var pending = new ArrayList<PendingReply>();
   for (var provider : providers) {
-    var reply =
-        CompletableFuture.supplyAsync(() -> provider.quote(request), executor)
-            .orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS);
+    var reply = new CompletableFuture<RateQuote>();
+    var task = executor.submit(() -> {
+      try {
+        reply.complete(provider.quote(request));
+      } catch (RuntimeException e) {
+        reply.completeExceptionally(e);
+      }
+    });
+    // a reply that times out or is cancelled also cancels its task, interrupting the provider call
+    reply.orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS)
+        .whenComplete((quote, failure) -> {
+          if (failure != null && !task.isDone()) {
+            task.cancel(true);
+          }
+        });
     pending.add(new PendingReply(provider, reply));
   }
   return pending;
@@ -223,7 +235,7 @@ Benefits:
 Trade-offs:
 
 * **Partial results**: the caller must be able to live with an answer built from a subset of recipients, and the aggregator must handle an empty set.
-* **Resource usage**: every request occupies one thread or connection per recipient; a dropped reply may still be computed by the recipient.
+* **Resource usage**: every request occupies one thread or connection per recipient; dropping a reply interrupts the local call, but a remote recipient may still finish computing an answer nobody reads.
 * **Tuning**: the timeout is a compromise between completeness and responsiveness and usually needs measurement to get right.
 
 ## Related Java Design Patterns

scatter-gather/src/main/java/com/iluwatar/scattergather/ScatterGather.java

@@ -42,9 +42,9 @@
  *   <li><b>Scatter</b>: the same request is sent to every provider concurrently.
  *   <li><b>Gather</b>: replies are collected until each one has either arrived, failed, or exceeded
  *       the timeout. Late and failed replies are logged and dropped so a single slow provider
- *       cannot hold up the whole answer. Dropping a reply does not stop the provider: the task of a
- *       timed-out provider keeps running on its pool thread until it finishes on its own or the
- *       coordinator is closed.
+ *       cannot hold up the whole answer. Dropping a reply also cancels the provider call, so a
+ *       timed-out provider is interrupted and its pool thread is freed instead of staying busy with
+ *       work nobody will read.
  *   <li><b>Aggregate</b>: the gathered replies are reduced by an {@link Aggregator}.
  * </ol>
  *
@@ -107,9 +107,26 @@ public List<PendingReply> scatter(RateRequest request, List<RateProvider> provid
         providers.size());
     var pending = new ArrayList<PendingReply>();
     for (var provider : providers) {
-      var reply =
-          CompletableFuture.supplyAsync(() -> provider.quote(request), executor)
-              .orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS);
+      var reply = new CompletableFuture<RateQuote>();
+      var task =
+          executor.submit(
+              () -> {
+                try {
+                  reply.complete(provider.quote(request));
+                } catch (RuntimeException e) {
+                  reply.completeExceptionally(e);
+                }
+              });
+      // A reply that times out or is cancelled also cancels its task, interrupting the provider
+      // call so that the pool thread is handed back instead of finishing work nobody will read.
+      reply
+          .orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS)
+          .whenComplete(
+              (quote, failure) -> {
+                if (failure != null && !task.isDone()) {
+                  task.cancel(true);
+                }
+              });
       pending.add(new PendingReply(provider, reply));
     }
     return pending;

scatter-gather/src/test/java/com/iluwatar/scattergather/ScatterGatherTest.java

@@ -35,6 +35,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -81,6 +82,39 @@ void shouldDropProviderThatMissesTheTimeout() {
     assertEquals(List.of(new RateQuote("fast", new BigDecimal("200.00"))), quotes);
   }
 
+  @Test
+  void shouldInterruptProviderThatMissesTheTimeoutAndFreeItsThread() throws Exception {
+    var interrupted = new CountDownLatch(1);
+    var slow =
+        new RateProvider() {
+          @Override
+          public String name() {
+            return "slow";
+          }
+
+          @Override
+          public RateQuote quote(RateRequest request) {
+            try {
+              gate.await();
+            } catch (InterruptedException e) {
+              interrupted.countDown();
+              Thread.currentThread().interrupt();
+              throw new IllegalStateException("interrupted", e);
+            }
+            return new RateQuote(name(), BigDecimal.ONE);
+          }
+        };
+    var ownExecutor = Executors.newSingleThreadExecutor();
+    try (var subject = new ScatterGather(ownExecutor, TIMEOUT)) {
+      assertTrue(subject.gather(subject.scatter(REQUEST, List.of(slow))).isEmpty());
+
+      assertTrue(interrupted.await(1, TimeUnit.SECONDS), "timed-out provider was not interrupted");
+      // the only pool thread is free again, so a fresh call answers well within the timeout
+      var quotes = subject.gather(subject.scatter(REQUEST, List.of(provider("fast", "100.00"))));
+      assertEquals(List.of(new RateQuote("fast", new BigDecimal("200.00"))), quotes);
+    }
+  }
+
   @Test
   void shouldDropProviderThatFails() {
     var providers = List.<RateProvider>of(new FailingRateProvider("down"), provider("up", "50.00"));