SOLR-16505: Switch UpdateShardHandler.getRecoveryOnlyHttpClient to Jetty HTTP2

solr/CHANGES.txt

@@ -140,6 +140,8 @@ Other Changes
 * SOLR-17066: GenericSolrRequest now has a `setRequiresCollection` setter that allows it to specify whether
   it should make use of the client-level default collection/core. (Jason Gerlowski)
 
+* SOLR-16505: Switch internal replica recovery commands to Jetty HTTP2 (Sanjay Dutt, David Smiley)
+
 ==================  9.5.0 ==================
 New Features
 ---------------------

solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java

@@ -24,15 +24,14 @@
 import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
 import java.util.concurrent.TimeUnit;
-import org.apache.http.client.methods.HttpUriRequest;
 import org.apache.lucene.index.IndexCommit;
 import org.apache.lucene.search.MatchAllDocsQuery;
 import org.apache.lucene.store.Directory;
 import org.apache.solr.client.solrj.SolrClient;
 import org.apache.solr.client.solrj.SolrServerException;
-import org.apache.solr.client.solrj.impl.HttpSolrClient;
-import org.apache.solr.client.solrj.impl.HttpSolrClient.HttpUriRequestResponse;
+import org.apache.solr.client.solrj.impl.Http2SolrClient;
 import org.apache.solr.client.solrj.request.AbstractUpdateRequest;
 import org.apache.solr.client.solrj.request.CoreAdminRequest.WaitForState;
 import org.apache.solr.client.solrj.request.UpdateRequest;
