Properly catch errors thrown by request handlers

If the request handlers throw any exceptions, the call to
JettyHttpClient would previously fail with an inline exception, rather
than returning a failed future. This is fixed here.

http-client/src/main/java/io/airlift/http/client/jetty/FailedHttpResponseFuture.java

@@ -0,0 +1,66 @@
+package io.airlift.http.client.jetty;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import io.airlift.http.client.HttpClient;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static com.google.common.util.concurrent.Futures.immediateFailedFuture;
+
+public class FailedHttpResponseFuture<T>
+        implements HttpClient.HttpResponseFuture<T>
+{
+    private final ListenableFuture<T> delegate;
+
+    public FailedHttpResponseFuture(Throwable throwable)
+    {
+        this.delegate = immediateFailedFuture(throwable);
+    }
+
+    @Override
+    public String getState()
+    {
+        return "FAILED";
+    }
+
+    @Override
+    public void addListener(Runnable listener, Executor executor)
+    {
+        delegate.addListener(listener, executor);
+    }
+
+    @Override
+    public boolean cancel(boolean mayInterruptIfRunning)
+    {
+        return delegate.cancel(mayInterruptIfRunning);
+    }
+
+    @Override
+    public boolean isCancelled()
+    {
+        return delegate.isCancelled();
+    }
+
+    @Override
+    public boolean isDone()
+    {
+        return delegate.isDone();
+    }
+
+    @Override
+    public T get()
+            throws InterruptedException, ExecutionException
+    {
+        return delegate.get();
+    }
+
+    @Override
+    public T get(long timeout, TimeUnit unit)
+            throws InterruptedException, ExecutionException, TimeoutException
+    {
+        return delegate.get(timeout, unit);
+    }
+}

http-client/src/main/java/io/airlift/http/client/jetty/JettyHttpClient.java

@@ -765,7 +765,16 @@ public <T, E extends Exception> HttpResponseFuture<T> executeAsync(Request reque
         requireNonNull(request, "request is null");
         requireNonNull(responseHandler, "responseHandler is null");
 
-        request = applyRequestFilters(request);
+        try {
+            request = applyRequestFilters(request);
+        }
+        catch (RuntimeException e) {
+            startSpan(request)
+                    .setStatus(StatusCode.ERROR, e.getMessage())
+                    .recordException(e, Attributes.of(ExceptionAttributes.EXCEPTION_ESCAPED, true))
+                    .end();
+            return new FailedHttpResponseFuture<>(e);
+        }
 
         Span span = startSpan(request);
         request = injectTracing(request, span);