Allow multiple hosts for elasticsearch connector

plugin/trino-elasticsearch/src/main/java/io/trino/plugin/elasticsearch/ElasticsearchConfig.java

@@ -25,6 +25,7 @@
 import javax.validation.constraints.NotNull;
 
 import java.io.File;
+import java.util.List;
 import java.util.Optional;
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
@@ -54,7 +55,7 @@ public enum Security
         PASSWORD,
     }
 
-    private String host;
+    private List<String> hosts;
     private int port = 9200;
     private String defaultSchema = "default";
     private int scrollSize = 1_000;
@@ -79,15 +80,15 @@ public enum Security
     private Security security;
 
     @NotNull
-    public String getHost()
+    public List<String> getHosts()
     {
-        return host;
+        return hosts;
     }
 
     @Config("elasticsearch.host")
-    public ElasticsearchConfig setHost(String host)
+    public ElasticsearchConfig setHosts(List<String> hosts)
     {
-        this.host = host;
+        this.hosts = hosts;
         return this;
     }
 

plugin/trino-elasticsearch/src/main/java/io/trino/plugin/elasticsearch/client/ElasticsearchClient.java

@@ -202,7 +202,9 @@ private static BackpressureRestHighLevelClient createClient(
             TimeStat backpressureStats)
     {
         RestClientBuilder builder = RestClient.builder(
-                new HttpHost(config.getHost(), config.getPort(), config.isTlsEnabled() ? "https" : "http"))
+                config.getHosts().stream()
+                        .map(httpHost -> new HttpHost(httpHost, config.getPort(), config.isTlsEnabled() ? "https" : "http"))
+                        .toArray(HttpHost[]::new))
                 .setMaxRetryTimeoutMillis(toIntExact(config.getMaxRetryTime().toMillis()));
 
         builder.setHttpClientConfigCallback(ignored -> {

plugin/trino-elasticsearch/src/test/java/io/trino/plugin/elasticsearch/TestElasticsearchConfig.java

@@ -20,6 +20,7 @@
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.Arrays;
 import java.util.Map;
 
 import static io.airlift.configuration.testing.ConfigAssertions.assertFullMapping;
@@ -36,7 +37,7 @@ public class TestElasticsearchConfig
     public void testDefaults()
     {
         assertRecordedDefaults(recordDefaults(ElasticsearchConfig.class)
-                .setHost(null)
+                .setHosts(null)
                 .setPort(9200)
                 .setDefaultSchema("default")
                 .setScrollSize(1000)
@@ -91,7 +92,7 @@ public void testExplicitPropertyMappings()
                 .buildOrThrow();
 
         ElasticsearchConfig expected = new ElasticsearchConfig()
-                .setHost("example.com")
+                .setHosts(Arrays.asList("example.com"))
                 .setPort(9999)
                 .setDefaultSchema("test")
                 .setScrollSize(4000)