Mid-level service client and updated high-level clients.

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

@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.druid.common.guava;
+
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.concurrent.ExecutionException;
+import java.util.function.Function;
+
+public class FutureUtils
+{
+  /**
+   * Waits for a given future and returns its value, like {@code future.get()}.
+   *
+   * On InterruptedException, cancels the provided future if {@code cancelIfInterrupted}, then re-throws the
+   * original InterruptedException.
+   *
+   * Passes through CancellationExceptions and ExecutionExceptions as-is.
+   */
+  public static <T> T get(final ListenableFuture<T> future, final boolean cancelIfInterrupted)
+      throws InterruptedException, ExecutionException
+  {
+    try {
+      return future.get();
+    }
+    catch (InterruptedException e) {
+      if (cancelIfInterrupted) {
+        future.cancel(true);
+      }
+
+      throw e;
+    }
+  }
+
+  /**
+   * Waits for a given future and returns its value, like {@code future.get()}.
+   *
+   * On InterruptException, cancels the provided future if {@code cancelIfInterrupted}, and in either case, throws
+   * a RuntimeException that wraps the original InterruptException.
+   *
+   * Passes through CancellationExceptions as-is.
+   *
+   * Re-wraps the causes of ExecutionExceptions using RuntimeException.
+   */
+  public static <T> T getUnchecked(final ListenableFuture<T> future, final boolean cancelIfInterrupted)
+  {
+    try {
+      return FutureUtils.get(future, cancelIfInterrupted);
+    }
+    catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new RuntimeException(e);
+    }
+    catch (ExecutionException e) {
+      throw new RuntimeException(e.getCause());
+    }
+  }
+
+  /**
+   * Like {@link Futures#transform}, but works better with lambdas due to not having overloads.
+   *
+   * One can write {@code FutureUtils.transform(future, v -> ...)} instead of
+   * {@code Futures.transform(future, (Function<? super T, ?>) v -> ...)}
+   */
+  public static <T, R> ListenableFuture<R> transform(final ListenableFuture<T> future, final Function<T, R> fn)
+  {
+    return Futures.transform(future, fn::apply);
+  }
+}

core/src/test/java/org/apache/druid/common/guava/FutureUtilsTest.java

