SOLR-17821: Fix error scenario for ShardInstall or Restore

changelog/unreleased/solr-17821-fix-restore-error-scenario.yml

@@ -0,0 +1,9 @@
+# See https://github.com/apache/solr/blob/main/dev-docs/changelog.adoc
+title: Fix error scenario in InstallShardData and Restore
+type: fixed # added, changed, fixed, deprecated, removed, dependency_update, security, other
+authors:
+  - name: Houston Putman
+    nick: HoustonPutman
+links:
+  - name: SOLR-17821
+    url: https://issues.apache.org/jira/browse/SOLR-17821

solr/api/src/java/org/apache/solr/client/api/model/InstallShardDataRequestBody.java

@@ -24,5 +24,9 @@ public class InstallShardDataRequestBody {
 
   @JsonProperty public String repository;
 
+  @JsonProperty public String name;
+
+  @JsonProperty public String shardBackupId;
+
   @JsonProperty public String async;
 }

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

@@ -74,11 +74,6 @@ public SyncStrategy(CoreContainer cc) {
     updateExecutor = updateShardHandler.getUpdateExecutor();
   }
 
-  private static class ShardCoreRequest extends ShardRequest {
-    String coreName;
-    public String baseUrl;
-  }
-
   public PeerSync.PeerSyncResult sync(
       ZkController zkController, SolrCore core, ZkNodeProps leaderProps) {
     return sync(zkController, core, leaderProps, false, false);
@@ -322,8 +317,8 @@ private void syncToMe(
         } else {
           RecoveryRequest rr = new RecoveryRequest();
           rr.leaderProps = leaderProps;
-          rr.baseUrl = ((ShardCoreRequest) srsp.getShardRequest()).baseUrl;
-          rr.coreName = ((ShardCoreRequest) srsp.getShardRequest()).coreName;
+          rr.baseUrl = srsp.getShardRequest().nodeName;
+          rr.coreName = srsp.getShardRequest().coreName;
           recoveryRequests.add(rr);
         }
       } else {
@@ -355,9 +350,9 @@ private boolean handleResponse(ShardResponse srsp) {
   private void requestSync(
       String baseUrl, String replica, String leaderUrl, String coreName, int nUpdates) {
     // TODO should we use peerSyncWithLeader instead?
-    ShardCoreRequest sreq = new ShardCoreRequest();
+    ShardRequest sreq = new ShardRequest();
     sreq.coreName = coreName;
-    sreq.baseUrl = baseUrl;
+    sreq.nodeName = baseUrl;
     sreq.purpose = ShardRequest.PURPOSE_PRIVATE;
     sreq.shards = new String[] {replica};
     sreq.actualShards = sreq.shards;

solr/core/src/java/org/apache/solr/cloud/api/collections/InstallShardDataCmd.java

@@ -17,20 +17,27 @@
 
 package org.apache.solr.cloud.api.collections;
 
-import static org.apache.solr.cloud.Overseer.QUEUE_OPERATION;
-import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
-
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.lang.invoke.MethodHandles;
-import java.util.HashMap;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
 import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import org.apache.solr.cloud.ZkShardTerms;
+import org.apache.solr.common.SolrErrorWrappingException;
 import org.apache.solr.common.SolrException;
 import org.apache.solr.common.cloud.ClusterState;
 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.cloud.ZkNodeProps;
-import org.apache.solr.common.params.CollectionParams;
 import org.apache.solr.common.params.CoreAdminParams;
 import org.apache.solr.common.params.ModifiableSolrParams;
 import org.apache.solr.common.util.NamedList;
@@ -80,27 +87,114 @@ public void call(AdminCmdContext adminCmdContext, ZkNodeProps message, NamedList
     // Build the core-admin request
     final ModifiableSolrParams coreApiParams = new ModifiableSolrParams();
     coreApiParams.set(
-        CoreAdminParams.ACTION, CoreAdminParams.CoreAdminAction.INSTALLCOREDATA.toString());
-    typedMessage.toMap(new HashMap<>()).forEach((k, v) -> coreApiParams.set(k, v.toString()));
+        CoreAdminParams.ACTION, CoreAdminParams.CoreAdminAction.RESTORECORE.toString());
+    coreApiParams.set(CoreAdminParams.BACKUP_LOCATION, typedMessage.location);
+    coreApiParams.set(CoreAdminParams.BACKUP_REPOSITORY, typedMessage.repository);
+    coreApiParams.set(CoreAdminParams.NAME, typedMessage.name);
+    coreApiParams.set(CoreAdminParams.SHARD_BACKUP_ID, typedMessage.shardBackupId);
 
     // Send the core-admin request to each replica in the slice
     final ShardHandler shardHandler = ccc.newShardHandler();
-    shardRequestTracker.sliceCmd(clusterState, coreApiParams, null, installSlice, shardHandler);
+    List<Replica> notLiveReplicas =
+        shardRequestTracker.sliceCmd(clusterState, coreApiParams, null, installSlice, shardHandler);
     final String errorMessage =
         String.format(
             Locale.ROOT,
-            "Could not install data to collection [%s] and shard [%s]",
+            "Could not install data to collection [%s] and shard [%s] on any leader-eligible replicas",
             typedMessage.collection,
             typedMessage.shard);
-    shardRequestTracker.processResponses(results, shardHandler, true, errorMessage);
+    shardRequestTracker.processResponses(results, shardHandler, false, errorMessage);
+    Collection<Replica> allReplicas =
+        clusterState
+            .getCollection(typedMessage.collection)
+            .getSlice(typedMessage.shard)
+            .getReplicas();
+
+    // Ensure that terms are correct for this shard after the execution is done
+    // We only care about leader eligible replicas, all others will eventually get updated.
+    List<Replica> leaderEligibleReplicas =
+        allReplicas.stream().filter(r -> r.getType().leaderEligible).collect(Collectors.toList());
+
+    NamedList<Object> failures = (NamedList<Object>) results.get("failure");
+    Set<Replica> successfulReplicas =
+        leaderEligibleReplicas.stream()
+            .filter(replica -> !notLiveReplicas.contains(replica))
+            .filter(
+                replica ->
+                    failures == null
+                        || failures.get(CollectionHandlingUtils.requestKey(replica)) == null)
+            .collect(Collectors.toSet());
+
+    if (successfulReplicas.isEmpty()) {
+      // No leader-eligible replicas succeeded, return failure
+      if (failures == null) {
+        throw new SolrException(
+            SolrException.ErrorCode.SERVER_ERROR,
+            errorMessage + ". No leader-eligible replicas are live.");
+      } else {
+        throw new SolrErrorWrappingException(
+            SolrException.ErrorCode.SERVER_ERROR,
+            errorMessage,
+            Collections.singletonList(failures.asMap(1)));
+      }
+    } else if (successfulReplicas.size() < leaderEligibleReplicas.size()) {
+      // Some, but not all, leader-eligible replicas succeeded.
+      // Ensure the shard terms are correct so that the non-successful replicas go into recovery
+      ZkShardTerms shardTerms =
+          ccc.getCoreContainer()
+              .getZkController()
+              .getShardTerms(typedMessage.collection, typedMessage.shard);
+      final Set<String> replicasToStartRecovery = new HashSet<>();
+      leaderEligibleReplicas.stream()
+          .filter(r -> !successfulReplicas.contains(r))
+          .map(Replica::getName)
+          .forEach(replicasToStartRecovery::add);
+      log.info("Putting the unsuccessful replicas into recovery: {}", replicasToStartRecovery);
+      shardTerms.ensureHighestTerms(
+          installCollection,
+          successfulReplicas.stream().map(Replica::getName).collect(Collectors.toSet()));
+      ccc.getZkStateReader()
+          .waitForState(
+              typedMessage.collection,
+              30,
+              TimeUnit.SECONDS,
+              (liveNodes, collectionState) -> {
+                collectionState.getSlice(typedMessage.shard).getReplicas().stream()
+                    .filter(r -> Replica.State.RECOVERING.equals(r.getState()))
+                    .map(Replica::getName)
+                    .forEach(replicasToStartRecovery::remove);
+                return replicasToStartRecovery.isEmpty();
+              });
+
+      // In order for the async request to succeed, we need to ensure that there is no failure
+      // message
+      NamedList<Object> successes = (NamedList<Object>) results.get("success");
+      failures.forEach(
+          (replicaKey, value) -> {
+            successes.add(
+                replicaKey,
+                new NamedList<>(
+                    Map.of(
+                        "explanation",
+                        "Core install failed, but is now recovering from the leader",
+                        "failure",
+                        value)));
+          });
+      results.remove("failure");
+    } else {
+      // other replicas to-be-created will know that they are out of date by
+      // looking at their term : 0 compare to term of this core : 1
+      ccc.getCoreContainer()
+          .getZkController()
+          .getShardTerms(typedMessage.collection, typedMessage.shard)
+          .ensureHighestTermsAreNotZero();
+    }
   }
 
   /** A value-type representing the message received by {@link InstallShardDataCmd} */
+  @JsonIgnoreProperties(ignoreUnknown = true)
   public static class RemoteMessage implements JacksonReflectMapWriter {
 
-    @JsonProperty(QUEUE_OPERATION)
-    public String operation = CollectionParams.CollectionAction.INSTALLSHARDDATA.toLower();
-
     @JsonProperty public String collection;
 
     @JsonProperty public String shard;
@@ -109,8 +203,9 @@ public static class RemoteMessage implements JacksonReflectMapWriter {
 
     @JsonProperty public String location;
 
-    @JsonProperty(ASYNC)
-    public String asyncId;
+    @JsonProperty public String name = "";
+
+    @JsonProperty public String shardBackupId;
 
     public void validate() {
       if (StrUtils.isBlank(collection)) {

solr/core/src/java/org/apache/solr/cloud/api/collections/RestoreCmd.java

@@ -24,6 +24,7 @@
 import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDREPLICA;
 import static org.apache.solr.common.params.CollectionParams.CollectionAction.CREATE;
 import static org.apache.solr.common.params.CollectionParams.CollectionAction.CREATESHARD;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.INSTALLSHARDDATA;
 import static org.apache.solr.common.params.CollectionParams.CollectionAction.MODIFYCOLLECTION;
 import static org.apache.solr.common.params.CommonParams.NAME;
 
@@ -108,7 +109,7 @@ public void call(AdminCmdContext adminCmdContext, ZkNodeProps message, NamedList
     }
   }
 
-  private void requestReplicasToRestore(
+  private void requestShardsToRestore(
       NamedList<Object> results,
       DocCollection restoreCollection,
       AdminCmdContext adminCmdContext,
@@ -117,11 +118,13 @@ private void requestReplicasToRestore(
       String repo,
       ShardHandler shardHandler) {
     ShardRequestTracker shardRequestTracker =
-        CollectionHandlingUtils.asyncRequestTracker(adminCmdContext, ccc);
+        CollectionHandlingUtils.asyncRequestTracker(adminCmdContext, "/admin/collections", ccc);
     // Copy data from backed up index to each replica
     for (Slice slice : restoreCollection.getSlices()) {
       ModifiableSolrParams params = new ModifiableSolrParams();
-      params.set(CoreAdminParams.ACTION, CoreAdminParams.CoreAdminAction.RESTORECORE.toString());
+      params.set(CollectionAdminParams.COLLECTION, slice.getCollection());
+      params.set(CollectionAdminParams.SHARD, slice.getName());
+      params.set(CoreAdminParams.ACTION, INSTALLSHARDDATA.toString());
       Optional<ShardBackupId> shardBackupId = backupProperties.getShardBackupIdFor(slice.getName());
       if (shardBackupId.isPresent()) {
         params.set(CoreAdminParams.SHARD_BACKUP_ID, shardBackupId.get().getIdAsString());
@@ -130,11 +133,24 @@ private void requestReplicasToRestore(
       }
       params.set(CoreAdminParams.BACKUP_LOCATION, backupPath.toASCIIString());
       params.set(CoreAdminParams.BACKUP_REPOSITORY, repo);
-      shardRequestTracker.sliceCmd(
-          adminCmdContext.getClusterState(), params, null, slice, shardHandler);
+      Replica replica = slice.getLeader();
+      if (replica == null) {
+        replica =
+            slice.getReplicas().stream()
+                .findFirst()
+                .orElseThrow(
+                    () ->
+                        new SolrException(
+                            ErrorCode.INVALID_STATE,
+                            String.format(
+                                Locale.ROOT,
+                                "No replicas for shard %s in collection %s. Cannot restore to a shard with no replicas",
+                                slice.getName(),
+                                slice.getCollection())));
+      }
+      shardRequestTracker.sendShardRequest(replica, params, shardHandler);
     }
-    shardRequestTracker.processResponses(
-        new NamedList<>(), shardHandler, true, "Could not restore core");
+    shardRequestTracker.processResponses(results, shardHandler, true, "Could not restore shard");
   }
 
   /** Encapsulates the parsing and access for common parameters restore parameters and values */
@@ -273,7 +289,7 @@ public void process(NamedList<Object> results, RestoreContext rc) throws Excepti
       // refresh the location copy of collection state
       restoreCollection =
           rc.zkStateReader.getClusterState().getCollection(rc.restoreCollectionName);
-      requestReplicasToRestore(
+      requestShardsToRestore(
           results,
           restoreCollection,
           rc.adminCmdContext.withClusterState(rc.zkStateReader.getClusterState()),
@@ -625,7 +641,7 @@ public void process(RestoreContext rc, NamedList<Object> results) throws Excepti
           rc.adminCmdContext.withClusterState(rc.zkStateReader.getClusterState()),
           restoreCollection);
       try {
-        requestReplicasToRestore(
+        requestShardsToRestore(
             results,
             restoreCollection,
             rc.adminCmdContext.withClusterState(rc.zkStateReader.getClusterState()),
@@ -649,8 +665,7 @@ private void disableReadOnly(AdminCmdContext adminCmdContext, DocCollection rest
               ZkStateReader.COLLECTION_PROP, restoreCollection.getName(),
               ZkStateReader.READ_ONLY, null);
       new CollApiCmds.ModifyCollectionCmd(ccc)
-          .call(
-              adminCmdContext.subRequestContext(MODIFYCOLLECTION, null), params, new NamedList<>());
+          .call(adminCmdContext.subRequestContext(MODIFYCOLLECTION), params, new NamedList<>());
     }
 
     private void enableReadOnly(AdminCmdContext adminCmdContext, DocCollection restoreCollection)
@@ -662,8 +677,7 @@ private void enableReadOnly(AdminCmdContext adminCmdContext, DocCollection resto
               ZkStateReader.COLLECTION_PROP, restoreCollection.getName(),
               ZkStateReader.READ_ONLY, "true");
       new CollApiCmds.ModifyCollectionCmd(ccc)
-          .call(
-              adminCmdContext.subRequestContext(MODIFYCOLLECTION, null), params, new NamedList<>());
+          .call(adminCmdContext.subRequestContext(MODIFYCOLLECTION), params, new NamedList<>());
     }
   }
 }

solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java

@@ -101,6 +101,7 @@
 import static org.apache.solr.common.params.CommonParams.VALUE_LONG;
 import static org.apache.solr.common.params.CoreAdminParams.BACKUP_LOCATION;
 import static org.apache.solr.common.params.CoreAdminParams.BACKUP_REPOSITORY;
+import static org.apache.solr.common.params.CoreAdminParams.SHARD_BACKUP_ID;
 import static org.apache.solr.common.util.StrUtils.formatString;
 
 import java.lang.invoke.MethodHandles;
@@ -1068,6 +1069,8 @@ public Map<String, Object> execute(
           reqBody.async = req.getParams().get(ASYNC);
           reqBody.repository = req.getParams().get(BACKUP_REPOSITORY);
           reqBody.location = req.getParams().get(BACKUP_LOCATION);
+          reqBody.name = req.getParams().get(NAME);
+          reqBody.shardBackupId = req.getParams().get(SHARD_BACKUP_ID);
 
           final InstallShardData installApi = new InstallShardData(h.coreContainer, req, rsp);
           final SolrJerseyResponse installResponse =

solr/core/src/java/org/apache/solr/handler/admin/api/InstallCoreData.java

@@ -95,12 +95,6 @@ public SolrJerseyResponse installCoreData(String coreName, InstallCoreDataReques
             SolrException.ErrorCode.SERVER_ERROR,
             "Failed to install data to core=" + core.getName());
       }
-
-      // other replicas to-be-created will know that they are out of date by
-      // looking at their term : 0 compare to term of this core : 1
-      zkController
-          .getShardTerms(cd.getCollectionName(), cd.getShardId())
-          .ensureHighestTermsAreNotZero();
     }
 
     return response;

solr/core/src/java/org/apache/solr/handler/admin/api/InstallShardData.java

@@ -78,10 +78,10 @@ public SubResponseAccumulatingJerseyResponse installShardData(
     // Only install data to shards which belong to a collection in read-only mode
     final DocCollection dc =
         coreContainer.getZkController().getZkStateReader().getCollection(collName);
-    if (!dc.isReadOnly()) {
+    if (dc.getSlice(shardName).getReplicas().size() > 1 && !dc.isReadOnly()) {
       throw new SolrException(
           SolrException.ErrorCode.BAD_REQUEST,
-          "Collection must be in readOnly mode before installing data to shard");
+          "Collection must be in readOnly mode before installing data to shard with more than 1 replica");
     }
 
     submitRemoteMessageAndHandleResponse(
@@ -112,6 +112,8 @@ public static ZkNodeProps createRemoteMessage(
     if (requestBody != null) {
       messageTyped.location = requestBody.location;
       messageTyped.repository = requestBody.repository;
+      messageTyped.name = requestBody.name;
+      messageTyped.shardBackupId = requestBody.shardBackupId;
     }
 
     messageTyped.validate();

solr/core/src/java/org/apache/solr/handler/admin/api/RestoreCore.java

@@ -132,12 +132,6 @@ private void doRestore(String coreName, RestoreCoreRequestBody requestBody) thro
         throw new SolrException(
             SolrException.ErrorCode.SERVER_ERROR, "Failed to restore core=" + core.getName());
       }
-      // other replicas to-be-created will know that they are out of date by
-      // looking at their term : 0 compare to term of this core : 1
-      coreContainer
-          .getZkController()
-          .getShardTerms(cd.getCollectionName(), cd.getShardId())
-          .ensureHighestTermsAreNotZero();
 
       // transitions state of update log to ACTIVE
       UpdateLog updateLog = core.getUpdateHandler().getUpdateLog();

solr/core/src/java/org/apache/solr/handler/component/ShardRequest.java

@@ -60,6 +60,9 @@ public class ShardRequest {
   /** may be null */
   public String coreNodeName;
 
+  /** may be null */
+  public String coreName;
+
   /** may be null */
   public Map<String, String> headers;