Repository: smallrye/smallrye-reactive-messaging
Issue: Host disappears if used with AmqpClientOptions

If I have
`mp.messaging.outgoing.source.client-options-name=ssl-client-options`

which points to 
```
    public AmqpClientOptions getOptions() {
        return new AmqpClientOptions()
                .setSsl(true)
                // TODO Having to set the host is a bit weird since we don't need it when not using SSL
                .setHost("localhost")
                .setPfxTrustOptions(
                        new PfxOptions()
                        .setPath("/blah/blah/client.truststore.p12")
                        .setPassword("clientts")
                );
    }
```
If I remove the `setHost()` call it fails:
```
2021-11-01 15:38:11,505 ERROR [io.smallrye.reactive.messaging.amqp] (MSC service thread 1-3) SRMSG16215: Unable to connect to the broker, retry will be attempted: java.lang.NullPointerException: Host must be set
at java.util.Objects.requireNonNull(Objects.java:228)
at io.vertx.amqp.impl.AmqpClientImpl.connect(AmqpClientImpl.java:58)
at io.vertx.amqp.impl.AmqpClientImpl.connect(AmqpClientImpl.java:52)
at io.vertx.mutiny.amqp.AmqpClient.lambda$connect$1(AmqpClient.java:102)
at io.smallrye.mutiny.vertx.AsyncResultUni.subscribe(AsyncResultUni.java:31)
--- SNIP --
```
This happens whether or not I have amqp-host set in my MP Config or not. Without the AmqpClientOptions I don't need AMQP-host, since it in that case defaults to zero.

Changed files:
smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/ClientConfigurationBean.java
smallrye-reactive-messaging-amqp/src/main/java/io/smallrye/reactive/messaging/amqp/AmqpClientHelper.java
smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/AmqpSourceCDIConfigTest.java

