Support config reactive datasource with list of database urls for fault tolerance and load balance

extensions/reactive-datasource/runtime/src/main/java/io/quarkus/reactive/datasource/runtime/DataSourceReactiveRuntimeConfig.java

@@ -2,6 +2,7 @@
 
 import java.time.Duration;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.OptionalInt;
@@ -23,10 +24,14 @@ public class DataSourceReactiveRuntimeConfig {
     public boolean cachePreparedStatements = false;
 
     /**
-     * The datasource URL.
+     * The datasource URLs.
+     * <p>
+     * If multiple values are set, this datasource will create a pool with a list of servers instead of a single server.
+     * The pool uses a round-robin load balancing when a connection is created to select different servers.
+     * Note: some driver may not support multiple values here.
      */
     @ConfigItem
-    public Optional<String> url = Optional.empty();
+    public Optional<List<String>> url = Optional.empty();
 
     /**
      * The datasource pool maximum size.

extensions/reactive-db2-client/deployment/pom.xml

@@ -39,6 +39,21 @@
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-smallrye-health-spi</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-resteasy-reactive-deployment</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.rest-assured</groupId>
+            <artifactId>rest-assured</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-junit5-internal</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
@@ -55,6 +70,12 @@
                     </annotationProcessorPaths>
                 </configuration>
             </plugin>
+            <plugin>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <configuration>
+                    <skip>true</skip>
+                </configuration>
+            </plugin>
         </plugins>
     </build>
 </project>

extensions/reactive-db2-client/deployment/src/test/java/io.quarkus.reactive.db2.client/DB2PoolLoadBalanceTest.java

@@ -0,0 +1,31 @@
+package io.quarkus.reactive.db2.client;
+
+import org.hamcrest.Matchers;
+import org.jboss.shrinkwrap.api.asset.StringAsset;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusDevModeTest;
+import io.restassured.RestAssured;
+
+public class DB2PoolLoadBalanceTest {
+
+    @RegisterExtension
+    public static final QuarkusDevModeTest test = new QuarkusDevModeTest()
+            .withApplicationRoot((jar) -> jar
+                    .addClass(DevModeResource.class)
+                    .add(new StringAsset("quarkus.datasource.db-kind=db2\n" +
+                            "quarkus.datasource.reactive.url=vertx-reactive:db2://localhost:6133/load_balance_test," +
+                            "vertx-reactive:db2://localhost:6134/load_balance_test," +
+                            "vertx-reactive:db2://localhost:6135/load_balance_test"),
+                            "application.properties"));
+
+    @Test
+    public void testLoadBalance() {
+        RestAssured
+                .get("/dev/error")
+                .then()
+                .statusCode(200)
+                .body(Matchers.anyOf(Matchers.endsWith(":6133"), Matchers.endsWith(":6134"), Matchers.endsWith(":6135")));
+    }
+}

extensions/reactive-db2-client/deployment/src/test/java/io.quarkus.reactive.db2.client/DevModeResource.java

@@ -0,0 +1,42 @@
+package io.quarkus.reactive.db2.client;
+
+import java.net.ConnectException;
+import java.sql.SQLException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+
+import jakarta.inject.Inject;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.core.MediaType;
+import jakarta.ws.rs.core.Response;
+
+import io.vertx.db2client.DB2Pool;
+
+@Path("/dev")
+public class DevModeResource {
+
+    @Inject
+    DB2Pool client;
+
+    @GET
+    @Path("/error")
+    @Produces(MediaType.TEXT_PLAIN)
+    public CompletionStage<Response> getErrorMessage() throws SQLException {
+        CompletableFuture<Response> future = new CompletableFuture<>();
+        client.query("SELECT 1").execute(ar -> {
+            Class<?> expectedExceptionClass = ConnectException.class;
+            if (ar.succeeded()) {
+                future.complete(Response.serverError().entity("Expected SQL query to fail").build());
+            } else if (!expectedExceptionClass.isAssignableFrom(ar.cause().getClass())) {
+                ar.cause().printStackTrace();
+                future.complete(Response.serverError()
+                        .entity("Expected " + expectedExceptionClass + ", got " + ar.cause().getClass()).build());
+            } else {
+                future.complete(Response.ok(ar.cause().getMessage()).build());
+            }
+        });
+        return future;
+    }
+}

extensions/reactive-db2-client/runtime/src/main/java/io/quarkus/reactive/db2/client/DB2PoolCreator.java

@@ -1,5 +1,7 @@
 package io.quarkus.reactive.db2.client;
 
+import java.util.List;
+
 import io.quarkus.reactive.datasource.ReactiveDataSource;
 import io.vertx.core.Vertx;
 import io.vertx.db2client.DB2ConnectOptions;
@@ -25,6 +27,6 @@ interface Input {
 
         PoolOptions poolOptions();
 
-        DB2ConnectOptions db2ConnectOptions();
+        List<DB2ConnectOptions> db2ConnectOptionsList();
     }
 }

extensions/reactive-db2-client/runtime/src/main/java/io/quarkus/reactive/db2/client/runtime/DB2PoolRecorder.java

@@ -9,6 +9,8 @@
 import static io.quarkus.vertx.core.runtime.SSLConfigHelper.configurePfxKeyCertOptions;
 import static io.quarkus.vertx.core.runtime.SSLConfigHelper.configurePfxTrustOptions;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
@@ -71,16 +73,16 @@ private DB2Pool initialize(Vertx vertx,
             DataSourceReactiveDB2Config dataSourceReactiveDB2Config) {
         PoolOptions poolOptions = toPoolOptions(eventLoopCount, dataSourceRuntimeConfig, dataSourceReactiveRuntimeConfig,
                 dataSourceReactiveDB2Config);
-        DB2ConnectOptions connectOptions = toConnectOptions(dataSourceRuntimeConfig, dataSourceReactiveRuntimeConfig,
-                dataSourceReactiveDB2Config);
+        List<DB2ConnectOptions> connectOptionsList = toConnectOptions(dataSourceRuntimeConfig,
+                dataSourceReactiveRuntimeConfig, dataSourceReactiveDB2Config);
 
         // Use the convention defined by Quarkus Micrometer Vert.x metrics to create metrics prefixed with db2. and
         // the client_name as tag.
         // See io.quarkus.micrometer.runtime.binder.vertx.VertxMeterBinderAdapter.extractPrefix and
         // io.quarkus.micrometer.runtime.binder.vertx.VertxMeterBinderAdapter.extractClientName
-        connectOptions.setMetricsName("db2|" + dataSourceName);
+        connectOptionsList.forEach(connectOptions -> connectOptions.setMetricsName("db2|" + dataSourceName));
 
-        return createPool(vertx, poolOptions, connectOptions, dataSourceName);
+        return createPool(vertx, poolOptions, connectOptionsList, dataSourceName);
     }
 
     private PoolOptions toPoolOptions(Integer eventLoopCount,
@@ -113,75 +115,73 @@ private PoolOptions toPoolOptions(Integer eventLoopCount,
         return poolOptions;
     }
 
-    private DB2ConnectOptions toConnectOptions(DataSourceRuntimeConfig dataSourceRuntimeConfig,
+    private List<DB2ConnectOptions> toConnectOptions(DataSourceRuntimeConfig dataSourceRuntimeConfig,
             DataSourceReactiveRuntimeConfig dataSourceReactiveRuntimeConfig,
             DataSourceReactiveDB2Config dataSourceReactiveDB2Config) {
-        DB2ConnectOptions connectOptions;
+        List<DB2ConnectOptions> connectOptionsList = new ArrayList<>();
 
         if (dataSourceReactiveRuntimeConfig.url.isPresent()) {
-            String url = dataSourceReactiveRuntimeConfig.url.get();
-            // clean up the URL to make migrations easier
-            if (url.matches("^vertx-reactive:db2://.*$")) {
-                url = url.substring("vertx-reactive:".length());
-            }
-            connectOptions = DB2ConnectOptions.fromUri(url);
+            List<String> urls = dataSourceReactiveRuntimeConfig.url.get();
+            urls.forEach(url -> {
+                // clean up the URL to make migrations easier
+                if (url.matches("^vertx-reactive:db2://.*$")) {
+                    url = url.substring("vertx-reactive:".length());
+                }
+                connectOptionsList.add(DB2ConnectOptions.fromUri(url));
+            });
         } else {
-            connectOptions = new DB2ConnectOptions();
+            connectOptionsList.add(new DB2ConnectOptions());
         }
 
-        if (dataSourceRuntimeConfig.username.isPresent()) {
-            connectOptions.setUser(dataSourceRuntimeConfig.username.get());
-        }
-
-        if (dataSourceRuntimeConfig.password.isPresent()) {
-            connectOptions.setPassword(dataSourceRuntimeConfig.password.get());
-        }
-
-        // credentials provider
-        if (dataSourceRuntimeConfig.credentialsProvider.isPresent()) {
-            String beanName = dataSourceRuntimeConfig.credentialsProviderName.orElse(null);
-            CredentialsProvider credentialsProvider = CredentialsProviderFinder.find(beanName);
-            String name = dataSourceRuntimeConfig.credentialsProvider.get();
-            Map<String, String> credentials = credentialsProvider.getCredentials(name);
-            String user = credentials.get(USER_PROPERTY_NAME);
-            String password = credentials.get(PASSWORD_PROPERTY_NAME);
-            if (user != null) {
-                connectOptions.setUser(user);
-            }
-            if (password != null) {
-                connectOptions.setPassword(user);
+        connectOptionsList.forEach(connectOptions -> {
+            dataSourceRuntimeConfig.username.ifPresent(connectOptions::setUser);
+
+            dataSourceRuntimeConfig.password.ifPresent(connectOptions::setPassword);
+
+            // credentials provider
+            if (dataSourceRuntimeConfig.credentialsProvider.isPresent()) {
+                String beanName = dataSourceRuntimeConfig.credentialsProviderName.orElse(null);
+                CredentialsProvider credentialsProvider = CredentialsProviderFinder.find(beanName);
+                String name = dataSourceRuntimeConfig.credentialsProvider.get();
+                Map<String, String> credentials = credentialsProvider.getCredentials(name);
+                String user = credentials.get(USER_PROPERTY_NAME);
+                String password = credentials.get(PASSWORD_PROPERTY_NAME);
+                if (user != null) {
+                    connectOptions.setUser(user);
+                }
+                if (password != null) {
+                    connectOptions.setPassword(password);
+                }
             }
-        }
 
-        connectOptions.setCachePreparedStatements(dataSourceReactiveRuntimeConfig.cachePreparedStatements);
+            connectOptions.setCachePreparedStatements(dataSourceReactiveRuntimeConfig.cachePreparedStatements);
 
-        connectOptions.setSsl(dataSourceReactiveDB2Config.ssl);
+            connectOptions.setSsl(dataSourceReactiveDB2Config.ssl);
 
-        connectOptions.setTrustAll(dataSourceReactiveRuntimeConfig.trustAll);
+            connectOptions.setTrustAll(dataSourceReactiveRuntimeConfig.trustAll);
 
-        configurePemTrustOptions(connectOptions, dataSourceReactiveRuntimeConfig.trustCertificatePem);
-        configureJksTrustOptions(connectOptions, dataSourceReactiveRuntimeConfig.trustCertificateJks);
-        configurePfxTrustOptions(connectOptions, dataSourceReactiveRuntimeConfig.trustCertificatePfx);
+            configurePemTrustOptions(connectOptions, dataSourceReactiveRuntimeConfig.trustCertificatePem);
+            configureJksTrustOptions(connectOptions, dataSourceReactiveRuntimeConfig.trustCertificateJks);
+            configurePfxTrustOptions(connectOptions, dataSourceReactiveRuntimeConfig.trustCertificatePfx);
 
-        configurePemKeyCertOptions(connectOptions, dataSourceReactiveRuntimeConfig.keyCertificatePem);
-        configureJksKeyCertOptions(connectOptions, dataSourceReactiveRuntimeConfig.keyCertificateJks);
-        configurePfxKeyCertOptions(connectOptions, dataSourceReactiveRuntimeConfig.keyCertificatePfx);
+            configurePemKeyCertOptions(connectOptions, dataSourceReactiveRuntimeConfig.keyCertificatePem);
+            configureJksKeyCertOptions(connectOptions, dataSourceReactiveRuntimeConfig.keyCertificateJks);
+            configurePfxKeyCertOptions(connectOptions, dataSourceReactiveRuntimeConfig.keyCertificatePfx);
 
-        connectOptions.setReconnectAttempts(dataSourceReactiveRuntimeConfig.reconnectAttempts);
+            connectOptions.setReconnectAttempts(dataSourceReactiveRuntimeConfig.reconnectAttempts);
 
-        connectOptions.setReconnectInterval(dataSourceReactiveRuntimeConfig.reconnectInterval.toMillis());
+            connectOptions.setReconnectInterval(dataSourceReactiveRuntimeConfig.reconnectInterval.toMillis());
 
-        if (dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm.isPresent()) {
-            connectOptions.setHostnameVerificationAlgorithm(
-                    dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm.get());
-        }
+            dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm.ifPresent(
+                    connectOptions::setHostnameVerificationAlgorithm);
 
-        dataSourceReactiveRuntimeConfig.additionalProperties.forEach(connectOptions::addProperty);
+            dataSourceReactiveRuntimeConfig.additionalProperties.forEach(connectOptions::addProperty);
+        });
 
-        return connectOptions;
+        return connectOptionsList;
     }
 
-    private DB2Pool createPool(Vertx vertx, PoolOptions poolOptions, DB2ConnectOptions dB2ConnectOptions,
+    private DB2Pool createPool(Vertx vertx, PoolOptions poolOptions, List<DB2ConnectOptions> dB2ConnectOptionsList,
             String dataSourceName) {
         Instance<DB2PoolCreator> instance;
         if (DataSourceUtil.isDefault(dataSourceName)) {
@@ -191,21 +191,21 @@ private DB2Pool createPool(Vertx vertx, PoolOptions poolOptions, DB2ConnectOptio
                     new ReactiveDataSource.ReactiveDataSourceLiteral(dataSourceName));
         }
         if (instance.isResolvable()) {
-            DB2PoolCreator.Input input = new DefaultInput(vertx, poolOptions, dB2ConnectOptions);
+            DB2PoolCreator.Input input = new DefaultInput(vertx, poolOptions, dB2ConnectOptionsList);
             return instance.get().create(input);
         }
-        return DB2Pool.pool(vertx, dB2ConnectOptions, poolOptions);
+        return DB2Pool.pool(vertx, dB2ConnectOptionsList, poolOptions);
     }
 
     private static class DefaultInput implements DB2PoolCreator.Input {
         private final Vertx vertx;
         private final PoolOptions poolOptions;
-        private final DB2ConnectOptions dB2ConnectOptions;
+        private final List<DB2ConnectOptions> dB2ConnectOptionsList;
 
-        public DefaultInput(Vertx vertx, PoolOptions poolOptions, DB2ConnectOptions dB2ConnectOptions) {
+        public DefaultInput(Vertx vertx, PoolOptions poolOptions, List<DB2ConnectOptions> dB2ConnectOptionsList) {
             this.vertx = vertx;
             this.poolOptions = poolOptions;
-            this.dB2ConnectOptions = dB2ConnectOptions;
+            this.dB2ConnectOptionsList = dB2ConnectOptionsList;
         }
 
         @Override
@@ -219,8 +219,8 @@ public PoolOptions poolOptions() {
         }
 
         @Override
-        public DB2ConnectOptions db2ConnectOptions() {
-            return dB2ConnectOptions;
+        public List<DB2ConnectOptions> db2ConnectOptionsList() {
+            return dB2ConnectOptionsList;
         }
     }
 }

extensions/reactive-mssql-client/deployment/src/test/java/io/quarkus/reactive/mssql/client/LocalhostMSSQLPoolCreator.java

@@ -9,7 +9,7 @@ public class LocalhostMSSQLPoolCreator implements MSSQLPoolCreator {
 
     @Override
     public MSSQLPool create(Input input) {
-        return MSSQLPool.pool(input.vertx(), input.msSQLConnectOptions().setHost("localhost").setPort(1435),
+        return MSSQLPool.pool(input.vertx(), input.msSQLConnectOptionsList().get(0).setHost("localhost").setPort(1435),
                 input.poolOptions());
     }
 }

extensions/reactive-mssql-client/deployment/src/test/java/io/quarkus/reactive/mssql/client/MSSQLPoolLoadBalanceTest.java

@@ -0,0 +1,31 @@
+package io.quarkus.reactive.mssql.client;
+
+import org.hamcrest.Matchers;
+import org.jboss.shrinkwrap.api.asset.StringAsset;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusDevModeTest;
+import io.restassured.RestAssured;
+
+public class MSSQLPoolLoadBalanceTest {
+
+    @RegisterExtension
+    public static final QuarkusDevModeTest test = new QuarkusDevModeTest()
+            .withApplicationRoot((jar) -> jar
+                    .addClass(DevModeResource.class)
+                    .add(new StringAsset("quarkus.datasource.db-kind=mssql\n" +
+                            "quarkus.datasource.reactive.url=vertx-reactive:sqlserver://localhost:9434/load_balance_test," +
+                            "vertx-reactive:sqlserver://localhost:9435/load_balance_test," +
+                            "vertx-reactive:sqlserver://localhost:9436/load_balance_test"),
+                            "application.properties"));
+
+    @Test
+    public void testLoadBalance() {
+        RestAssured
+                .get("/dev/error")
+                .then()
+                .statusCode(200)
+                .body(Matchers.anyOf(Matchers.endsWith(":9434"), Matchers.endsWith(":9435"), Matchers.endsWith(":9436")));
+    }
+}