@@ -124,7 +123,7 @@ public static interface RecoveryListener {
   private int retries;
   private boolean recoveringAfterStartup;
   private CoreContainer cc;
-  private volatile HttpUriRequest prevSendPreRecoveryHttpUriRequest;
+  private volatile FutureTask<NamedList<Object>> prevSendPreRecoveryHttpUriRequest;
   private final Replica.Type replicaType;
 
   private CoreDescriptor coreDescriptor;
@@ -175,24 +174,19 @@ public final void setRecoveringAfterStartup(boolean recoveringAfterStartup) {
     this.recoveringAfterStartup = recoveringAfterStartup;
   }
 
-  /** Builds a new HttpSolrClient for use in recovery. Caller must close */
-  private HttpSolrClient.Builder recoverySolrClientBuilder(String baseUrl, String leaderCoreName) {
-    // workaround for SOLR-13605: get the configured timeouts & set them directly
-    // (even though getRecoveryOnlyHttpClient() already has them set)
+  private Http2SolrClient.Builder recoverySolrClientBuilder(String baseUrl, String leaderCoreName) {
     final UpdateShardHandlerConfig cfg = cc.getConfig().getUpdateShardHandlerConfig();
-    return (new HttpSolrClient.Builder(baseUrl)
+    return new Http2SolrClient.Builder(baseUrl)
         .withDefaultCollection(leaderCoreName)
-        .withConnectionTimeout(cfg.getDistributedConnectionTimeout(), TimeUnit.MILLISECONDS)
-        .withSocketTimeout(cfg.getDistributedSocketTimeout(), TimeUnit.MILLISECONDS)
-        .withHttpClient(cc.getUpdateShardHandler().getRecoveryOnlyHttpClient()));
+        .withHttpClient(cc.getUpdateShardHandler().getRecoveryOnlyHttpClient());
   }
 
   // make sure any threads stop retrying
   @Override
   public final void close() {
     close = true;
     if (prevSendPreRecoveryHttpUriRequest != null) {
-      prevSendPreRecoveryHttpUriRequest.abort();
+      prevSendPreRecoveryHttpUriRequest.cancel(true);
     }
     log.warn("Stopping recovery for core=[{}] coreNodeName=[{}]", coreName, coreZkNodeName);
   }
@@ -634,10 +628,8 @@ public final void doSyncOrReplicateRecovery(SolrCore core) throws Exception {
                 .getCollection(cloudDesc.getCollectionName())
                 .getSlice(cloudDesc.getShardId());
 
-        try {
-          prevSendPreRecoveryHttpUriRequest.abort();
-        } catch (NullPointerException e) {
-          // okay
+        if (prevSendPreRecoveryHttpUriRequest != null) {
+          prevSendPreRecoveryHttpUriRequest.cancel(true);
         }
 
         if (isClosed()) {
@@ -894,7 +886,6 @@ public final boolean isClosed() {
 
   private final void sendPrepRecoveryCmd(String leaderBaseUrl, String leaderCoreName, Slice slice)
       throws SolrServerException, IOException, InterruptedException, ExecutionException {
-
     WaitForState prepCmd = new WaitForState();
     prepCmd.setCoreName(leaderCoreName);
     prepCmd.setNodeName(zkController.getNodeName());
@@ -915,18 +906,15 @@ private final void sendPrepRecoveryCmd(String leaderBaseUrl, String leaderCoreNa
     int readTimeout =
         conflictWaitMs
             + Integer.parseInt(System.getProperty("prepRecoveryReadTimeoutExtraWait", "8000"));
-    try (HttpSolrClient client =
+    try (SolrClient client =
         recoverySolrClientBuilder(
                 leaderBaseUrl,
                 null) // leader core omitted since client only used for 'admin' request
-            .withSocketTimeout(readTimeout, TimeUnit.MILLISECONDS)
+            .withIdleTimeout(readTimeout, TimeUnit.MILLISECONDS)
             .build()) {
-      HttpUriRequestResponse mrr = client.httpUriRequest(prepCmd);
-      prevSendPreRecoveryHttpUriRequest = mrr.httpUriRequest;
-
+      prevSendPreRecoveryHttpUriRequest = new FutureTask<>(() -> client.request(prepCmd));
       log.info("Sending prep recovery command to [{}]; [{}]", leaderBaseUrl, prepCmd);
-
-      mrr.future.get();
+      prevSendPreRecoveryHttpUriRequest.run();
     }
   }
 }

solr/core/src/java/org/apache/solr/update/UpdateShardHandler.java

@@ -75,7 +75,7 @@ public class UpdateShardHandler implements SolrInfoBean {
 
   private final Http2SolrClient updateOnlyClient;
 
-  private final CloseableHttpClient recoveryOnlyClient;
+  private final Http2SolrClient recoveryOnlyClient;
 
   private final CloseableHttpClient defaultClient;
 
@@ -85,7 +85,7 @@ public class UpdateShardHandler implements SolrInfoBean {
 
   private final InstrumentedHttpRequestExecutor httpRequestExecutor;
 
-  private final InstrumentedHttpListenerFactory updateHttpListenerFactory;
+  private final InstrumentedHttpListenerFactory trackHttpSolrMetrics;
 
   private SolrMetricsContext solrMetricsContext;
 
@@ -120,10 +120,8 @@ public UpdateShardHandler(UpdateShardHandlerConfig cfg) {
     log.debug("Created default UpdateShardHandler HTTP client with params: {}", clientParams);
 
     httpRequestExecutor = new InstrumentedHttpRequestExecutor(getMetricNameStrategy(cfg));
-    updateHttpListenerFactory = new InstrumentedHttpListenerFactory(getNameStrategy(cfg));
-    recoveryOnlyClient =
-        HttpClientUtil.createClient(
-            clientParams, recoveryOnlyConnectionManager, false, httpRequestExecutor);
+    trackHttpSolrMetrics = new InstrumentedHttpListenerFactory(getNameStrategy(cfg));
+
     defaultClient =
         HttpClientUtil.createClient(
             clientParams, defaultConnectionManager, false, httpRequestExecutor);
@@ -133,15 +131,24 @@ public UpdateShardHandler(UpdateShardHandlerConfig cfg) {
             DistributedUpdateProcessor.DISTRIB_FROM,
             DistributingUpdateProcessorFactory.DISTRIB_UPDATE_PARAM);
     Http2SolrClient.Builder updateOnlyClientBuilder = new Http2SolrClient.Builder();
+    Http2SolrClient.Builder recoveryOnlyClientBuilder = new Http2SolrClient.Builder();
     if (cfg != null) {
       updateOnlyClientBuilder
           .withConnectionTimeout(cfg.getDistributedConnectionTimeout(), TimeUnit.MILLISECONDS)
           .withIdleTimeout(cfg.getDistributedSocketTimeout(), TimeUnit.MILLISECONDS)
           .withMaxConnectionsPerHost(cfg.getMaxUpdateConnectionsPerHost());
+      recoveryOnlyClientBuilder
+          .withConnectionTimeout(cfg.getDistributedConnectionTimeout(), TimeUnit.MILLISECONDS)
+          .withIdleTimeout(cfg.getDistributedSocketTimeout(), TimeUnit.MILLISECONDS)
+          .withMaxConnectionsPerHost(cfg.getMaxUpdateConnectionsPerHost());
     }
+
     updateOnlyClientBuilder.withTheseParamNamesInTheUrl(urlParamNames);
     updateOnlyClient = updateOnlyClientBuilder.build();
-    updateOnlyClient.addListenerFactory(updateHttpListenerFactory);
+    updateOnlyClient.addListenerFactory(trackHttpSolrMetrics);
+
+    recoveryOnlyClient = recoveryOnlyClientBuilder.build();
+    recoveryOnlyClient.addListenerFactory(trackHttpSolrMetrics);
 
     ThreadFactory recoveryThreadFactory = new SolrNamedThreadFactory("recoveryExecutor");
     if (cfg != null && cfg.getMaxRecoveryThreads() > 0) {
@@ -205,7 +212,7 @@ public String getName() {
   public void initializeMetrics(SolrMetricsContext parentContext, String scope) {
     solrMetricsContext = parentContext.getChildContext(this);
     String expandedScope = SolrMetricManager.mkName(scope, getCategory().name());
-    updateHttpListenerFactory.initializeMetrics(solrMetricsContext, expandedScope);
+    trackHttpSolrMetrics.initializeMetrics(solrMetricsContext, expandedScope);
     defaultConnectionManager.initializeMetrics(solrMetricsContext, expandedScope);
     updateExecutor =
         MetricUtils.instrumentedExecutorService(
@@ -247,7 +254,7 @@ public Http2SolrClient getUpdateOnlyHttpClient() {
   }
 
   // don't introduce a bug, this client is for recovery ops only!
-  public HttpClient getRecoveryOnlyHttpClient() {
+  public Http2SolrClient getRecoveryOnlyHttpClient() {
     return recoveryOnlyClient;
   }
 
@@ -290,7 +297,7 @@ public void close() {
         // do nothing
       }
       IOUtils.closeQuietly(updateOnlyClient);
-      HttpClientUtil.close(recoveryOnlyClient);
+      IOUtils.closeQuietly(recoveryOnlyClient);
       HttpClientUtil.close(defaultClient);
       defaultConnectionManager.close();
       recoveryOnlyConnectionManager.close();

solr/core/src/test/org/apache/solr/cloud/RecoveryStrategyStressTest.java

@@ -0,0 +1,135 @@
+/*
+ * 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.cloud;
+
+import com.carrotsearch.randomizedtesting.annotations.Nightly;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrQuery;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.impl.CloudLegacySolrClient;
+import org.apache.solr.client.solrj.impl.HttpSolrClient;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.Slice;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.apache.solr.embedded.JettySolrRunner;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+@Nightly
+public class RecoveryStrategyStressTest extends SolrCloudTestCase {
+
+  @BeforeClass
+  public static void setupCluster() throws Exception {
+    cluster = configureCluster(4).addConfig("conf", configset("cloud-minimal")).configure();
+  }
+
+  @Test
+  public void stressTestRecovery() throws Exception {
+    final String collection = "recoveryStressTest";
+    CollectionAdminRequest.createCollection(collection, "conf", 1, 4)
+        .process(cluster.getSolrClient());
+    waitForState(
+        "Expected a collection with one shard and two replicas", collection, clusterShape(1, 4));
+    final var scheduledExecutorService =
+        Executors.newScheduledThreadPool(1, new SolrNamedThreadFactory("stressTestRecovery"));
+    try (SolrClient solrClient =
+        cluster.basicSolrClientBuilder().withDefaultCollection(collection).build()) {
+      final StoppableIndexingThread indexThread =
+          new StoppableIndexingThread(null, solrClient, "1", true, 10, 1, true);
+
+      final var startAndStopCount = new CountDownLatch(50);
+      final Thread startAndStopRandomReplicas =
+          new Thread(
+              () -> {
+                try {
+                  while (startAndStopCount.getCount() > 0) {
+                    DocCollection state = getCollectionState(collection);
+                    Replica leader = state.getLeader("shard1");
+                    Replica replica =
+                        getRandomReplica(state.getSlice("shard1"), (r) -> !leader.equals(r));
+
+                    JettySolrRunner jetty = cluster.getReplicaJetty(replica);
+                    jetty.stop();
+                    Thread.sleep(100);
+                    jetty.start();
+                    startAndStopCount.countDown();
+                  }
+                } catch (Exception e) {
+                  throw new RuntimeException(e);
+                }
+              });
+      startAndStopRandomReplicas.start();
+      // index and commit doc after fixed interval of 10 sec
+      scheduledExecutorService.scheduleWithFixedDelay(
+          indexThread, 1000, 10000, TimeUnit.MILLISECONDS);
+      scheduledExecutorService.scheduleWithFixedDelay(
+          () -> {
+            try {
+              new UpdateRequest().commit(solrClient, collection);
+            } catch (IOException e) {
+              throw new RuntimeException(e);
+            } catch (SolrServerException e) {
+              throw new RuntimeException(e);
+            }
+          },
+          100,
+          10000,
+          TimeUnit.MILLISECONDS);
+
+      startAndStopCount.await();
+      scheduledExecutorService.shutdownNow();
+      // final commit to make documents visible for replicas
+      new UpdateRequest().commit(solrClient, collection);
+    }
+    cluster.getZkStateReader().waitForState(collection, 120, TimeUnit.SECONDS, clusterShape(1, 4));
+
+    // test that leader and replica have same doc count
+    DocCollection state = getCollectionState(collection);
+    assertShardConsistency(state.getSlice("shard1"), true);
+  }
+
+  private void assertShardConsistency(Slice shard, boolean expectDocs) throws Exception {
+    List<Replica> replicas = shard.getReplicas(r -> r.getState() == Replica.State.ACTIVE);
+    long[] numCounts = new long[replicas.size()];
+    int i = 0;
+    for (Replica replica : replicas) {
+      try (var client =
+          new HttpSolrClient.Builder(replica.getBaseUrl())
+              .withDefaultCollection(replica.getCoreName())
+              .withHttpClient(((CloudLegacySolrClient) cluster.getSolrClient()).getHttpClient())
+              .build()) {
+        numCounts[i] =
+            client.query(new SolrQuery("*:*").add("distrib", "false")).getResults().getNumFound();
+        i++;
+      }
+    }
+    for (int j = 1; j < replicas.size(); j++) {
+      if (numCounts[j] != numCounts[j - 1])
+        fail("Mismatch in counts between replicas"); // TODO improve this!
+      if (numCounts[j] == 0 && expectDocs)
+        fail("Expected docs on shard " + shard.getName() + " but found none");
+    }
+  }
+}