Move Caching Cluster Client to java streams and allow parallel intermediate merges

core/src/main/java/org/apache/druid/collections/SerializablePair.java

@@ -31,15 +31,17 @@ public SerializablePair(@JsonProperty("lhs") T1 lhs, @JsonProperty("rhs") T2 rhs
     super(lhs, rhs);
   }
 
+  @Override
   @JsonProperty
   public T1 getLhs()
   {
-    return lhs;
+    return super.getLhs();
   }
 
+  @Override
   @JsonProperty
   public T2 getRhs()
   {
-    return rhs;
+    return super.getRhs();
   }
 }

core/src/main/java/org/apache/druid/common/guava/GuavaUtils.java

@@ -22,8 +22,12 @@
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.common.primitives.Longs;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
 
 import javax.annotation.Nullable;
+import java.util.List;
+import java.util.stream.Stream;
 
 /**
  */
@@ -62,4 +66,17 @@ public static <T extends Enum<T>> T getEnumIfPresent(final Class<T> enumClass, f
 
     return null;
   }
+
+  /**
+   * Materialze the stream of futures into a single listenable future that will return the list of results.
+   *
+   * @param futures The futures to collect into a single Listenable future
+   * @param <V>     The return value for the futures
+   *
+   * @return A single ListenableFuture whose return value is a list of the completed values of the input stream.
+   */
+  public static <V> ListenableFuture<List<V>> allFuturesAsList(Stream<ListenableFuture<? extends V>> futures)
+  {
+    return Futures.allAsList(futures::iterator);
+  }
 }

core/src/main/java/org/apache/druid/java/util/common/JodaUtils.java

@@ -28,6 +28,7 @@
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.SortedSet;
+import java.util.concurrent.TimeoutException;
 
 /**
  */
@@ -138,4 +139,23 @@ public static DateTime maxDateTime(DateTime... times)
         return max;
     }
   }
+
+  /**
+   * Return a qty of millisconds approximately until deadline. If deadline has passed, throw TimeoutException
+   *
+   * @param deadline The time on or after which things should be considered "timed out"
+   *
+   * @return A millisecond number where, if one were to wait that many milliseconds, the deadline would
+   * probably have passed. Always greater than zero
+   *
+   * @throws TimeoutException If the deadline has already passed (ties are treated as having passed the deadline)
+   */
+  public static long timeoutForDeadline(DateTime deadline) throws TimeoutException
+  {
+    final DateTime now = DateTimes.nowUtc();
+    if (now.isAfter(deadline) || now.isEqual(deadline)) {
+      throw new TimeoutException(StringUtils.format("Deadline passed: [%s]", deadline));
+    }
+    return deadline.getMillis() - now.getMillis();
+  }
 }

core/src/main/java/org/apache/druid/java/util/common/Pair.java

@@ -20,7 +20,10 @@
 package org.apache.druid.java.util.common;
 
 import javax.annotation.Nullable;
+import java.util.Map;
 import java.util.Objects;
+import java.util.stream.Collector;
+import java.util.stream.Collectors;
 
 /**
  */
@@ -32,6 +35,14 @@ public static <T1, T2> Pair<T1, T2> of(@Nullable T1 lhs, @Nullable T2 rhs)
     return new Pair<>(lhs, rhs);
   }
 
+  public static <T1, T2> Collector<Pair<T1, T2>, ?, Map<T1, T2>> mapCollector()
+  {
+    return Collectors.toMap(
+        Pair<T1, T2>::getLhs,
+        Pair<T1, T2>::getRhs
+    );
+  }
+
   @Nullable
   public final T1 lhs;
 
@@ -47,6 +58,18 @@ public Pair(
     this.rhs = rhs;
   }
 
+  @Nullable
+  public T1 getLhs()
+  {
+    return lhs;
+  }
+
+  @Nullable
+  public T2 getRhs()
+  {
+    return rhs;
+  }
+
   @Override
   public boolean equals(Object o)
   {

core/src/main/java/org/apache/druid/java/util/common/concurrent/Execs.java

@@ -22,20 +22,30 @@
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.JodaUtils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.joda.time.DateTime;
 
 import javax.annotation.Nullable;
 import javax.validation.constraints.NotNull;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinWorkerThread;
+import java.util.concurrent.Future;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.RejectedExecutionHandler;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  */
@@ -147,4 +157,113 @@ public void rejectedExecution(Runnable r, ThreadPoolExecutor executor)
         }
     );
   }
