explicit 500 error on exception, clean up some javadsl stuff (#84)

* explicit 500 error on exception, clean up some javadsl stuff

* fix race in test

* remove unneeded dep

Author: BrandonArp

pom.xml

@@ -109,7 +109,6 @@
     <protobuf.version>3.4.0</protobuf.version>
     <reflections.version>0.9.11</reflections.version>
     <scala.version>2.11</scala.version>
-    <scala.java.compat.version>0.7.0</scala.java.compat.version>
     <scala.library.version>2.11.7</scala.library.version>
     <slf4j.version>1.7.25</slf4j.version>
     <statsd.client.timgroup>3.0.1</statsd.client.timgroup>
@@ -577,11 +576,6 @@
       <artifactId>scala-library</artifactId>
       <version>${scala.library.version}</version>
     </dependency>
-    <dependency>
-      <groupId>org.scala-lang.modules</groupId>
-      <artifactId>scala-java8-compat_${scala.version}</artifactId>
-      <version>${scala.java.compat.version}</version>
-    </dependency>
     <dependency>
       <groupId>io.vertx</groupId>
       <artifactId>vertx-core</artifactId>

src/main/java/com/arpnetworking/http/Routes.java

@@ -16,6 +16,7 @@
 package com.arpnetworking.http;
 
 import akka.NotUsed;
+import akka.actor.ActorNotFound;
 import akka.actor.ActorRef;
 import akka.actor.ActorSystem;
 import akka.actor.PoisonPill;
@@ -31,7 +32,8 @@
 import akka.http.javadsl.model.ws.Message;
 import akka.japi.JavaPartialFunction;
 import akka.japi.function.Function;
-import akka.pattern.Patterns;
+import akka.japi.pf.PFBuilder;
+import akka.pattern.PatternsCS;
 import akka.stream.OverflowStrategy;
 import akka.stream.javadsl.Flow;
 import akka.stream.javadsl.Sink;
@@ -57,8 +59,6 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.io.Resources;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
-import scala.compat.java8.FutureConverters;
-import scala.concurrent.Future;
 import scala.concurrent.duration.FiniteDuration;
 
 import java.util.Objects;
@@ -104,7 +104,11 @@ public Routes(
      */
     public Flow<HttpRequest, HttpResponse, NotUsed> flow() {
         return Flow.<HttpRequest>create()
-                .mapAsync(1, this);
+                .mapAsync(1, this)
+                .recover(
+                        new PFBuilder<Throwable, HttpResponse>()
+                                .match(Exception.class, e -> HttpResponse.create().withStatus(StatusCodes.INTERNAL_SERVER_ERROR))
+                                .build());
     }
 
     @Override
@@ -204,21 +208,32 @@ private CompletionStage<HttpResponse> process(final HttpRequest request) {
             }
         }
 
-        return CompletableFuture.completedFuture(HttpResponse.create().withStatus(404));
+        return CompletableFuture.completedFuture(HttpResponse.create().withStatus(StatusCodes.NOT_FOUND));
     }
 
     private CompletionStage<HttpResponse> dispatchHttpRequest(final HttpRequest request, final String actorName) {
-        final Future<ActorRef> refFuture = _actorSystem.actorSelection(actorName)
-                .resolveOne(FiniteDuration.create(1, TimeUnit.SECONDS));
-        return FutureConverters.toJava(refFuture).thenCompose(
+        final CompletionStage<ActorRef> refFuture = _actorSystem.actorSelection(actorName)
+                .resolveOneCS(FiniteDuration.create(1, TimeUnit.SECONDS));
+        return refFuture.thenCompose(
                 ref -> {
                     final CompletableFuture<HttpResponse> response = new CompletableFuture<>();
                     ref.tell(new RequestReply(request, response), ActorRef.noSender());
                     return response;
                 })
                 // We return 404 here since actor startup is controlled by config and
                 // the actors may not be running.
-                .exceptionally(err -> HttpResponse.create().withStatus(404));
+                .exceptionally(err -> {
+                    final Throwable cause = err.getCause();
+                    if (cause instanceof ActorNotFound) {
+                        return HttpResponse.create().withStatus(StatusCodes.NOT_FOUND);
+                    }
+                    LOGGER.error()
+                            .setMessage("Unhandled exception when looking up actor for http request routing")
+                            .addData("actorName", actorName)
+                            .setThrowable(cause)
+                            .log();
+                    return HttpResponse.create().withStatus(StatusCodes.INTERNAL_SERVER_ERROR);
+                });
     }
 
     private CompletionStage<HttpResponse> getHttpResponseForTelemetry(
@@ -258,11 +273,10 @@ public Object apply(final ActorRef telemetry, final boolean isCheck) throws Exce
 
     @SuppressWarnings("unchecked")
     private <T> CompletionStage<T> ask(final String actorPath, final Object request, final T defaultValue) {
-        return FutureConverters.toJava(
-                (Future<T>) Patterns.ask(
+        return (CompletionStage<T>) PatternsCS.ask(
                         _actorSystem.actorSelection(actorPath),
                         request,
-                        Timeout.apply(1, TimeUnit.SECONDS)))
+                        Timeout.apply(1, TimeUnit.SECONDS))
                 .exceptionally(throwable -> defaultValue);
     }
 

src/main/java/com/arpnetworking/metrics/mad/Main.java

@@ -24,6 +24,8 @@
 import akka.http.javadsl.ServerBinding;
 import akka.stream.ActorMaterializer;
 import akka.stream.Materializer;
+import akka.stream.javadsl.Sink;
+import akka.stream.javadsl.Source;
 import ch.qos.logback.classic.LoggerContext;
 import com.arpnetworking.commons.builder.Builder;
 import com.arpnetworking.commons.jackson.databind.ObjectMapperFactory;
@@ -225,13 +227,12 @@ private void launchActors(final Injector injector) {
                 _configuration.getHttpStatusPath(),
                 supplementalHttpRoutes.build());
         final Http http = Http.get(actorSystem);
-        final akka.stream.javadsl.Source<IncomingConnection, CompletionStage<ServerBinding>> binding = http.bind(
+        final Source<IncomingConnection, CompletionStage<ServerBinding>> binding = http.bind(
                 ConnectHttp.toHost(
                         _configuration.getHttpHost(),
-                        _configuration.getHttpPort()),
-                materializer);
+                        _configuration.getHttpPort()));
         binding.to(
-                akka.stream.javadsl.Sink.foreach(
+                Sink.foreach(
                         connection -> connection.handleWith(routes.flow(), materializer)))
                 .run(materializer);
     }

src/test/java/com/arpnetworking/tsdcore/sinks/PeriodicStatisticsSinkTest.java

@@ -104,7 +104,7 @@ public void testRecordProcessedAggregateData() {
         statisticsSink.recordAggregateData(TestBeanFactory.createPeriodicData());
         statisticsSink.close();
         Mockito.verify(_mockMetrics).incrementCounter(COUNTER_NAME, 1);
-        Mockito.verify(_mockMetrics).close();
+        Mockito.verify(_mockMetrics, Mockito.atLeastOnce()).close();
     }
 
     private PeriodicStatisticsSink.Builder _statisticsSinkBuilder;