diff --git a/smallrye-reactive-messaging-amqp/src/main/java/io/smallrye/reactive/messaging/amqp/AmqpClientHelper.java b/smallrye-reactive-messaging-amqp/src/main/java/io/smallrye/reactive/messaging/amqp/AmqpClientHelper.java
index aac28490f..dc9002534 100644
--- a/smallrye-reactive-messaging-amqp/src/main/java/io/smallrye/reactive/messaging/amqp/AmqpClientHelper.java
+++ b/smallrye-reactive-messaging-amqp/src/main/java/io/smallrye/reactive/messaging/amqp/AmqpClientHelper.java
@@ -26,7 +26,7 @@ public class AmqpClientHelper {
         Optional<String> clientOptionsName = config.getClientOptionsName();
         Vertx vertx = connector.getVertx();
         if (clientOptionsName.isPresent()) {
-            client = createClientFromClientOptionsBean(vertx, instance, clientOptionsName.get());
+            client = createClientFromClientOptionsBean(vertx, instance, clientOptionsName.get(), config);
         } else {
             client = getClient(vertx, config);
         }
@@ -35,7 +35,7 @@ public class AmqpClientHelper {
     }
 
     static AmqpClient createClientFromClientOptionsBean(Vertx vertx, Instance<AmqpClientOptions> instance,
-            String optionsBeanName) {
+            String optionsBeanName, AmqpConnectorCommonConfiguration config) {
         Instance<AmqpClientOptions> options = instance.select(Identifier.Literal.of(optionsBeanName));
         if (options.isUnsatisfied()) {
             // this `if` block should be removed when support for the `@Named` annotation is removed
@@ -48,39 +48,92 @@ public class AmqpClientHelper {
             throw ex.illegalStateFindingBean(AmqpClientOptions.class.getName(), optionsBeanName);
         }
         log.createClientFromBean(optionsBeanName);
-        return AmqpClient.create(vertx, options.get());
+
+        // We must merge the channel config and the AMQP Client options.
+        // In case of conflict, use the channel config.
+        AmqpClientOptions clientOptions = getOptions(config);
+        AmqpClientOptions completeOptions = options.get();
+        mergeTo(clientOptions, completeOptions);
+        return AmqpClient.create(vertx, completeOptions);
+    }
+
+    /**
+     * Merges the values from {@code channel} (the channel configuration), into the {@code custom}.
+     * Values from {@code channel} replaces the values from {@code custom}.
+     *
+     * @param channel the channel configuration
+     * @param custom the custom configuration
+     */
+    static void mergeTo(AmqpClientOptions channel, AmqpClientOptions custom) {
+        String username = channel.getUsername();
+        String password = channel.getPassword();
+        String host = channel.getHost();
+        int port = channel.getPort();
+        boolean useSsl = channel.isSsl();
+        int reconnectAttempts = channel.getReconnectAttempts();
+        long reconnectInterval = channel.getReconnectInterval();
+        int connectTimeout = channel.getConnectTimeout();
+
+        if (username != null) {
+            custom.setUsername(channel.getUsername());
+        }
+        if (password != null) {
+            custom.setPassword(channel.getPassword());
+        }
+        if (host != null) {
+            custom.setHost(channel.getHost());
+        }
+        if (port > 0) {
+            custom.setPort(channel.getPort());
+        }
+        if (useSsl) {
+            custom.setSsl(channel.isSsl());
+        }
+        if (reconnectAttempts > 0) {
+            custom.setReconnectAttempts(channel.getReconnectAttempts());
+        }
+        if (reconnectInterval > 0) {
+            custom.setReconnectInterval(channel.getReconnectInterval());
+        }
+        if (connectTimeout > 0) {
+            custom.setConnectTimeout(channel.getConnectTimeout());
+        }
+    }
+
+    static AmqpClientOptions getOptions(AmqpConnectorCommonConfiguration config) {
+        String username = config.getUsername().orElse(null);
+        String password = config.getPassword().orElse(null);
+        String host = config.getHost();
+        int port = config.getPort();
+        log.brokerConfigured(host, port, config.getChannel());
+        boolean useSsl = config.getUseSsl();
+        int reconnectAttempts = config.getReconnectAttempts();
+        int reconnectInterval = config.getReconnectInterval();
+        int connectTimeout = config.getConnectTimeout();
+
+        // We renamed containerID into container-id. So we must check both.
+        String containerId = config.getContainerId()
+                .orElseGet(() -> config.config.getOptionalValue("containerId", String.class).orElse(null));
+
+        AmqpClientOptions options = new AmqpClientOptions()
+                .setUsername(username)
+                .setPassword(password)
+                .setHost(host)
+                .setPort(port)
+                .setContainerId(containerId)
+                .setSsl(useSsl)
+                .setReconnectAttempts(reconnectAttempts)
+                .setReconnectInterval(reconnectInterval)
+                .setConnectTimeout(connectTimeout);
+
+        config.getSniServerName().ifPresent(options::setSniServerName);
+        config.getVirtualHost().ifPresent(options::setVirtualHost);
+        return options;
     }
 
     static AmqpClient getClient(Vertx vertx, AmqpConnectorCommonConfiguration config) {
         try {
-            String username = config.getUsername().orElse(null);
-            String password = config.getPassword().orElse(null);
-            String host = config.getHost();
-            int port = config.getPort();
-            log.brokerConfigured(host, port, config.getChannel());
-            boolean useSsl = config.getUseSsl();
-            int reconnectAttempts = config.getReconnectAttempts();
-            int reconnectInterval = config.getReconnectInterval();
-            int connectTimeout = config.getConnectTimeout();
-
-            // We renamed containerID into container-id. So we must check both.
-            String containerId = config.getContainerId()
-                    .orElseGet(() -> config.config.getOptionalValue("containerId", String.class).orElse(null));
-
-            AmqpClientOptions options = new AmqpClientOptions()
-                    .setUsername(username)
-                    .setPassword(password)
-                    .setHost(host)
-                    .setPort(port)
-                    .setContainerId(containerId)
-                    .setSsl(useSsl)
-                    .setReconnectAttempts(reconnectAttempts)
-                    .setReconnectInterval(reconnectInterval)
-                    .setConnectTimeout(connectTimeout);
-
-            config.getSniServerName().ifPresent(options::setSniServerName);
-            config.getVirtualHost().ifPresent(options::setVirtualHost);
-
+            AmqpClientOptions options = getOptions(config);
             return AmqpClient.create(vertx, options);
         } catch (Exception e) {
             log.unableToCreateClient(e);
diff --git a/smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/AmqpSourceCDIConfigTest.java b/smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/AmqpSourceCDIConfigTest.java
index 9c077359d..3d3c5bd9a 100644
--- a/smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/AmqpSourceCDIConfigTest.java
+++ b/smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/AmqpSourceCDIConfigTest.java
@@ -2,7 +2,6 @@ package io.smallrye.reactive.messaging.amqp;
 
 import static org.assertj.core.api.Assertions.*;
 import static org.awaitility.Awaitility.await;
-import static org.eclipse.microprofile.reactive.messaging.spi.ConnectorFactory.CHANNEL_NAME_ATTRIBUTE;
 
 import java.util.*;
 import java.util.concurrent.TimeUnit;
@@ -15,7 +14,6 @@ import org.jboss.weld.exceptions.DeploymentException;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 
-import io.smallrye.common.constraint.NotNull;
 import io.smallrye.config.SmallRyeConfigProviderResolver;
 import io.smallrye.reactive.messaging.connectors.ExecutionHolder;
 import io.smallrye.reactive.messaging.test.common.config.MapBasedConfig;
@@ -198,29 +196,40 @@ public class AmqpSourceCDIConfigTest extends AmqpBrokerTestBase {
                 .isInstanceOf(DeploymentException.class);
     }
 
-    @NotNull
-    private Map<String, Object> getConfig(String topic) {
-        Map<String, Object> config = new HashMap<>();
-        config.put("address", topic);
-        config.put(CHANNEL_NAME_ATTRIBUTE, UUID.randomUUID().toString());
-        config.put("host", host);
-        config.put("port", port);
-        config.put("name", "some name");
-        config.put("username", username);
-        config.put("password", password);
-        return config;
-    }
+    /**
+     * Reproduce https://github.com/smallrye/smallrye-reactive-messaging/issues/1491.
+     */
+    @Test
+    public void testClientConfigWithHostSet() {
+        Weld weld = new Weld();
+
+        String address = UUID.randomUUID().toString();
+        weld.addBeanClass(ClientConfigurationBean.class);
+        weld.addBeanClass(ConsumptionBean.class);
+
+        new MapBasedConfig()
+                .with("mp.messaging.incoming.data.address", address)
+                .with("mp.messaging.incoming.data.connector", AmqpConnector.CONNECTOR_NAME)
+                .with("mp.messaging.incoming.data.host", host)
+                .with("mp.messaging.incoming.data.port", port)
+                .with("mp.messaging.incoming.data.tracing-enabled", false)
+                .with("amqp-username", username)
+                .with("amqp-password", password)
+                .with("amqp-client-options-name", "myclientoptions2")
+                .write();
+
+        container = weld.initialize();
+        await().until(() -> isAmqpConnectorAlive(container));
+        await().until(() -> isAmqpConnectorReady(container));
+        List<Integer> list = container.select(ConsumptionBean.class).get().getResults();
+        assertThat(list).isEmpty();
+
+        AtomicInteger counter = new AtomicInteger();
+        usage.produceTenIntegers(address, counter::getAndIncrement);
+
+        await().atMost(2, TimeUnit.MINUTES).until(() -> list.size() >= 10);
+        assertThat(list).containsExactly(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
 
-    @NotNull
-    private Map<String, Object> getConfigUsingChannelName(String topic) {
-        Map<String, Object> config = new HashMap<>();
-        config.put(CHANNEL_NAME_ATTRIBUTE, topic);
-        config.put("host", host);
-        config.put("port", port);
-        config.put("name", "some name");
-        config.put("username", username);
-        config.put("password", password);
-        return config;
     }
 
 }
diff --git a/smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/ClientConfigurationBean.java b/smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/ClientConfigurationBean.java
index 254169ac2..c1ac3db79 100644
--- a/smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/ClientConfigurationBean.java
+++ b/smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/ClientConfigurationBean.java
@@ -19,4 +19,14 @@ public class ClientConfigurationBean {
                 .setPassword(System.getProperty("amqp-pwd"));
     }
 
+    @Produces
+    @Identifier("myclientoptions2")
+    public AmqpClientOptions options2() {
+        return new AmqpClientOptions()
+                .setContainerId("bla bla")
+                .setVirtualHost("foo bar")
+                .setUsername(System.getProperty("amqp-user"))
+                .setPassword(System.getProperty("amqp-pwd"));
+    }
+
 }