diff --git a/changelog/unreleased/SOLR-18601-migrate-cleanup-on-failure.yml b/changelog/unreleased/SOLR-18601-migrate-cleanup-on-failure.yml new file mode 100644 index 000000000000..14fa95ee8f29 --- /dev/null +++ b/changelog/unreleased/SOLR-18601-migrate-cleanup-on-failure.yml @@ -0,0 +1,12 @@ +# See https://github.com/apache/solr/blob/main/dev-docs/changelog.adoc + +title: > + A MIGRATE that fails now removes its routing rule and drops the updates buffered on the target + leader, instead of leaving the target buffering updates so that retrying the MIGRATE fails +type: fixed +authors: + - name: Boyi Dai + nick: SpookyJumpyBeans +links: + - name: SOLR-18601 + url: https://issues.apache.org/jira/browse/SOLR-18601 diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/MigrateCmd.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/MigrateCmd.java index 2869f9866b6e..de32a1509d40 100644 --- a/solr/core/src/java/org/apache/solr/cloud/api/collections/MigrateCmd.java +++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/MigrateCmd.java @@ -237,244 +237,260 @@ private void migrateKey( shardRequestTracker.processResponses( results, shardHandler, true, "MIGRATE failed to request node to buffer updates"); } - ZkNodeProps m = - new ZkNodeProps( - Overseer.QUEUE_OPERATION, - OverseerAction.ADDROUTINGRULE.toLower(), - COLLECTION_PROP, - sourceCollection.getName(), - SHARD_ID_PROP, - sourceSlice.getName(), - "routeKey", - sourceRouter.getRouteKeyNoSuffix(splitKey) + "!", - "range", - splitRange.toString(), - "targetCollection", - targetCollection.getName(), - "expireAt", - RoutingRule.makeExpiryAt(timeout)); - log.info("Adding routing rule: {}", m); - if (ccc.getDistributedClusterStateUpdater().isDistributedStateUpdate()) { - ccc.getDistributedClusterStateUpdater() - .doSingleStateUpdate( - DistributedClusterStateUpdater.MutatingCommand.SliceAddRoutingRule, - m, - ccc.getSolrCloudManager(), - ccc.getZkStateReader()); - } else { - ccc.offerStateUpdate(m); - } - - // wait for a while until we see the new rule - log.info("Waiting to see routing rule updated in clusterstate"); - + // From here on, the target leader is buffering updates. If MIGRATE fails before it is asked + // to apply them, undo the routing rule and the buffering, or the target stays buffering. try { - sourceCollection = - zkStateReader.waitForState( + ZkNodeProps m = + new ZkNodeProps( + Overseer.QUEUE_OPERATION, + OverseerAction.ADDROUTINGRULE.toLower(), + COLLECTION_PROP, sourceCollection.getName(), - 60, - TimeUnit.SECONDS, - c -> { - Slice s = c.getSlice(sourceSlice.getName()); - Map rules = s.getRoutingRules(); - if (rules != null) { - RoutingRule rule = rules.get(sourceRouter.getRouteKeyNoSuffix(splitKey) + "!"); - return rule != null && rule.getRouteRanges().contains(splitRange); - } - return false; - }); - } catch (TimeoutException e) { - throw new SolrException( - SolrException.ErrorCode.SERVER_ERROR, "Could not add routing rule: " + m, e); - } - log.info("Routing rule added successfully"); - - // Create temp core on source shard - Replica sourceLeader = - zkStateReader.getLeaderRetry(sourceCollection.getName(), sourceSlice.getName(), 10000); - - // create a temporary collection with just one node on the shard leader - String configName = sourceCollection.getConfigName(); - Map props = - Utils.makeMap( - Overseer.QUEUE_OPERATION, - CREATE.toLower(), - NAME, - tempSourceCollectionName, - Replica.Type.defaultType().numReplicasPropertyName, - 1, - CollectionHandlingUtils.NUM_SLICES, - 1, - CollectionAdminParams.COLL_CONF, - configName, - CollectionHandlingUtils.CREATE_NODE_SET, - sourceLeader.getNodeName()); - String internalAsyncId = null; - if (adminCmdContext.getAsyncId() != null) { - internalAsyncId = adminCmdContext.getAsyncId() + Math.abs(System.nanoTime()); - } - - log.info("Creating temporary collection: {}", props); - new CreateCollectionCmd(ccc) - .call( - adminCmdContext.subRequestContext(CREATE, internalAsyncId), - new ZkNodeProps(props), - results); - // refresh cluster state - adminCmdContext.withClusterState(zkStateReader.getClusterState()); - Slice tempSourceSlice = - adminCmdContext - .getClusterState() - .getCollection(tempSourceCollectionName) - .getSlices() - .iterator() - .next(); - Replica tempSourceLeader = - zkStateReader.getLeaderRetry(tempSourceCollectionName, tempSourceSlice.getName(), 120000); - - String tempCollectionReplica1 = tempSourceLeader.getCoreName(); - String coreNodeName = - CollectionHandlingUtils.waitForCoreNodeName( - tempSourceCollectionName, - sourceLeader.getNodeName(), - tempCollectionReplica1, - ccc.getZkStateReader()); - // wait for the replicas to be seen as active on temp source leader - if (log.isInfoEnabled()) { - log.info( - "Asking source leader to wait for: {} to be alive on: {}", - tempCollectionReplica1, - sourceLeader.getNodeName()); - } - CoreAdminRequest.WaitForState cmd = new CoreAdminRequest.WaitForState(); - cmd.setCoreName(tempCollectionReplica1); - cmd.setNodeName(sourceLeader.getNodeName()); - cmd.setCoreNodeName(coreNodeName); - cmd.setState(Replica.State.ACTIVE); - cmd.setCheckLive(true); - cmd.setOnlyIfLeader(true); - { - final ShardRequestTracker syncRequestTracker = - CollectionHandlingUtils.syncRequestTracker(adminCmdContext, ccc); - // we don't want this to happen asynchronously - syncRequestTracker.sendShardRequest( - tempSourceLeader, new ModifiableSolrParams(cmd.getParams()), shardHandler); + SHARD_ID_PROP, + sourceSlice.getName(), + "routeKey", + sourceRouter.getRouteKeyNoSuffix(splitKey) + "!", + "range", + splitRange.toString(), + "targetCollection", + targetCollection.getName(), + "expireAt", + RoutingRule.makeExpiryAt(timeout)); + log.info("Adding routing rule: {}", m); + if (ccc.getDistributedClusterStateUpdater().isDistributedStateUpdate()) { + ccc.getDistributedClusterStateUpdater() + .doSingleStateUpdate( + DistributedClusterStateUpdater.MutatingCommand.SliceAddRoutingRule, + m, + ccc.getSolrCloudManager(), + ccc.getZkStateReader()); + } else { + ccc.offerStateUpdate(m); + } - syncRequestTracker.processResponses( - results, - shardHandler, - true, - "MIGRATE failed to create temp collection leader" - + " or timed out waiting for it to come up"); - } + // wait for a while until we see the new rule + log.info("Waiting to see routing rule updated in clusterstate"); - log.info("Asking source leader to split index"); - params = new ModifiableSolrParams(); - params.set(CoreAdminParams.ACTION, CoreAdminParams.CoreAdminAction.SPLIT.toString()); - params.set(CoreAdminParams.CORE, sourceLeader.getStr("core")); - params.add(CoreAdminParams.TARGET_CORE, tempSourceLeader.getStr("core")); - params.set(CoreAdminParams.RANGES, splitRange.toString()); - params.set("split.key", splitKey); + try { + sourceCollection = + zkStateReader.waitForState( + sourceCollection.getName(), + 60, + TimeUnit.SECONDS, + c -> { + Slice s = c.getSlice(sourceSlice.getName()); + Map rules = s.getRoutingRules(); + if (rules != null) { + RoutingRule rule = rules.get(sourceRouter.getRouteKeyNoSuffix(splitKey) + "!"); + return rule != null && rule.getRouteRanges().contains(splitRange); + } + return false; + }); + } catch (TimeoutException e) { + throw new SolrException( + SolrException.ErrorCode.SERVER_ERROR, "Could not add routing rule: " + m, e); + } + log.info("Routing rule added successfully"); + + // Create temp core on source shard + Replica sourceLeader = + zkStateReader.getLeaderRetry(sourceCollection.getName(), sourceSlice.getName(), 10000); + + // create a temporary collection with just one node on the shard leader + String configName = sourceCollection.getConfigName(); + Map props = + Utils.makeMap( + Overseer.QUEUE_OPERATION, + CREATE.toLower(), + NAME, + tempSourceCollectionName, + Replica.Type.defaultType().numReplicasPropertyName, + 1, + CollectionHandlingUtils.NUM_SLICES, + 1, + CollectionAdminParams.COLL_CONF, + configName, + CollectionHandlingUtils.CREATE_NODE_SET, + sourceLeader.getNodeName()); + String internalAsyncId = null; + if (adminCmdContext.getAsyncId() != null) { + internalAsyncId = adminCmdContext.getAsyncId() + Math.abs(System.nanoTime()); + } - { - final ShardRequestTracker shardRequestTracker = - CollectionHandlingUtils.asyncRequestTracker(adminCmdContext, ccc); - shardRequestTracker.sendShardRequest(sourceLeader, params, shardHandler); - shardRequestTracker.processResponses( - results, shardHandler, true, "MIGRATE failed to invoke SPLIT core admin command"); - } - if (log.isInfoEnabled()) { - log.info( - "Creating a replica of temporary collection: {} on the target leader node: {}", - tempSourceCollectionName, - targetLeader.getNodeName()); - } - String tempCollectionReplica2 = - Assign.buildSolrCoreName( - ccc.getSolrCloudManager().getDistribStateManager(), - zkStateReader.getClusterState().getCollection(tempSourceCollectionName), - tempSourceSlice.getName(), - Replica.Type.defaultType()); - props = new HashMap<>(); - props.put(COLLECTION_PROP, tempSourceCollectionName); - props.put(SHARD_ID_PROP, tempSourceSlice.getName()); - props.put("node", targetLeader.getNodeName()); - props.put(CoreAdminParams.NAME, tempCollectionReplica2); - // copy over property params: - for (String key : message.keySet()) { - if (key.startsWith(CollectionAdminParams.PROPERTY_PREFIX)) { - props.put(key, message.getStr(key)); + log.info("Creating temporary collection: {}", props); + new CreateCollectionCmd(ccc) + .call( + adminCmdContext.subRequestContext(CREATE, internalAsyncId), + new ZkNodeProps(props), + results); + // refresh cluster state + adminCmdContext.withClusterState(zkStateReader.getClusterState()); + Slice tempSourceSlice = + adminCmdContext + .getClusterState() + .getCollection(tempSourceCollectionName) + .getSlices() + .iterator() + .next(); + Replica tempSourceLeader = + zkStateReader.getLeaderRetry(tempSourceCollectionName, tempSourceSlice.getName(), 120000); + + String tempCollectionReplica1 = tempSourceLeader.getCoreName(); + String coreNodeName = + CollectionHandlingUtils.waitForCoreNodeName( + tempSourceCollectionName, + sourceLeader.getNodeName(), + tempCollectionReplica1, + ccc.getZkStateReader()); + // wait for the replicas to be seen as active on temp source leader + if (log.isInfoEnabled()) { + log.info( + "Asking source leader to wait for: {} to be alive on: {}", + tempCollectionReplica1, + sourceLeader.getNodeName()); + } + CoreAdminRequest.WaitForState cmd = new CoreAdminRequest.WaitForState(); + cmd.setCoreName(tempCollectionReplica1); + cmd.setNodeName(sourceLeader.getNodeName()); + cmd.setCoreNodeName(coreNodeName); + cmd.setState(Replica.State.ACTIVE); + cmd.setCheckLive(true); + cmd.setOnlyIfLeader(true); + { + final ShardRequestTracker syncRequestTracker = + CollectionHandlingUtils.syncRequestTracker(adminCmdContext, ccc); + // we don't want this to happen asynchronously + syncRequestTracker.sendShardRequest( + tempSourceLeader, new ModifiableSolrParams(cmd.getParams()), shardHandler); + + syncRequestTracker.processResponses( + results, + shardHandler, + true, + "MIGRATE failed to create temp collection leader" + + " or timed out waiting for it to come up"); } - } - // add async param - new AddReplicaCmd(ccc) - .addReplica( - adminCmdContext.subRequestContext(ADDREPLICA), new ZkNodeProps(props), results, null); - { - final ShardRequestTracker syncRequestTracker = - CollectionHandlingUtils.syncRequestTracker(adminCmdContext, ccc); - syncRequestTracker.processResponses( - results, - shardHandler, - true, - "MIGRATE failed to create replica of " + "temporary collection in target leader node."); - } - coreNodeName = - CollectionHandlingUtils.waitForCoreNodeName( + log.info("Asking source leader to split index"); + params = new ModifiableSolrParams(); + params.set(CoreAdminParams.ACTION, CoreAdminParams.CoreAdminAction.SPLIT.toString()); + params.set(CoreAdminParams.CORE, sourceLeader.getStr("core")); + params.add(CoreAdminParams.TARGET_CORE, tempSourceLeader.getStr("core")); + params.set(CoreAdminParams.RANGES, splitRange.toString()); + params.set("split.key", splitKey); + + { + final ShardRequestTracker shardRequestTracker = + CollectionHandlingUtils.asyncRequestTracker(adminCmdContext, ccc); + shardRequestTracker.sendShardRequest(sourceLeader, params, shardHandler); + shardRequestTracker.processResponses( + results, shardHandler, true, "MIGRATE failed to invoke SPLIT core admin command"); + } + if (log.isInfoEnabled()) { + log.info( + "Creating a replica of temporary collection: {} on the target leader node: {}", tempSourceCollectionName, - targetLeader.getNodeName(), + targetLeader.getNodeName()); + } + String tempCollectionReplica2 = + Assign.buildSolrCoreName( + ccc.getSolrCloudManager().getDistribStateManager(), + zkStateReader.getClusterState().getCollection(tempSourceCollectionName), + tempSourceSlice.getName(), + Replica.Type.defaultType()); + props = new HashMap<>(); + props.put(COLLECTION_PROP, tempSourceCollectionName); + props.put(SHARD_ID_PROP, tempSourceSlice.getName()); + props.put("node", targetLeader.getNodeName()); + props.put(CoreAdminParams.NAME, tempCollectionReplica2); + // copy over property params: + for (String key : message.keySet()) { + if (key.startsWith(CollectionAdminParams.PROPERTY_PREFIX)) { + props.put(key, message.getStr(key)); + } + } + // add async param + new AddReplicaCmd(ccc) + .addReplica( + adminCmdContext.subRequestContext(ADDREPLICA), new ZkNodeProps(props), results, null); + + { + final ShardRequestTracker syncRequestTracker = + CollectionHandlingUtils.syncRequestTracker(adminCmdContext, ccc); + syncRequestTracker.processResponses( + results, + shardHandler, + true, + "MIGRATE failed to create replica of " + "temporary collection in target leader node."); + } + coreNodeName = + CollectionHandlingUtils.waitForCoreNodeName( + tempSourceCollectionName, + targetLeader.getNodeName(), + tempCollectionReplica2, + ccc.getZkStateReader()); + // wait for the replicas to be seen as active on temp source leader + if (log.isInfoEnabled()) { + log.info( + "Asking temp source leader to wait for: {} to be alive on: {}", tempCollectionReplica2, - ccc.getZkStateReader()); - // wait for the replicas to be seen as active on temp source leader - if (log.isInfoEnabled()) { - log.info( - "Asking temp source leader to wait for: {} to be alive on: {}", - tempCollectionReplica2, - targetLeader.getNodeName()); - } - cmd = new CoreAdminRequest.WaitForState(); - cmd.setCoreName(tempSourceLeader.getStr("core")); - cmd.setNodeName(targetLeader.getNodeName()); - cmd.setCoreNodeName(coreNodeName); - cmd.setState(Replica.State.ACTIVE); - cmd.setCheckLive(true); - cmd.setOnlyIfLeader(true); - params = new ModifiableSolrParams(cmd.getParams()); - - { - final ShardRequestTracker shardRequestTracker = - CollectionHandlingUtils.asyncRequestTracker(adminCmdContext, ccc); - shardRequestTracker.sendShardRequest(tempSourceLeader, params, shardHandler); - - shardRequestTracker.processResponses( - results, - shardHandler, - true, - "MIGRATE failed to create temp collection" - + " replica or timed out waiting for them to come up"); - } - log.info("Successfully created replica of temp source collection on target leader node"); - - log.info("Requesting merge of temp source collection replica to target leader"); - params = new ModifiableSolrParams(); - params.set(CoreAdminParams.ACTION, CoreAdminParams.CoreAdminAction.MERGEINDEXES.toString()); - params.set(CoreAdminParams.CORE, targetLeader.getStr("core")); - params.set(CoreAdminParams.SRC_CORE, tempCollectionReplica2); - - { - final ShardRequestTracker shardRequestTracker = - CollectionHandlingUtils.asyncRequestTracker(adminCmdContext, ccc); - - shardRequestTracker.sendShardRequest(targetLeader, params, shardHandler); - String msg = - "MIGRATE failed to merge " - + tempCollectionReplica2 - + " to " - + targetLeader.getStr("core") - + " on node: " - + targetLeader.getNodeName(); - shardRequestTracker.processResponses(results, shardHandler, true, msg); + targetLeader.getNodeName()); + } + cmd = new CoreAdminRequest.WaitForState(); + cmd.setCoreName(tempSourceLeader.getStr("core")); + cmd.setNodeName(targetLeader.getNodeName()); + cmd.setCoreNodeName(coreNodeName); + cmd.setState(Replica.State.ACTIVE); + cmd.setCheckLive(true); + cmd.setOnlyIfLeader(true); + params = new ModifiableSolrParams(cmd.getParams()); + + { + final ShardRequestTracker shardRequestTracker = + CollectionHandlingUtils.asyncRequestTracker(adminCmdContext, ccc); + shardRequestTracker.sendShardRequest(tempSourceLeader, params, shardHandler); + + shardRequestTracker.processResponses( + results, + shardHandler, + true, + "MIGRATE failed to create temp collection" + + " replica or timed out waiting for them to come up"); + } + log.info("Successfully created replica of temp source collection on target leader node"); + + log.info("Requesting merge of temp source collection replica to target leader"); + params = new ModifiableSolrParams(); + params.set(CoreAdminParams.ACTION, CoreAdminParams.CoreAdminAction.MERGEINDEXES.toString()); + params.set(CoreAdminParams.CORE, targetLeader.getStr("core")); + params.set(CoreAdminParams.SRC_CORE, tempCollectionReplica2); + + { + final ShardRequestTracker shardRequestTracker = + CollectionHandlingUtils.asyncRequestTracker(adminCmdContext, ccc); + + shardRequestTracker.sendShardRequest(targetLeader, params, shardHandler); + String msg = + "MIGRATE failed to merge " + + tempCollectionReplica2 + + " to " + + targetLeader.getStr("core") + + " on node: " + + targetLeader.getNodeName(); + shardRequestTracker.processResponses(results, shardHandler, true, msg); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw e; + } catch (Exception e) { + cleanupFailedMigrate( + adminCmdContext, + sourceCollection, + sourceSlice, + sourceRouter.getRouteKeyNoSuffix(splitKey) + "!", + targetLeader, + e); + throw e; } log.info("Asking target leader to apply buffered updates"); params = new ModifiableSolrParams(); @@ -506,6 +522,92 @@ private void migrateKey( } } + /** + * Undoes what a failed MIGRATE did before it asked the target leader to apply its buffered + * updates: removes the routing rule from the source shard, then drops the target leader's + * buffered updates, so that the target leaves the buffering state and a retry can start again. + * The source shard still has the documents that were forwarded. Errors are added to the failure + * as suppressed exceptions. + */ + private void cleanupFailedMigrate( + AdminCmdContext adminCmdContext, + DocCollection sourceCollection, + Slice sourceSlice, + String routeKey, + Replica targetLeader, + Exception failure) { + try { + ZkNodeProps m = + new ZkNodeProps( + Overseer.QUEUE_OPERATION, + OverseerAction.REMOVEROUTINGRULE.toLower(), + COLLECTION_PROP, + sourceCollection.getName(), + SHARD_ID_PROP, + sourceSlice.getName(), + "routeKey", + routeKey); + log.info("Removing routing rule of failed MIGRATE: {}", m); + if (ccc.getDistributedClusterStateUpdater().isDistributedStateUpdate()) { + ccc.getDistributedClusterStateUpdater() + .doSingleStateUpdate( + DistributedClusterStateUpdater.MutatingCommand.SliceRemoveRoutingRule, + m, + ccc.getSolrCloudManager(), + ccc.getZkStateReader()); + } else { + ccc.offerStateUpdate(m); + } + // Updates forwarded until the rule is gone would be indexed by the target once it is no + // longer buffering, so wait for the removal before dropping the buffered updates. + ccc.getZkStateReader() + .waitForState( + sourceCollection.getName(), + 60, + TimeUnit.SECONDS, + c -> { + if (c == null || c.getSlice(sourceSlice.getName()) == null) return true; + Map rules = + c.getSlice(sourceSlice.getName()).getRoutingRules(); + return rules == null || !rules.containsKey(routeKey); + }); + } catch (Exception e) { + addCleanupFailure(failure, e); + } + + try { + if (log.isInfoEnabled()) { + log.info( + "Asking target leader node: {} core: {} to drop buffered updates", + targetLeader.getNodeName(), + targetLeader.getStr("core")); + } + ModifiableSolrParams params = new ModifiableSolrParams(); + params.set( + CoreAdminParams.ACTION, + CoreAdminParams.CoreAdminAction.REQUESTDROPBUFFEREDUPDATES.toString()); + params.set(CoreAdminParams.NAME, targetLeader.getStr("core")); + ShardHandler shardHandler = ccc.newShardHandler(); + ShardRequestTracker shardRequestTracker = + CollectionHandlingUtils.syncRequestTracker(adminCmdContext, ccc); + shardRequestTracker.sendShardRequest(targetLeader, params, shardHandler); + shardRequestTracker.processResponses( + new NamedList<>(), + shardHandler, + true, + "MIGRATE failed to request node to drop buffered updates"); + } catch (Exception e) { + addCleanupFailure(failure, e); + } + } + + private static void addCleanupFailure(Exception failure, Exception cleanupFailure) { + if (cleanupFailure instanceof InterruptedException) { + Thread.currentThread().interrupt(); + } + failure.addSuppressed(cleanupFailure); + } + DocRouter.Range intersect(DocRouter.Range a, DocRouter.Range b) { if (a == null || b == null || !a.overlaps(b)) { return null; diff --git a/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminOperation.java b/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminOperation.java index 2edc70aef30b..c3c304eabf72 100644 --- a/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminOperation.java +++ b/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminOperation.java @@ -30,6 +30,7 @@ import static org.apache.solr.common.params.CoreAdminParams.CoreAdminAction.RENAME; import static org.apache.solr.common.params.CoreAdminParams.CoreAdminAction.REQUESTAPPLYUPDATES; import static org.apache.solr.common.params.CoreAdminParams.CoreAdminAction.REQUESTBUFFERUPDATES; +import static org.apache.solr.common.params.CoreAdminParams.CoreAdminAction.REQUESTDROPBUFFEREDUPDATES; import static org.apache.solr.common.params.CoreAdminParams.CoreAdminAction.REQUESTRECOVERY; import static org.apache.solr.common.params.CoreAdminParams.CoreAdminAction.REQUESTSTATUS; import static org.apache.solr.common.params.CoreAdminParams.CoreAdminAction.REQUESTSYNCSHARD; @@ -196,6 +197,31 @@ public enum CoreAdminOperation implements CoreAdminOp { }), REQUESTAPPLYUPDATES_OP(REQUESTAPPLYUPDATES, new RequestApplyUpdatesOp()), + REQUESTDROPBUFFEREDUPDATES_OP( + REQUESTDROPBUFFEREDUPDATES, + it -> { + SolrParams params = it.req.getParams(); + String cname = params.required().get(CoreAdminParams.NAME); + log().info("Dropping buffered updates on core: {}", cname); + + try (SolrCore core = it.handler.coreContainer.getCore(cname)) { + if (core == null) + throw new SolrException(ErrorCode.BAD_REQUEST, "Core [" + cname + "] does not exist"); + if (!core.getUpdateHandler().getUpdateLog().dropBufferedUpdates()) { + throw new SolrException( + ErrorCode.SERVER_ERROR, "Core " + cname + " not in buffering state"); + } + it.rsp.add("core", cname); + it.rsp.add("status", "BUFFER_DROPPED"); + } catch (Exception e) { + if (e instanceof SolrException) throw (SolrException) e; + else + throw new SolrException(ErrorCode.SERVER_ERROR, "Could not drop buffered updates", e); + } finally { + if (it.req != null) it.req.close(); + } + }), + REQUESTSTATUS_OP( REQUESTSTATUS, it -> { diff --git a/solr/core/src/test/org/apache/solr/cloud/MigrateRouteKeyTest.java b/solr/core/src/test/org/apache/solr/cloud/MigrateRouteKeyTest.java index c0b1ce96d67a..2eff00cbb697 100644 --- a/solr/core/src/test/org/apache/solr/cloud/MigrateRouteKeyTest.java +++ b/solr/core/src/test/org/apache/solr/cloud/MigrateRouteKeyTest.java @@ -27,12 +27,15 @@ import org.apache.solr.client.solrj.request.CollectionAdminRequest; import org.apache.solr.client.solrj.request.SolrQuery; import org.apache.solr.client.solrj.response.QueryResponse; +import org.apache.solr.client.solrj.response.RequestStatusState; import org.apache.solr.common.SolrInputDocument; import org.apache.solr.common.cloud.DocCollection; import org.apache.solr.common.cloud.Replica; import org.apache.solr.common.cloud.RoutingRule; import org.apache.solr.common.cloud.Slice; import org.apache.solr.common.util.TimeSource; +import org.apache.solr.core.SolrCore; +import org.apache.solr.update.UpdateLog; import org.apache.solr.util.TimeOut; import org.apache.zookeeper.KeeperException; import org.junit.BeforeClass; @@ -152,6 +155,79 @@ public void updateSucceedsAfterMigrateTargetIsDeleted() throws Exception { }); } + @Test + public void testRetryAfterFailedMigrate() throws Exception { + String sourceCollection = "failedMigrate-source"; + CollectionAdminRequest.createCollection(sourceCollection, "conf", 1, 1) + .process(cluster.getSolrClient()); + String targetCollection = "failedMigrate-target"; + CollectionAdminRequest.createCollection(targetCollection, "conf", 1, 1) + .process(cluster.getSolrClient()); + + cluster.getSolrClient().add(sourceCollection, new SolrInputDocument("id", "a!1")); + cluster.getSolrClient().add(sourceCollection, new SolrInputDocument("id", "a!2")); + cluster.getSolrClient().add(sourceCollection, new SolrInputDocument("id", "b!3")); + cluster.getSolrClient().commit(sourceCollection); + + // MIGRATE deletes a leftover temporary collection before it starts, but can't delete one that + // an alias refers to. Creating the temporary collection then fails, after the target leader + // has started buffering updates. + String tempCollection = "split_shard1_temp_shard1"; + CollectionAdminRequest.createCollection(tempCollection, "conf", 1, 1) + .process(cluster.getSolrClient()); + CollectionAdminRequest.createAlias("failedMigrate-blocker", tempCollection) + .process(cluster.getSolrClient()); + RemoteSolrException e = + expectThrows( + RemoteSolrException.class, + () -> + CollectionAdminRequest.migrateData(sourceCollection, targetCollection, "a!") + .setForwardTimeout(45) + .process(cluster.getSolrClient())); + assertTrue(e.getMessage(), e.getMessage().contains("collection already exists")); + assertFailedMigrateUndone(sourceCollection, targetCollection); + + assertEquals( + RequestStatusState.FAILED, + CollectionAdminRequest.migrateData(sourceCollection, targetCollection, "a!") + .setForwardTimeout(45) + .processAndWait(cluster.getSolrClient(), 60)); + assertFailedMigrateUndone(sourceCollection, targetCollection); + + CollectionAdminRequest.deleteAlias("failedMigrate-blocker").process(cluster.getSolrClient()); + CollectionAdminRequest.migrateData(sourceCollection, targetCollection, "a!") + .setForwardTimeout(45) + .process(cluster.getSolrClient()); + cluster.getSolrClient().commit(targetCollection); + assertEquals( + 2, + cluster + .getSolrClient() + .query(targetCollection, new SolrQuery("*:*")) + .getResults() + .getNumFound()); + } + + /** Checks that the target leader isn't buffering updates and the routing rule is gone. */ + private void assertFailedMigrateUndone(String sourceCollection, String targetCollection) + throws Exception { + Replica targetLeader = getCollectionState(targetCollection).getLeader("shard1"); + try (SolrCore core = + cluster + .getReplicaJetty(targetLeader) + .getCoreContainer() + .getCore(targetLeader.getCoreName())) { + assertEquals(UpdateLog.State.ACTIVE, core.getUpdateHandler().getUpdateLog().getState()); + } + waitForState( + "Expected the routing rule of the failed MIGRATE to be removed", + sourceCollection, + c -> { + Map routingRules = c.getSlice("shard1").getRoutingRules(); + return routingRules == null || !routingRules.containsKey("a!"); + }); + } + @Test public void multipleShardMigrateTest() throws Exception { diff --git a/solr/core/src/test/org/apache/solr/handler/admin/CoreAdminOperationTest.java b/solr/core/src/test/org/apache/solr/handler/admin/CoreAdminOperationTest.java index 97f40bdecc9e..d224db1cc18e 100644 --- a/solr/core/src/test/org/apache/solr/handler/admin/CoreAdminOperationTest.java +++ b/solr/core/src/test/org/apache/solr/handler/admin/CoreAdminOperationTest.java @@ -323,6 +323,29 @@ public void testRequestApplyUpdatesMissingCoreParamResultsIn400SolrException() { assertSolrExceptionWithCode(ex, ErrorCode.BAD_REQUEST.code); } + @Test + public void testRequestDropBufferedUpdatesUnexpectedFailuresResultIn500Exception() { + final Throwable cause = new NullPointerException(); + whenUnexpectedErrorOccursDuringCoreAdminOp(cause); + + Exception ex = + expectThrows( + Exception.class, + () -> CoreAdminOperation.REQUESTDROPBUFFEREDUPDATES_OP.execute(callInfo)); + assertSolrExceptionWithCodeAndCause(ex, ErrorCode.SERVER_ERROR.code, cause); + } + + @Test + public void testRequestDropBufferedUpdatesMissingCoreParamResultsIn400SolrException() { + whenCoreAdminOpHasParams(new HashMap<>()); + + Exception ex = + expectThrows( + Exception.class, + () -> CoreAdminOperation.REQUESTDROPBUFFEREDUPDATES_OP.execute(callInfo)); + assertSolrExceptionWithCode(ex, ErrorCode.BAD_REQUEST.code); + } + @Test public void testOverseerOpUnexpectedFailuresResultIn500Exception() { final Throwable cause = new NullPointerException(); diff --git a/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java b/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java index 0401de380f1b..736bbec7d142 100644 --- a/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java +++ b/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java @@ -169,6 +169,7 @@ public enum CoreAdminAction { REQUESTSYNCSHARD, REQUESTBUFFERUPDATES, REQUESTAPPLYUPDATES, + REQUESTDROPBUFFEREDUPDATES, OVERSEEROP, REQUESTSTATUS(true), REJOINLEADERELECTION,