[SOLR-18130][WIP] Unified connection string CloudSolrClient.Builder implementation

solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/SolrClientCache.java

@@ -18,14 +18,14 @@
 
 import java.io.Closeable;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.impl.CloudClientConnectionString;
 import org.apache.solr.client.solrj.impl.CloudSolrClient;
 import org.apache.solr.client.solrj.impl.HttpSolrClientBase;
 import org.apache.solr.client.solrj.impl.HttpSolrClientBuilderBase;
@@ -77,29 +77,33 @@ public void setDefaultZKHost(String zkHost) {
     }
   }
 
-  public synchronized CloudSolrClient getCloudSolrClient(String zkHost) {
+  public synchronized CloudSolrClient getCloudSolrClient(String connectionString) {
     ensureOpen();
-    Objects.requireNonNull(zkHost, "ZooKeeper host cannot be null!");
-    if (solrClients.containsKey(zkHost)) {
-      return (CloudSolrClient) solrClients.get(zkHost);
+    Objects.requireNonNull(connectionString, "Connection string cannot be null!");
+    if (solrClients.containsKey(connectionString)) {
+      return (CloudSolrClient) solrClients.get(connectionString);
     }
     // Can only use ZK ACLs if there is a default ZK Host, and the given ZK host contains that
     // default.
     // Basically the ZK ACLs are assumed to be only used for the default ZK host,
     // thus we should only provide the ACLs to that Zookeeper instance.
-    String zkHostNoChroot = zkHost.split("/")[0];
-    boolean canUseACLs =
-        Optional.ofNullable(defaultZkHost.get()).map(zkHostNoChroot::equals).orElse(false);
+    boolean canUseACLs = false;
+    CloudClientConnectionString cloudClientConnectionString =
+        CloudClientConnectionString.parse(connectionString);
+    if (cloudClientConnectionString.isZk()) {
+      String zkHostNoChroot = connectionString.split("/")[0];
+      canUseACLs =
+          Optional.ofNullable(defaultZkHost.get()).map(zkHostNoChroot::equals).orElse(false);
+    }
 
-    final var client = newCloudSolrClient(zkHost, httpSolrClient, canUseACLs);
-    solrClients.put(zkHost, client);
+    final var client = newCloudSolrClient(connectionString, httpSolrClient, canUseACLs);
+    solrClients.put(connectionString, client);
     return client;
   }
 
   protected CloudSolrClient newCloudSolrClient(
-      String zkHost, HttpSolrClientBase httpSolrClient, boolean canUseACLs) {
-    final List<String> hosts = List.of(zkHost);
-    var builder = new CloudSolrClient.Builder(hosts, Optional.empty());
+      String connectionString, HttpSolrClientBase httpSolrClient, boolean canUseACLs) {
+    var builder = new CloudSolrClient.Builder(connectionString);
     builder.canUseZkACLs(canUseACLs);
     // using internal builder to ensure the internal client gets closed
     builder = builder.withHttpClientBuilder(newHttpSolrClientBuilder(null, httpSolrClient));

solr/solrj-streaming/src/test/org/apache/solr/client/solrj/io/SolrClientCacheTest.java

@@ -17,13 +17,16 @@
 package org.apache.solr.client.solrj.io;
 
 import java.util.Map;
+import org.apache.solr.client.solrj.impl.CloudSolrClient;
 import org.apache.solr.cloud.SolrCloudTestCase;
 import org.apache.solr.common.SolrException;
+import org.apache.solr.common.cloud.ClusterState;
 import org.apache.solr.common.cloud.DigestZkACLProvider;
 import org.apache.solr.common.cloud.DigestZkCredentialsProvider;
 import org.apache.solr.common.cloud.SolrZkClient;
 import org.apache.solr.common.cloud.VMParamsZkCredentialsInjector;
 import org.junit.AfterClass;
+import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
@@ -66,6 +69,27 @@ public void testZkACLsNotUsedWithDifferentZkHost() {
     }
   }
 
+  @Test
+  public void testGetClientWithHttp() {
+    String solrUrl = cluster.getJettySolrRunner(0).getBaseUrl().toString();
+    try (SolrClientCache cache = new SolrClientCache()) {
+      CloudSolrClient cloudSolrClient = cache.getCloudSolrClient(solrUrl);
+      ClusterState clusterState = cloudSolrClient.getClusterStateProvider().getClusterState();
+      Assert.assertEquals(1, clusterState.getLiveNodes().size());
+    }
+  }
+
+  @Test
+  public void testGetClientWithZookeeper() {
+    String zkConnectionString = zkClient().getZkServerAddress();
+    try (SolrClientCache cache = new SolrClientCache()) {
+      cache.setDefaultZKHost(zkClient().getZkServerAddress());
+      CloudSolrClient cloudSolrClient = cache.getCloudSolrClient(zkConnectionString);
+      ClusterState clusterState = cloudSolrClient.getClusterStateProvider().getClusterState();
+      Assert.assertEquals(1, clusterState.getLiveNodes().size());
+    }
+  }
+
   @Test
   public void testZkACLsUsedWithDifferentChroot() {
     try (SolrClientCache cache = new SolrClientCache()) {

solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudClientConnectionString.java

@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.impl;
+
+import java.net.URI;
+import java.util.List;
+import org.apache.solr.common.util.StrUtils;
+
+/** Universal connection string parser logic. */
+public record CloudClientConnectionString(boolean isZk, List<String> quorumItems, String zkChroot) {
+
+  public CloudClientConnectionString {
+    if (quorumItems == null || quorumItems.isEmpty()) {
+      throw new IllegalArgumentException("No valid hosts/urls found");
+    }
+  }
+
+  public static boolean isValidHttpUrl(String s) {
+    if (s == null || s.isBlank()) return false;
+
+    try {
+      URI uri = new URI(s);
+
+      return ("http".equalsIgnoreCase(uri.getScheme()) || "https".equalsIgnoreCase(uri.getScheme()))
+          && uri.getHost() != null;
+
+    } catch (Exception e) {
+      return false;
+    }
+  }
+
+  public static CloudClientConnectionString parse(String connectionString) {
+    if (connectionString == null || connectionString.trim().isEmpty()) {
+      throw new IllegalArgumentException("Connection string must not be null or empty");
+    }
+    connectionString = connectionString.trim();
+    if (connectionString.contains("://")) {
+      return parseHttpQuorum(connectionString);
+    }
+    return parseZkQuorum(connectionString);
+  }
+
+  private static CloudClientConnectionString parseZkQuorum(String connectionString) {
+    String zkChroot = null;
+    String zkHosts = connectionString;
+    int slashIndex = connectionString.indexOf('/');
+    if (slashIndex != -1) {
+      zkHosts = connectionString.substring(0, slashIndex);
+      zkChroot = connectionString.substring(slashIndex);
+    }
+    List<String> quorumItems = StrUtils.split(zkHosts, ',').stream().map(String::trim).toList();
+    for (String host : quorumItems) {
+      if (host == null || host.isBlank()) {
+        throw new IllegalArgumentException("Empty host in Zookeeper connection string");
+      }
+    }
+    return new CloudClientConnectionString(true, quorumItems, zkChroot);
+  }
+
+  private static CloudClientConnectionString parseHttpQuorum(String connectionString) {
+    List<String> httpUrls =
+        StrUtils.split(connectionString, ',').stream().map(String::trim).toList();
+    for (String part : httpUrls) {
+      if (!isValidHttpUrl(part)) {
+        throw new IllegalArgumentException("Malformed HTTP(S) URL: " + part);
+      }
+    }
+    return new CloudClientConnectionString(false, httpUrls, null);
+  }
+}

solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java

@@ -241,6 +241,44 @@ public Builder(ClusterStateProvider stateProvider) {
       this.stateProvider = stateProvider;
     }
 
+    /**
+     * Creates a client builder based on a connection string of 2 possible formats:
+     *
+     * <ul>
+     *   <li>ZooKeeper connection string (optionally with chroot), e.g. {@code
+     *       zk1:2181,zk2:2181,zk3:2181/solr}
+     *   <li>Comma-separated list of Solr node base URLs (HTTP or HTTPS), e.g. {@code
+     *       http://solr1:8983/solr,http://solr2:8983/solr}
+     * </ul>
+     *
+     * <p>Usage examples:
+     *
+     * <pre>{@code
+     * // ZooKeeper with chroot
+     * new CloudSolrClient.Builder("zk1:2181,zk2:2181,zk3:2181/solr");
+     *
+     * // ZooKeeper without chroot
+     * new CloudSolrClient.Builder("zk1:2181,zk2:2181,zk3:2181");
+     *
+     * // Direct HTTPS connections
+     * new CloudSolrClient.Builder("https://solr1:8983/solr,https://solr2:8983/solr");
+     * }</pre>
+     *
+     * @param connectionString a string specifying either ZooKeeper connection string or HTTP(S)
+     *     Solr URLs
+     * @throws IllegalArgumentException if string is null, empty, or malformed
+     * @since 11.0.0
+     */
+    public Builder(String connectionString) {
+      CloudClientConnectionString connStr = CloudClientConnectionString.parse(connectionString);
+      if (connStr.isZk()) {
+        this.zkHosts = connStr.quorumItems();
+        this.zkChroot = connStr.zkChroot();
+      } else {
+        this.solrUrls = connStr.quorumItems();
+      }
+    }
+
     /** Whether to use the default ZK ACLs when building a ZK Client. */
     public Builder canUseZkACLs(boolean canUseZkACLs) {
       this.canUseZkACLs = canUseZkACLs;