SOLR-17995: Deprecate "core URL" SolrStream constructors

changelog/unreleased/SOLR-17995-solrstream-coreUrl-removal.yml

@@ -0,0 +1,13 @@
+# See https://github.com/apache/solr/blob/main/dev-docs/changelog.adoc
+
+title: >
+  Deprecated the `SolrStream` constructors that accept a "core URL" (e.g.
+  `SolrStream(String collectionOrCoreUrl, SolrParams params)` and
+  `SolrStream(String collectionOrCoreUrl, String path, SolrParams params)`). Use one of the
+  "base URL" constructors instead, which take the core/collection name as a separate parameter.
+type: deprecated
+authors:
+  - name: Jason Gerlowski
+links:
+  - name: SOLR-17995
+    url: https://issues.apache.org/jira/browse/SOLR-17995

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

@@ -55,6 +55,7 @@
 import org.apache.solr.common.cloud.Slice;
 import org.apache.solr.common.params.ModifiableSolrParams;
 import org.apache.solr.common.params.SolrParams;
+import org.apache.solr.common.util.URLUtil;
 
 /**
  * Connects to Zookeeper to pick replicas from a specific collection to send the query to. Under the
@@ -410,7 +411,15 @@ protected void constructStreams() throws IOException {
             getShards(this.solrConnection, this.collection, this.streamContext, mParams);
         if (shards.isEmpty())
           throw new IOException("No shards available from ZooKeeper: " + this.solrConnection);
-        streamOfSolrStream = shards.stream().map(s -> new SolrStream(s, path, mParams));
+        streamOfSolrStream =
+            shards.stream()
+                .map(
+                    s ->
+                        new SolrStream(
+                            URLUtil.extractBaseUrl(s),
+                            URLUtil.extractCoreFromCoreUrl(s),
+                            path,
+                            mParams));
       } else {
         // stream of replicas to reuse the same SolrHttpClient per baseUrl
         // avoids re-parsing data we already have in the replicas

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

@@ -50,6 +50,7 @@
 import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
 import org.apache.solr.common.params.ModifiableSolrParams;
 import org.apache.solr.common.params.SolrParams;
+import org.apache.solr.common.util.URLUtil;
 
 /**
  * Connects to Zookeeper to pick replicas from a specific collection to send the query to. Under the
@@ -295,8 +296,9 @@ protected void constructStreams() throws IOException {
         } else {
           useParams = mParams;
         }
-
-        SolrStream solrStream = new SolrStream(shardUrl, useParams);
+        final var baseUrl = URLUtil.extractBaseUrl(shardUrl);
+        final var core = URLUtil.extractCoreFromCoreUrl(shardUrl);
+        SolrStream solrStream = new SolrStream(baseUrl, useParams, core);
         if (streamContext != null) {
           solrStream.setStreamContext(streamContext);
         }

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

@@ -36,6 +36,7 @@
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
 import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
 import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.URLUtil;
 
 /**
  * The ParallelStream decorates a TupleStream implementation and pushes it to N workers for parallel
@@ -299,7 +300,12 @@ protected void constructStreams() throws IOException {
         paramsLoc.set("expr", pushStream.toString());
 
         String url = shardUrls.get(w);
-        SolrStream solrStream = new SolrStream(url, "/stream", paramsLoc);
+        SolrStream solrStream =
+            new SolrStream(
+                URLUtil.extractBaseUrl(url),
+                URLUtil.extractCoreFromCoreUrl(url),
+                "/stream",
+                paramsLoc);
         solrStream.setStreamContext(streamContext);
         solrStreams.add(solrStream);
       }

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

@@ -69,13 +69,13 @@ public class SolrStream extends TupleStream {
   private transient SolrClientCache clientCache;
   private transient boolean doCloseCache;
 
-  // TODO SOLR-17995 proposes that we should deprecate this constructor in favor of one of the other
-  // constructors that requires users to provide the core as an explicit parameter
   /**
    * @param collectionOrCoreUrl URL of the Solr core or collection to query, typically of the form
    *     "http://host:8983/solr/myCore".
    * @param params query-parameters sent with the streaming request
+   * @deprecatedUse base URL constructor instead.
    */