@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.druid.common.guava;
+
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.SettableFuture;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.Execs;
+import org.hamcrest.CoreMatchers;
+import org.hamcrest.MatcherAssert;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.internal.matchers.ThrowableMessageMatcher;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class FutureUtilsTest
+{
+  private ExecutorService exec;
+
+  @Before
+  public void setUp()
+  {
+    exec = Execs.singleThreaded(StringUtils.encodeForFormat(getClass().getName()) + "-%d");
+  }
+
+  @After
+  public void tearDown()
+  {
+    if (exec != null) {
+      exec.shutdownNow();
+      exec = null;
+    }
+  }
+
+  @Test
+  public void test_get_ok() throws Exception
+  {
+    final String s = FutureUtils.get(Futures.immediateFuture("x"), true);
+    Assert.assertEquals("x", s);
+  }
+
+  @Test
+  public void test_get_failed()
+  {
+    final ExecutionException e = Assert.assertThrows(
+        ExecutionException.class,
+        () -> FutureUtils.get(Futures.immediateFailedFuture(new ISE("oh no")), true)
+    );
+
+    MatcherAssert.assertThat(e.getCause(), CoreMatchers.instanceOf(IllegalStateException.class));
+    MatcherAssert.assertThat(e.getCause(), ThrowableMessageMatcher.hasMessage(CoreMatchers.containsString("oh no")));
+  }
+
+  @Test
+  public void test_getUnchecked_interrupted_cancelOnInterrupt() throws InterruptedException
+  {
+    final SettableFuture<String> neverGoingToResolve = SettableFuture.create();
+    final AtomicReference<Throwable> exceptionFromOtherThread = new AtomicReference<>();
+    final CountDownLatch runningLatch = new CountDownLatch(1);
+
+    final Future<?> execResult = exec.submit(() -> {
+      runningLatch.countDown();
+
+      try {
+        FutureUtils.getUnchecked(neverGoingToResolve, true);
+      }
+      catch (Throwable t) {
+        exceptionFromOtherThread.set(t);
+      }
+    });
+
+    runningLatch.await();
+    Assert.assertTrue(execResult.cancel(true));
+    exec.shutdown();
+
+    Assert.assertTrue(exec.awaitTermination(1, TimeUnit.MINUTES));
+    exec = null;
+
+    Assert.assertTrue(neverGoingToResolve.isCancelled());
+
+    final Throwable e = exceptionFromOtherThread.get();
+    MatcherAssert.assertThat(e, CoreMatchers.instanceOf(RuntimeException.class));
+    MatcherAssert.assertThat(e.getCause(), CoreMatchers.instanceOf(InterruptedException.class));
+  }
+
+  @Test
+  public void test_getUnchecked_interrupted_dontCancelOnInterrupt() throws InterruptedException
+  {
+    final SettableFuture<String> neverGoingToResolve = SettableFuture.create();
+    final AtomicReference<Throwable> exceptionFromOtherThread = new AtomicReference<>();
+    final CountDownLatch runningLatch = new CountDownLatch(1);
+
+    final Future<?> execResult = exec.submit(() -> {
+      runningLatch.countDown();
+
+      try {
+        FutureUtils.getUnchecked(neverGoingToResolve, false);
+      }
+      catch (Throwable t) {
+        exceptionFromOtherThread.set(t);
+      }
+    });
+
+    runningLatch.await();
+    Assert.assertTrue(execResult.cancel(true));
+    exec.shutdown();
+
+    Assert.assertTrue(exec.awaitTermination(1, TimeUnit.MINUTES));
+    exec = null;
+
+    Assert.assertFalse(neverGoingToResolve.isCancelled());
+    Assert.assertFalse(neverGoingToResolve.isDone());
+
+    final Throwable e = exceptionFromOtherThread.get();
+    MatcherAssert.assertThat(e, CoreMatchers.instanceOf(RuntimeException.class));
+    MatcherAssert.assertThat(e.getCause(), CoreMatchers.instanceOf(InterruptedException.class));
+  }
+
+  @Test
+  public void test_getUnchecked_ok()
+  {
+    final String s = FutureUtils.getUnchecked(Futures.immediateFuture("x"), true);
+    Assert.assertEquals("x", s);
+  }
+
+  @Test
+  public void test_getUnchecked_failed()
+  {
+    final RuntimeException e = Assert.assertThrows(
+        RuntimeException.class,
+        () -> FutureUtils.getUnchecked(Futures.immediateFailedFuture(new ISE("oh no")), true)
+    );
+
+    MatcherAssert.assertThat(e.getCause(), CoreMatchers.instanceOf(IllegalStateException.class));
+    MatcherAssert.assertThat(e.getCause(), ThrowableMessageMatcher.hasMessage(CoreMatchers.containsString("oh no")));
+  }
+
+  @Test
+  public void test_transform() throws Exception
+  {
+    Assert.assertEquals(
+        "xy",
+        FutureUtils.transform(Futures.immediateFuture("x"), s -> s + "y").get()
+    );
+  }
+}

extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java

@@ -39,7 +39,7 @@
 import org.apache.druid.client.cache.CacheConfig;
 import org.apache.druid.client.cache.CachePopulatorStats;
 import org.apache.druid.client.cache.MapCache;
-import org.apache.druid.client.indexing.NoopIndexingServiceClient;
+import org.apache.druid.client.indexing.NoopOverlordClient;
 import org.apache.druid.data.input.InputEntity;
 import org.apache.druid.data.input.InputEntityReader;
 import org.apache.druid.data.input.InputFormat;
@@ -3147,7 +3147,7 @@ public void close()
         new NoopChatHandlerProvider(),
         testUtils.getRowIngestionMetersFactory(),
         new TestAppenderatorsManager(),
-        new NoopIndexingServiceClient(),
+        new NoopOverlordClient(),
         null,
         null,
         null

extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskTest.java

@@ -38,7 +38,7 @@
 import org.apache.druid.client.cache.CacheConfig;
 import org.apache.druid.client.cache.CachePopulatorStats;
 import org.apache.druid.client.cache.MapCache;
-import org.apache.druid.client.indexing.NoopIndexingServiceClient;
+import org.apache.druid.client.indexing.NoopOverlordClient;
 import org.apache.druid.common.aws.AWSCredentialsConfig;
 import org.apache.druid.data.input.impl.ByteEntity;
 import org.apache.druid.data.input.impl.DimensionsSpec;
@@ -3158,7 +3158,7 @@ public void close()
         new NoopChatHandlerProvider(),
         testUtils.getRowIngestionMetersFactory(),
         new TestAppenderatorsManager(),
-        new NoopIndexingServiceClient(),
+        new NoopOverlordClient(),
         null,
         null,
         null

indexing-service/src/main/java/org/apache/druid/indexing/common/IndexTaskClient.java

@@ -140,11 +140,6 @@ private static RetryPolicyFactory initializeRetryPolicyFactory(long numRetries)
     );
   }
 
-  protected HttpClient getHttpClient()
-  {
-    return httpClient;
-  }
-
   protected RetryPolicy newRetryPolicy()
   {
     return retryPolicyFactory.makeRetryPolicy();