Multi-address client: throw MalformedURLException instead of IllegalArgumentException (#1955)

Motivation:

`MalformedURLException` is more suitable for a use-case when an URL is
not in absolute-form.

Modifications:

- Throw  `MalformedURLException` instead of `IllegalArgumentException`;
- Try-catch `selectClient` operation and return failed `Single` in case
of an error;

Result:

`StreamingUrlHttpClient` try-catches all exception during client
selection and throws `MalformedURLException` when a non-absolute form
URL is provided.

Author: idelpivnitskiy

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/DefaultMultiAddressUrlHttpClientBuilder.java

@@ -45,6 +45,7 @@
 import io.servicetalk.transport.api.IoExecutor;
 
 import java.net.InetSocketAddress;
+import java.net.MalformedURLException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.function.Function;
@@ -63,6 +64,9 @@
  * A builder of {@link StreamingHttpClient} instances which have a capacity to call any server based on the parsed
  * absolute-form URL address information from each {@link StreamingHttpRequest}.
  * <p>
+ * If {@link HttpRequestMetaData#requestTarget()} is not an absolute-form URL, a {@link MalformedURLException} will be
+ * returned or thrown.
+ * <p>
  * It also provides a good set of default settings and configurations, which could be used by most users as-is or
  * could be overridden to address specific use cases.
  *
@@ -118,23 +122,21 @@ public StreamingHttpClient buildStreaming() {
     /**
      * Returns a cached {@link UrlKey} or creates a new one based on {@link StreamingHttpRequest} information.
      */
-    private static final class CachingKeyFactory
-            implements Function<HttpRequestMetaData, UrlKey>, AsyncCloseable {
+    private static final class CachingKeyFactory implements AsyncCloseable {
 
         private final ConcurrentMap<String, UrlKey> urlKeyCache = new ConcurrentHashMap<>();
 
-        @Override
-        public UrlKey apply(final HttpRequestMetaData metaData) {
+        public UrlKey get(final HttpRequestMetaData metaData) throws MalformedURLException {
             final String host = metaData.host();
             if (host == null) {
-                throw new IllegalArgumentException(
+                throw new MalformedURLException(
                         "Request-target does not contain target host address: " + metaData.requestTarget() +
                                 ", expected absolute-form URL");
             }
 
             final String scheme = metaData.scheme();
             if (scheme == null) {
-                throw new IllegalArgumentException("Request-target does not contains scheme: " +
+                throw new MalformedURLException("Request-target does not contains scheme: " +
                         metaData.requestTarget() + ", expected absolute-form URL");
             }
 
@@ -262,21 +264,32 @@ private static final class StreamingUrlHttpClient implements FilterableStreaming
             this.executionContext = requireNonNull(executionContext);
         }
 
-        private FilterableStreamingHttpClient selectClient(
-                HttpRequestMetaData metaData) {
-            return group.get(keyFactory.apply(metaData));
+        private FilterableStreamingHttpClient selectClient(HttpRequestMetaData metaData) throws MalformedURLException {
+            return group.get(keyFactory.get(metaData));
         }
 
         @Override
         public Single<? extends FilterableReservedStreamingHttpConnection> reserveConnection(
                 final HttpExecutionStrategy strategy, final HttpRequestMetaData metaData) {
-            return defer(() -> selectClient(metaData).reserveConnection(strategy, metaData).subscribeShareContext());
+            return defer(() -> {
+                try {
+                    return selectClient(metaData).reserveConnection(strategy, metaData).subscribeShareContext();
+                } catch (Throwable t) {
+                    return Single.<FilterableReservedStreamingHttpConnection>failed(t).subscribeShareContext();
+                }
+            });
         }
 
         @Override
         public Single<StreamingHttpResponse> request(final HttpExecutionStrategy strategy,
                                                      final StreamingHttpRequest request) {
-            return defer(() -> selectClient(request).request(strategy, request).subscribeShareContext());
+            return defer(() -> {
+                try {
+                    return selectClient(request).request(strategy, request).subscribeShareContext();
+                } catch (Throwable t) {
+                    return Single.<StreamingHttpResponse>failed(t).subscribeShareContext();
+                }
+            });
         }
 
         @Override

servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/MultiAddressUrlHttpClientTest.java

@@ -35,6 +35,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.net.InetSocketAddress;
+import java.net.MalformedURLException;
 import java.net.UnknownHostException;
 import java.util.Collection;
 import java.util.List;
@@ -167,23 +168,23 @@ void requestWithRelativeFormRequestTarget() {
         StreamingHttpRequest request = client.get("/200?param=value");
         // no host header
         toSource(client.request(request)).subscribe(subscriber);
-        assertThat(subscriber.awaitOnError(), is(instanceOf(IllegalArgumentException.class)));
+        assertThat(subscriber.awaitOnError(), is(instanceOf(MalformedURLException.class)));
     }
 
     @Test
     void requestWithRelativeFormRequestTargetAndHostHeader() {
         StreamingHttpRequest request = client.get("/200?param=value");
         request.headers().set(HOST, hostHeader);
         toSource(client.request(request)).subscribe(subscriber);
-        assertThat(subscriber.awaitOnError(), is(instanceOf(IllegalArgumentException.class)));
+        assertThat(subscriber.awaitOnError(), is(instanceOf(MalformedURLException.class)));
     }
 
     @Test
     void requestWithRequestTargetWithoutScheme() {
         StreamingHttpRequest request = client.get(format("%s/200?param=value#tag", hostHeader));
         // no host header
         toSource(client.request(request)).subscribe(subscriber);
-        assertThat(subscriber.awaitOnError(), is(instanceOf(IllegalArgumentException.class)));
+        assertThat(subscriber.awaitOnError(), is(instanceOf(MalformedURLException.class)));
     }
 
     @Test
@@ -240,7 +241,7 @@ void requestWithAsteriskFormRequestTargetWithHostHeader() {
         StreamingHttpRequest request = client.newRequest(OPTIONS, "*")
                 .setHeader(HOST, hostHeader);
         toSource(client.request(request)).subscribe(subscriber);
-        assertThat(subscriber.awaitOnError(), is(instanceOf(IllegalArgumentException.class)));
+        assertThat(subscriber.awaitOnError(), is(instanceOf(MalformedURLException.class)));
     }
 
     @Test