+  @Deprecated(since = "10.1")
   public SolrStream(String collectionOrCoreUrl, SolrParams params) {
     this.baseUrl = collectionOrCoreUrl;
     this.params = params;
@@ -89,7 +89,9 @@ public SolrStream(String collectionOrCoreUrl, SolrParams params) {
    * @param path the request handler path to query (e.g. "/export"). If not provided, defaults to
    *     "/select".
    * @param params query-parameters sent with the streaming request
+   * @deprecatedUse base URL constructor instead.
    */
+  @Deprecated(since = "10.1")
   public SolrStream(String collectionOrCoreUrl, String path, SolrParams params) {
     this(collectionOrCoreUrl, null, path, params);
   }

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

@@ -37,6 +37,7 @@
 import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
 import org.apache.solr.common.params.ModifiableSolrParams;
 import org.apache.solr.common.params.SolrParams;
+import org.apache.solr.common.util.URLUtil;
 
 /**
  * @since 7.0.0
@@ -194,7 +195,9 @@ protected void constructStream() throws IOException {
       Collections.shuffle(shardUrls, new Random());
       String url = shardUrls.get(0);
       ModifiableSolrParams mParams = new ModifiableSolrParams(params);
-      this.tupleStream = new SolrStream(url, "/sql", mParams);
+      this.tupleStream =
+          new SolrStream(
+              URLUtil.extractBaseUrl(url), URLUtil.extractCoreFromCoreUrl(url), "/sql", mParams);
       if (streamContext != null) {
         tupleStream.setStreamContext(streamContext);
         if (streamContext.isLocal()) {

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

@@ -420,8 +420,7 @@ private long getCheckpoint(Slice slice, Set<String> liveNodes) throws IOExceptio
     params.set("rows", 1);
     for (Replica replica : replicas) {
       if (replica.getState() == Replica.State.ACTIVE && liveNodes.contains(replica.getNodeName())) {
-        String coreUrl = replica.getCoreUrl();
-        SolrStream solrStream = new SolrStream(coreUrl, params);
+        SolrStream solrStream = new SolrStream(replica.getBaseUrl(), params, replica.getCoreName());
 
         if (streamContext != null) {
           StreamContext localContext = new StreamContext();
@@ -534,8 +533,7 @@ protected void constructStreams() throws IOException {
       }
 
       Replica rep = shuffler.get(random.nextInt(shuffler.size()));
-      String url = rep.getCoreUrl();
-      SolrStream solrStream = new SolrStream(url, localParams);
+      SolrStream solrStream = new SolrStream(rep.getBaseUrl(), localParams, rep.getCoreName());
       solrStream.setSlice(slice.getName());
       solrStream.setCheckpoint(checkpoint);
       solrStream.setTrace(true);

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

@@ -351,7 +351,8 @@ public void testIndirectUpdateStream() throws Exception {
     { // WRITE_X user should be able to update X via a (dummy) stream from Y...
       final SolrStream solrStream =
           new SolrStream(
-              solrUrl + "/" + COLLECTION_Y,
+              solrUrl,
+              COLLECTION_Y,
               "/stream",
               params(
                   "expr",
@@ -429,10 +430,11 @@ public void testIndirectUpdateStream() throws Exception {
   public void testIndirectUpdateStreamInsufficientCredentials() throws Exception {
 
     // regardless of how it's routed, WRITE_Y should NOT have authz to stream updates to X...
-    for (String path : Arrays.asList(COLLECTION_X, COLLECTION_Y)) {
+    for (String coll : Arrays.asList(COLLECTION_X, COLLECTION_Y)) {
       final SolrStream solrStream =
           new SolrStream(
-              solrUrl + "/" + path,
+              solrUrl,
+              coll,
               "/stream",
               params(
                   "expr",
@@ -502,8 +504,10 @@ public void testExecutorUpdateStreamInsufficientCredentials() throws Exception {
   }
 
   public void testDaemonUpdateStream() throws Exception {
-    final String daemonUrl = getRandomCoreUrl(COLLECTION_X);
-    log.info("Using Daemon @ {}", daemonUrl);
+    final Replica daemonReplica = getRandomReplica(COLLECTION_X);
+    if (log.isInfoEnabled()) {
+      log.info("Using Daemon @ {}", daemonReplica.getCoreUrl());
+    }
 
     {
       // NOTE: in spite of what is implied by 'terminate=true', this daemon will NEVER terminate on
@@ -513,7 +517,12 @@ public void testDaemonUpdateStream() throws Exception {
           "daemon(id=daemonId,runInterval=1000,terminate=true,update("
               + COLLECTION_X
               + ",tuple(id=42,a_i=1,b_i=5)))";
-      final SolrStream solrStream = new SolrStream(daemonUrl, "/stream", params("expr", expr));
+      final SolrStream solrStream =
+          new SolrStream(
+              daemonReplica.getBaseUrl(),
+              daemonReplica.getCoreName(),
+              "/stream",
+              params("expr", expr));
       solrStream.setCredentials(WRITE_X_USER, passwordFor(WRITE_X_USER));
       final List<Tuple> tuples = getTuples(solrStream);
       assertEquals(1, tuples.size()); // daemon starting status
@@ -524,7 +533,11 @@ public void testDaemonUpdateStream() throws Exception {
       final TimeOut timeout = new TimeOut(60, TimeUnit.SECONDS, TimeSource.NANO_TIME);
       while (!timeout.hasTimedOut()) {
         final SolrStream daemonCheck =
-            new SolrStream(daemonUrl, "/stream", params("action", "list"));
+            new SolrStream(
+                daemonReplica.getBaseUrl(),
+                daemonReplica.getCoreName(),
+                "/stream",
+                params("action", "list"));
         daemonCheck.setCredentials(WRITE_X_USER, passwordFor(WRITE_X_USER));
         final List<Tuple> tuples = getTuples(daemonCheck);
         assertEquals(1, tuples.size()); // our daemon;
@@ -541,7 +554,11 @@ public void testDaemonUpdateStream() throws Exception {
     } finally {
       // kill the damon...
       final SolrStream daemonKiller =
-          new SolrStream(daemonUrl, "/stream", params("action", "kill", "id", "daemonId"));
+          new SolrStream(
+              daemonReplica.getBaseUrl(),
+              daemonReplica.getCoreName(),
+              "/stream",
+              params("action", "kill", "id", "daemonId"));
       daemonKiller.setCredentials(WRITE_X_USER, passwordFor(WRITE_X_USER));
       final List<Tuple> tuples = getTuples(daemonKiller);
       assertEquals(1, tuples.size()); // daemon death status
@@ -551,8 +568,10 @@ public void testDaemonUpdateStream() throws Exception {
   }
 
   public void testDaemonUpdateStreamInsufficientCredentials() throws Exception {
-    final String daemonUrl = getRandomCoreUrl(COLLECTION_X);
-    log.info("Using Daemon @ {}", daemonUrl);
+    final Replica daemonReplica = getRandomReplica(COLLECTION_X);
+    if (log.isInfoEnabled()) {
+      log.info("Using Daemon @ {}", daemonReplica.getCoreUrl());
+    }
 
     // both of these users have valid credentials and authz read COLLECTION_X, but neither has
     // authz to write to X...
@@ -568,7 +587,10 @@ public void testDaemonUpdateStreamInsufficientCredentials() throws Exception {
                 + ",tuple(id=42,a_i=1,b_i=5)))     ";
         final SolrStream solrStream =
             new SolrStream(
-                daemonUrl, "/stream", params("_trace", "start_" + daemonId, "expr", expr));
+                daemonReplica.getBaseUrl(),
+                daemonReplica.getCoreName(),
+                "/stream",
+                params("_trace", "start_" + daemonId, "expr", expr));
         solrStream.setCredentials(user, passwordFor(user));
         final List<Tuple> tuples = getTuples(solrStream);
         assertEquals(1, tuples.size()); // daemon starting status
@@ -580,7 +602,10 @@ public void testDaemonUpdateStreamInsufficientCredentials() throws Exception {
         while (!timeout.hasTimedOut()) {
           final SolrStream daemonCheck =
               new SolrStream(
-                  daemonUrl, "/stream", params("_trace", "check_" + daemonId, "action", "list"));
+                  daemonReplica.getBaseUrl(),
+                  daemonReplica.getCoreName(),
+                  "/stream",
+                  params("_trace", "check_" + daemonId, "action", "list"));
           daemonCheck.setCredentials(user, passwordFor(user));
           final List<Tuple> tuples = getTuples(daemonCheck);
           assertEquals(1, tuples.size()); // our daemon;
@@ -604,7 +629,8 @@ public void testDaemonUpdateStreamInsufficientCredentials() throws Exception {
         // kill the damon...
         final SolrStream daemonKiller =
             new SolrStream(
-                daemonUrl,
+                daemonReplica.getBaseUrl(),
+                daemonReplica.getCoreName(),
                 "/stream",
                 params("_trace", "kill_" + daemonId, "action", "kill", "id", daemonId));
         daemonKiller.setCredentials(user, passwordFor(user));
@@ -765,7 +791,8 @@ public void testIndirectDeleteStream() throws Exception {
     { // WRITE_X user should be able to delete X via a (dummy) stream from Y...
       final SolrStream solrStream =
           new SolrStream(
-              solrUrl + "/" + COLLECTION_Y,
+              solrUrl,
+              COLLECTION_Y,
               "/stream",
               params("expr", "delete(" + COLLECTION_X + ",batchSize=1," + "tuple(id=42z))"));
       solrStream.setCredentials(WRITE_X_USER, passwordFor(WRITE_X_USER));
@@ -921,16 +948,15 @@ protected static List<Tuple> getTuples(final TupleStream tupleStream) throws IOE
   }
 
   /** Sigh. DaemonStream requires polling the same core where the stream was executed. */
-  protected static String getRandomCoreUrl(final String collection) {
-    final List<String> replicaUrls =
+  protected static Replica getRandomReplica(final String collection) {
+    final List<Replica> replicas =
         cluster
             .getZkStateReader()
             .getClusterState()
             .getCollectionOrNull(collection)
             .replicaStream()
-            .map(Replica::getCoreUrl)
             .collect(Collectors.toList());
-    Collections.shuffle(replicaUrls, random());
-    return replicaUrls.get(0);
+    Collections.shuffle(replicas, random());
+    return replicas.get(0);
   }
 }