Suggest load balancer setting in case of connection timeout

It may not always be an UnknownHostException, some environments
just time out if the host is not reachable.

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/impl/Utils.java

@@ -17,6 +17,7 @@
 
 import com.rabbitmq.stream.*;
 import com.rabbitmq.stream.impl.Client.ClientParameters;
+import io.netty.channel.ConnectTimeoutException;
 import java.net.UnknownHostException;
 import java.security.cert.X509Certificate;
 import java.time.Duration;
@@ -157,7 +158,9 @@ static ClientFactory coordinatorClientFactory(StreamEnvironment environment) {
                 context.key(), context1 -> new Client(context1.parameters()))
             .client(Utils.ClientFactoryContext.fromParameters(parametersCopy).key(context.key()));
       } catch (StreamException e) {
-        if (e.getCause() != null && e.getCause() instanceof UnknownHostException) {
+        if (e.getCause() != null
+            && (e.getCause() instanceof UnknownHostException
+                || e.getCause() instanceof ConnectTimeoutException)) {
           String message =
               e.getMessage()
                   + ". "

src/test/java/com/rabbitmq/stream/impl/ClientTest.java

@@ -43,6 +43,8 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.UnpooledByteBufAllocator;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ConnectTimeoutException;
 import java.io.ByteArrayOutputStream;
 import java.io.DataOutputStream;
 import java.net.UnknownHostException;
@@ -88,9 +90,16 @@ static boolean await(CountDownLatch latch, Duration timeout) {
 
   @Test
   void connectionErrorShouldReturnStreamExceptionForStackTrace() {
-    assertThatThrownBy(() -> cf.get((new ClientParameters().host(UUID.randomUUID().toString()))))
+    assertThatThrownBy(
+            () ->
+                cf.get(
+                    new ClientParameters()
+                        .host(UUID.randomUUID().toString())
+                        .bootstrapCustomizer(
+                            b -> b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1_000))))
         .isInstanceOf(StreamException.class)
-        .hasCauseInstanceOf(UnknownHostException.class);
+        .cause()
+        .isInstanceOfAny(ConnectTimeoutException.class, UnknownHostException.class);
   }
 
   @Test

src/test/java/com/rabbitmq/stream/impl/StreamConsumerTest.java

@@ -27,6 +27,8 @@
 import com.rabbitmq.stream.impl.TestUtils.BrokerVersion;
 import com.rabbitmq.stream.impl.TestUtils.BrokerVersionAtLeast;
 import com.rabbitmq.stream.impl.TestUtils.DisabledIfRabbitMqCtlNotSet;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ConnectTimeoutException;
 import io.netty.channel.EventLoopGroup;
 import java.net.UnknownHostException;
 import java.time.Duration;
@@ -971,6 +973,9 @@ void creationShouldFailWithDetailsWhenUnknownHost() {
         environmentBuilder()
             .host(localhost.host())
             .port(localhost.port())
+            .netty()
+            .bootstrapCustomizer(b -> b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1_000))
+            .environmentBuilder()
             .addressResolver(
                 n ->
                     connectionCount.getAndIncrement() == 0
@@ -982,11 +987,12 @@ void creationShouldFailWithDetailsWhenUnknownHost() {
                   env.consumerBuilder().stream(stream)
                       .messageHandler((context, message) -> {})
                       .build())
-          .hasCauseInstanceOf(UnknownHostException.class)
           .hasMessageContaining(
               "https://rabbitmq.github.io/rabbitmq-stream-java-client/stable/htmlsingle/#understanding-connection-logic")
           .hasMessageContaining(
-              "https://blog.rabbitmq.com/posts/2021/07/connecting-to-streams/#with-a-load-balancer");
+              "https://blog.rabbitmq.com/posts/2021/07/connecting-to-streams/#with-a-load-balancer")
+          .cause()
+          .isInstanceOfAny(ConnectTimeoutException.class, UnknownHostException.class);
     }
   }
 }

src/test/java/com/rabbitmq/stream/impl/StreamProducerTest.java

@@ -26,6 +26,8 @@
 import com.rabbitmq.stream.compression.Compression;
 import com.rabbitmq.stream.impl.MonitoringTestUtils.ProducerInfo;
 import com.rabbitmq.stream.impl.StreamProducer.Status;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ConnectTimeoutException;
 import io.netty.channel.EventLoopGroup;
 import java.net.UnknownHostException;
 import java.nio.charset.StandardCharsets;
@@ -649,18 +651,22 @@ void creationShouldFailWithDetailsWhenUnknownHost() {
         environmentBuilder()
             .host(localhost.host())
             .port(localhost.port())
+            .netty()
+            .bootstrapCustomizer(b -> b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1_000))
+            .environmentBuilder()
             .addressResolver(
                 n ->
                     connectionCount.getAndIncrement() == 0
                         ? n
                         : new Address(UUID.randomUUID().toString(), Client.DEFAULT_PORT));
     try (Environment env = builder.build()) {
       assertThatThrownBy(() -> env.producerBuilder().stream(stream).build())
-          .hasCauseInstanceOf(UnknownHostException.class)
           .hasMessageContaining(
               "https://rabbitmq.github.io/rabbitmq-stream-java-client/stable/htmlsingle/#understanding-connection-logic")
           .hasMessageContaining(
-              "https://blog.rabbitmq.com/posts/2021/07/connecting-to-streams/#with-a-load-balancer");
+              "https://blog.rabbitmq.com/posts/2021/07/connecting-to-streams/#with-a-load-balancer")
+          .cause()
+          .isInstanceOfAny(ConnectTimeoutException.class, UnknownHostException.class);
     }
   }
 }