Merge pull request #32128 from benstonezhang/reactive-db-connection-pool

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

docs/src/main/asciidoc/reactive-sql-clients.adoc

@@ -688,6 +688,23 @@ In `application.properties` add:
 quarkus.datasource.reactive.url=mysql:///quarkus_test?host=/var/run/mysqld/mysqld.sock
 ----
 
+== Load-balancing connections
+
+The reactive PostgreSQL and MariaDB/MySQL clients support defining several connections.
+
+A typical configuration with several connections would look like:
+
+[source,properties]
+----
+quarkus.datasource.reactive.url=postgresql://host1:5432/default,postgresql://host2:5432/default,postgresql://host3:5432/default
+----
+This can also be written with indexed property syntax:
+----
+quarkus.datasource.reactive.url[0]=postgresql://host1:5432/default
+quarkus.datasource.reactive.url[1]=postgresql://host2:5432/default
+quarkus.datasource.reactive.url[2]=postgresql://host3:5432/default
+----
+
 == Pooled Connection `idle-timeout`
 
 Reactive datasources can be configured with an `idle-timeout` (in milliseconds).

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/runtime/src/main/java/io/quarkus/reactive/db2/client/runtime/DB2PoolRecorder.java

@@ -9,6 +9,7 @@
 import static io.quarkus.vertx.core.runtime.SSLConfigHelper.configurePfxKeyCertOptions;
 import static io.quarkus.vertx.core.runtime.SSLConfigHelper.configurePfxTrustOptions;
 
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
@@ -119,7 +120,12 @@ private DB2ConnectOptions toConnectOptions(DataSourceRuntimeConfig dataSourceRun
         DB2ConnectOptions connectOptions;
 
         if (dataSourceReactiveRuntimeConfig.url.isPresent()) {
-            String url = dataSourceReactiveRuntimeConfig.url.get();
+            List<String> urls = dataSourceReactiveRuntimeConfig.url.get();
+            if (urls.size() > 1) {
+                log.warn("The Reactive DB2 client does not support multiple URLs. The first one will be used, and " +
+                        "others will be ignored.");
+            }
+            String url = urls.get(0);
             // clean up the URL to make migrations easier
             if (url.matches("^vertx-reactive:db2://.*$")) {
                 url = url.substring("vertx-reactive:".length());

extensions/reactive-mssql-client/runtime/src/main/java/io/quarkus/reactive/mssql/client/runtime/MSSQLPoolRecorder.java

@@ -9,6 +9,7 @@
 import static io.quarkus.vertx.core.runtime.SSLConfigHelper.configurePfxKeyCertOptions;
 import static io.quarkus.vertx.core.runtime.SSLConfigHelper.configurePfxTrustOptions;
 
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
@@ -118,7 +119,12 @@ private MSSQLConnectOptions toMSSQLConnectOptions(DataSourceRuntimeConfig dataSo
             DataSourceReactiveMSSQLConfig dataSourceReactiveMSSQLConfig) {
         MSSQLConnectOptions mssqlConnectOptions;
         if (dataSourceReactiveRuntimeConfig.url.isPresent()) {
-            String url = dataSourceReactiveRuntimeConfig.url.get();
+            List<String> urls = dataSourceReactiveRuntimeConfig.url.get();
+            if (urls.size() > 1) {
+                log.warn("The Reactive MSSQL client does not support multiple URLs. The first one will be used, and " +
+                        "others will be ignored.");
+            }
+            String url = urls.get(0);
             // clean up the URL to make migrations easier
             if (url.startsWith("vertx-reactive:sqlserver://")) {
                 url = url.substring("vertx-reactive:".length());

extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/LocalhostMySQLPoolCreator.java

@@ -9,7 +9,7 @@ public class LocalhostMySQLPoolCreator implements MySQLPoolCreator {
 
     @Override
     public MySQLPool create(Input input) {
-        return MySQLPool.pool(input.vertx(), input.mySQLConnectOptions().setHost("localhost").setPort(3308),
+        return MySQLPool.pool(input.vertx(), input.mySQLConnectOptionsList().get(0).setHost("localhost").setPort(3308),
                 input.poolOptions());
     }
 }

extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/MultipleDataSourcesAndMySQLPoolCreatorsTest.java

@@ -85,8 +85,8 @@ public static class DefaultMySQLPoolCreator implements MySQLPoolCreator {
 
         @Override
         public MySQLPool create(Input input) {
-            assertEquals(12345, input.mySQLConnectOptions().getPort()); // validate that the bean has been called for the proper datasource
-            return MySQLPool.pool(input.vertx(), input.mySQLConnectOptions().setHost("localhost").setPort(3308),
+            assertEquals(12345, input.mySQLConnectOptionsList().get(0).getPort()); // validate that the bean has been called for the proper datasource
+            return MySQLPool.pool(input.vertx(), input.mySQLConnectOptionsList().get(0).setHost("localhost").setPort(3308),
                     input.poolOptions());
         }
     }
@@ -97,8 +97,8 @@ public static class HibernateMySQLPoolCreator implements MySQLPoolCreator {
 
         @Override
         public MySQLPool create(Input input) {
-            assertEquals(55555, input.mySQLConnectOptions().getPort()); // validate that the bean has been called for the proper datasource
-            return MySQLPool.pool(input.vertx(), input.mySQLConnectOptions().setHost("localhost").setPort(3308),
+            assertEquals(55555, input.mySQLConnectOptionsList().get(0).getPort()); // validate that the bean has been called for the proper datasource
+            return MySQLPool.pool(input.vertx(), input.mySQLConnectOptionsList().get(0).setHost("localhost").setPort(3308),
                     input.poolOptions());
         }
     }

extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/MultipleMySQLPoolCreatorsForSameDatasourceTest.java

@@ -33,7 +33,7 @@ public static class AnotherMySQLPoolCreator implements MySQLPoolCreator {
 
         @Override
         public MySQLPool create(Input input) {
-            return MySQLPool.pool(input.vertx(), input.mySQLConnectOptions(), input.poolOptions());
+            return MySQLPool.pool(input.vertx(), input.mySQLConnectOptionsList(), input.poolOptions());
         }
     }
 

extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/MySQLPoolLoadBalanceTest.java

@@ -0,0 +1,31 @@
+package io.quarkus.reactive.mysql.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 MySQLPoolLoadBalanceTest {
+
+    @RegisterExtension
+    public static final QuarkusDevModeTest test = new QuarkusDevModeTest()
+            .withApplicationRoot((jar) -> jar
+                    .addClass(DevModeResource.class)
+                    .add(new StringAsset("quarkus.datasource.db-kind=mysql\n" +
+                            "quarkus.datasource.reactive.url=vertx-reactive:mysql://localhost:6033/load_balance_test," +
+                            "vertx-reactive:mysql://localhost:6034/load_balance_test," +
+                            "vertx-reactive:mysql://localhost:6035/load_balance_test"),
+                            "application.properties"));
+
+    @Test
+    public void testLoadBalance() {
+        RestAssured
+                .get("/dev/error")
+                .then()
+                .statusCode(200)
+                .body(Matchers.anyOf(Matchers.endsWith(":6033"), Matchers.endsWith(":6034"), Matchers.endsWith(":6035")));
+    }
+}

extensions/reactive-mysql-client/runtime/src/main/java/io/quarkus/reactive/mysql/client/MySQLPoolCreator.java

@@ -1,5 +1,7 @@
 package io.quarkus.reactive.mysql.client;
 
+import java.util.List;
+
 import io.quarkus.reactive.datasource.ReactiveDataSource;
 import io.vertx.core.Vertx;
 import io.vertx.mysqlclient.MySQLConnectOptions;
@@ -25,6 +27,6 @@ interface Input {
 
         PoolOptions poolOptions();
 
-        MySQLConnectOptions mySQLConnectOptions();
+        List<MySQLConnectOptions> mySQLConnectOptionsList();
     }
 }