+
+  private static final AtomicLong fjpWorkerThreadCount = new AtomicLong(0L);
+
+  public static ForkJoinWorkerThread makeWorkerThread(String name, ForkJoinPool pool)
+  {
+    final ForkJoinWorkerThread t = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
+    t.setDaemon(true);
+    final long threadNumber = fjpWorkerThreadCount.incrementAndGet();
+    t.setName(StringUtils.nonStrictFormat(name, threadNumber));
+    return t;
+  }
+
+  private static final int DUMMY_THREAD_NUMBER = 17;
+
+  /**
+   * Fail fast if the format can't take a single argument integer for a thread counter.
+   *
+   * Note that LACK of any argument in the format string still renders a valid name
+   *
+   * @param format The name format to check
+   *
+   * @throws java.util.IllegalFormatException if the format passed in does is not able to take a single thread parameter
+   */
+  public static void checkThreadNameFormat(String format)
+  {
+    StringUtils.format(format, DUMMY_THREAD_NUMBER);
+  }
+
+  /**
+   * Get the result for the future (without timeout), but do so in a way safe for running in a ForkJoinPool
+   *
+   * @param future The future to block on completion
+   * @param <T>    The type of the return value
+   *
+   * @return The result of the future if successfully completed, or one of the exceptions if not
+   *
+   * @throws InterruptedException If the call to future.get() was interrupted
+   * @throws ExecutionException   If the future completed with an exception
+   */
+  public static <T> T futureManagedBlockGet(final Future<? extends T> future)
+      throws InterruptedException, ExecutionException
+  {
+    ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker()
+    {
+      @Override
+      public boolean block() throws InterruptedException
+      {
+        try {
+          future.get();
+        }
+        catch (ExecutionException e) {
+          // Ignore, will be caught when get is called below
+        }
+        return true;
+      }
+
+      @Override
+      public boolean isReleasable()
+      {
+        return future.isDone();
+      }
+    });
+    return future.get();
+  }
+
+  /**
+   * Attempt to get the result of the future before the deadline, but do so in a way safe to run in a ForkJoinPool.
+   * The deadline is best effort. It is possible the future completes, but the deadline is exceeded before the result
+   * can be returned. In such a scenario a TimeoutException will be thrown.
+   *
+   * The caller is responsible for handling the state of the Future in the case of an exception being thrown.
+   * Specifically, if an InterruptedException or a TimeoutException is thrown, there is no attempt in this method
+   * to change the behavior of the future. The caller should handle the potentially still active future as they see fit.
+   *
+   * @param future   The future to await completion
+   * @param deadline Best effort deadline for the completion of the future.
+   * @param <T>      The future's yielded type
+   *
+   * @return The yield of the future or else a thrown exception
+   *
+   * @throws InterruptedException If the call to future.get is interrupted
+   * @throws TimeoutException     If the deadline is exceeded
+   * @throws ExecutionException   If the future completed with an exception
+   */
+  public static <T> T futureManagedBlockGet(final Future<? extends T> future, final DateTime deadline)
+      throws InterruptedException, TimeoutException, ExecutionException
+  {
+    ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker()
+    {
+      @Override
+      public boolean block() throws InterruptedException
+      {
+        try {
+          future.get(JodaUtils.timeoutForDeadline(deadline), TimeUnit.MILLISECONDS);
+        }
+        catch (ExecutionException | TimeoutException e) {
+          // Will get caught later
+        }
+        return true;
+      }
+
+      @Override
+      public boolean isReleasable()
+      {
+        return future.isDone() || deadline.isBefore(DateTimes.nowUtc());
+      }
+    });
+    return future.get(JodaUtils.timeoutForDeadline(deadline), TimeUnit.MILLISECONDS);
+  }
 }