| /* |
| * 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 java.lang.invoke.MethodHandles; |
| import java.nio.file.Files; |
| import java.nio.file.Path; |
| import java.nio.file.Paths; |
| import java.util.EnumSet; |
| import java.util.Set; |
| import java.util.concurrent.Semaphore; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.TimeoutException; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicInteger; |
| |
| import org.apache.solr.client.solrj.embedded.JettySolrRunner; |
| import org.apache.solr.client.solrj.request.CollectionAdminRequest; |
| import org.apache.solr.client.solrj.request.CollectionAdminRequest.Create; |
| import org.apache.solr.client.solrj.request.CoreStatus; |
| import org.apache.solr.cloud.overseer.OverseerAction; |
| import org.apache.solr.common.SolrException; |
| import org.apache.solr.common.SolrInputDocument; |
| import org.apache.solr.common.cloud.DocCollection; |
| import org.apache.solr.common.cloud.DocCollectionWatcher; |
| 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.cloud.ZkStateReader; |
| import org.apache.solr.common.cloud.ZkStateReaderAccessor; |
| import org.apache.solr.common.util.TimeSource; |
| import org.apache.solr.common.util.Utils; |
| import org.apache.solr.core.ZkContainer; |
| import org.apache.solr.util.TimeOut; |
| import org.junit.After; |
| import org.junit.Before; |
| import org.junit.BeforeClass; |
| import org.junit.Test; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import static org.apache.solr.common.cloud.Replica.State.DOWN; |
| |
| |
| public class DeleteReplicaTest extends SolrCloudTestCase { |
| |
| private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); |
| |
| @BeforeClass |
| public static void setupCluster() throws Exception { |
| System.setProperty("solr.zkclienttimeout", "45000"); |
| System.setProperty("distribUpdateSoTimeout", "15000"); |
| |
| } |
| |
| @Before |
| @Override |
| public void setUp() throws Exception { |
| super.setUp(); |
| System.setProperty("solr.zkclienttimeout", "45000"); |
| System.setProperty("distribUpdateSoTimeout", "15000"); |
| |
| // these tests need to be isolated, so we dont share the minicluster |
| configureCluster(4) |
| .addConfig("conf", configset("cloud-minimal")) |
| .configure(); |
| } |
| |
| @After |
| @Override |
| public void tearDown() throws Exception { |
| shutdownCluster(); |
| super.tearDown(); |
| } |
| |
| @Test |
| // commented out on: 01-Apr-2019 @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // annotated on: 24-Dec-2018 |
| public void deleteLiveReplicaTest() throws Exception { |
| |
| final String collectionName = "delLiveColl"; |
| |
| Create req = CollectionAdminRequest.createCollection(collectionName, "conf", 2, 2); |
| req.process(cluster.getSolrClient()); |
| |
| cluster.waitForActiveCollection(collectionName, 2, 4); |
| |
| DocCollection state = getCollectionState(collectionName); |
| Slice shard = getRandomShard(state); |
| |
| // don't choose the leader to shutdown, it just complicates things unneccessarily |
| Replica replica = getRandomReplica(shard, (r) -> |
| ( r.getState() == Replica.State.ACTIVE && |
| ! r.equals(shard.getLeader()))); |
| |
| CoreStatus coreStatus = getCoreStatus(replica); |
| Path dataDir = Paths.get(coreStatus.getDataDirectory()); |
| |
| Exception e = expectThrows(Exception.class, () -> { |
| CollectionAdminRequest.deleteReplica(collectionName, shard.getName(), replica.getName()) |
| .setOnlyIfDown(true) |
| .process(cluster.getSolrClient()); |
| }); |
| assertTrue("Unexpected error message: " + e.getMessage(), e.getMessage().contains("state is 'active'")); |
| assertTrue("Data directory for " + replica.getName() + " should not have been deleted", Files.exists(dataDir)); |
| |
| JettySolrRunner replicaJetty = cluster.getReplicaJetty(replica); |
| ZkStateReaderAccessor accessor = new ZkStateReaderAccessor(replicaJetty.getCoreContainer().getZkController().getZkStateReader()); |
| |
| final long preDeleteWatcherCount = countUnloadCoreOnDeletedWatchers |
| (accessor.getStateWatchers(collectionName)); |
| |
| CollectionAdminRequest.deleteReplica(collectionName, shard.getName(), replica.getName()) |
| .process(cluster.getSolrClient()); |
| waitForState("Expected replica " + replica.getName() + " to have been removed", collectionName, (n, c) -> { |
| Slice testShard = c.getSlice(shard.getName()); |
| return testShard.getReplica(replica.getName()) == null; |
| }); |
| |
| // the core should no longer have a watch collection state since it was removed |
| // the core should no longer have a watch collection state since it was removed |
| TimeOut timeOut = new TimeOut(60, TimeUnit.SECONDS, TimeSource.NANO_TIME); |
| timeOut.waitFor("Waiting for core's watcher to be removed", () -> { |
| final long postDeleteWatcherCount = countUnloadCoreOnDeletedWatchers |
| (accessor.getStateWatchers(collectionName)); |
| log.info("preDeleteWatcherCount={} vs postDeleteWatcherCount={}", |
| preDeleteWatcherCount, postDeleteWatcherCount); |
| return (preDeleteWatcherCount - 1L == postDeleteWatcherCount); |
| }); |
| |
| assertFalse("Data directory for " + replica.getName() + " should have been removed", Files.exists(dataDir)); |
| |
| } |
| |
| @Test |
| public void deleteReplicaAndVerifyDirectoryCleanup() throws Exception { |
| |
| final String collectionName = "deletereplica_test"; |
| CollectionAdminRequest.createCollection(collectionName, "conf", 1, 2).process(cluster.getSolrClient()); |
| |
| Replica leader = cluster.getSolrClient().getZkStateReader().getLeaderRetry(collectionName, "shard1"); |
| |
| //Confirm that the instance and data directory exist |
| CoreStatus coreStatus = getCoreStatus(leader); |
| assertTrue("Instance directory doesn't exist", Files.exists(Paths.get(coreStatus.getInstanceDirectory()))); |
| assertTrue("DataDirectory doesn't exist", Files.exists(Paths.get(coreStatus.getDataDirectory()))); |
| |
| CollectionAdminRequest.deleteReplica(collectionName, "shard1",leader.getName()) |
| .process(cluster.getSolrClient()); |
| |
| Replica newLeader = cluster.getSolrClient().getZkStateReader().getLeaderRetry(collectionName, "shard1"); |
| |
| assertFalse(leader.equals(newLeader)); |
| |
| //Confirm that the instance and data directory were deleted by default |
| assertFalse("Instance directory still exists", Files.exists(Paths.get(coreStatus.getInstanceDirectory()))); |
| assertFalse("DataDirectory still exists", Files.exists(Paths.get(coreStatus.getDataDirectory()))); |
| } |
| |
| @Test |
| public void deleteReplicaByCount() throws Exception { |
| |
| final String collectionName = "deleteByCount"; |
| |
| CollectionAdminRequest.createCollection(collectionName, "conf", 1, 3).process(cluster.getSolrClient()); |
| waitForState("Expected a single shard with three replicas", collectionName, clusterShape(1, 3)); |
| |
| CollectionAdminRequest.deleteReplicasFromShard(collectionName, "shard1", 2).process(cluster.getSolrClient()); |
| waitForState("Expected a single shard with a single replica", collectionName, clusterShape(1, 1)); |
| |
| SolrException e = expectThrows(SolrException.class, |
| "Can't delete the last replica by count", |
| () -> CollectionAdminRequest.deleteReplicasFromShard(collectionName, "shard1", 1).process(cluster.getSolrClient()) |
| ); |
| assertEquals(SolrException.ErrorCode.BAD_REQUEST.code, e.code()); |
| assertTrue(e.getMessage().contains("There is only one replica available")); |
| DocCollection docCollection = getCollectionState(collectionName); |
| // We know that since leaders are preserved, PULL replicas should not be left alone in the shard |
| assertEquals(0, docCollection.getSlice("shard1").getReplicas(EnumSet.of(Replica.Type.PULL)).size()); |
| } |
| |
| @Test |
| // commented out on: 17-Feb-2019 @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // annotated on: 24-Dec-2018 |
| public void deleteReplicaByCountForAllShards() throws Exception { |
| |
| final String collectionName = "deleteByCountNew"; |
| Create req = CollectionAdminRequest.createCollection(collectionName, "conf", 2, 2); |
| req.process(cluster.getSolrClient()); |
| |
| cluster.waitForActiveCollection(collectionName, 2, 4); |
| |
| waitForState("Expected two shards with two replicas each", collectionName, clusterShape(2, 4)); |
| |
| CollectionAdminRequest.deleteReplicasFromAllShards(collectionName, 1).process(cluster.getSolrClient()); |
| waitForState("Expected two shards with one replica each", collectionName, clusterShape(2, 2)); |
| |
| } |
| |
| @Test |
| public void deleteReplicaFromClusterState() throws Exception { |
| deleteReplicaFromClusterState("false"); |
| CollectionAdminRequest.setClusterProperty(ZkStateReader.LEGACY_CLOUD, null).process(cluster.getSolrClient()); |
| } |
| |
| @Test |
| public void deleteReplicaFromClusterStateLegacy() throws Exception { |
| deleteReplicaFromClusterState("true"); |
| CollectionAdminRequest.setClusterProperty(ZkStateReader.LEGACY_CLOUD, null).process(cluster.getSolrClient()); |
| } |
| |
| private void deleteReplicaFromClusterState(String legacyCloud) throws Exception { |
| CollectionAdminRequest.setClusterProperty(ZkStateReader.LEGACY_CLOUD, legacyCloud).process(cluster.getSolrClient()); |
| final String collectionName = "deleteFromClusterState_"+legacyCloud; |
| CollectionAdminRequest.createCollection(collectionName, "conf", 1, 3) |
| .process(cluster.getSolrClient()); |
| |
| cluster.waitForActiveCollection(collectionName, 1, 3); |
| |
| cluster.getSolrClient().add(collectionName, new SolrInputDocument("id", "1")); |
| cluster.getSolrClient().add(collectionName, new SolrInputDocument("id", "2")); |
| cluster.getSolrClient().commit(collectionName); |
| |
| cluster.waitForActiveCollection(collectionName, 1, 3); |
| |
| Slice shard = getCollectionState(collectionName).getSlice("shard1"); |
| |
| // don't choose the leader to shutdown, it just complicates things unneccessarily |
| Replica replica = getRandomReplica(shard, (r) -> |
| ( r.getState() == Replica.State.ACTIVE && |
| ! r.equals(shard.getLeader()))); |
| |
| JettySolrRunner replicaJetty = cluster.getReplicaJetty(replica); |
| ZkStateReaderAccessor accessor = new ZkStateReaderAccessor(replicaJetty.getCoreContainer().getZkController().getZkStateReader()); |
| |
| final long preDeleteWatcherCount = countUnloadCoreOnDeletedWatchers |
| (accessor.getStateWatchers(collectionName)); |
| |
| ZkNodeProps m = new ZkNodeProps( |
| Overseer.QUEUE_OPERATION, OverseerAction.DELETECORE.toLower(), |
| ZkStateReader.CORE_NAME_PROP, replica.getCoreName(), |
| ZkStateReader.NODE_NAME_PROP, replica.getNodeName(), |
| ZkStateReader.COLLECTION_PROP, collectionName, |
| ZkStateReader.CORE_NODE_NAME_PROP, replica.getName()); |
| |
| cluster.getOpenOverseer().getStateUpdateQueue().offer(Utils.toJSON(m)); |
| |
| waitForState("Timeout waiting for replica get deleted", collectionName, |
| (liveNodes, collectionState) -> collectionState.getSlice("shard1").getReplicas().size() == 2); |
| |
| TimeOut timeOut = new TimeOut(60, TimeUnit.SECONDS, TimeSource.NANO_TIME); |
| timeOut.waitFor("Waiting for replica get unloaded", () -> |
| replicaJetty.getCoreContainer().getCoreDescriptor(replica.getCoreName()) == null |
| ); |
| |
| // the core should no longer have a watch collection state since it was removed |
| timeOut = new TimeOut(60, TimeUnit.SECONDS, TimeSource.NANO_TIME); |
| timeOut.waitFor("Waiting for core's watcher to be removed", () -> { |
| final long postDeleteWatcherCount = countUnloadCoreOnDeletedWatchers |
| (accessor.getStateWatchers(collectionName)); |
| log.info("preDeleteWatcherCount={} vs postDeleteWatcherCount={}", |
| preDeleteWatcherCount, postDeleteWatcherCount); |
| return (preDeleteWatcherCount - 1L == postDeleteWatcherCount); |
| }); |
| |
| CollectionAdminRequest.deleteCollection(collectionName).process(cluster.getSolrClient()); |
| } |
| |
| @Test |
| @Slow |
| // commented out on: 17-Feb-2019 @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // annotated on: 24-Dec-2018 |
| public void raceConditionOnDeleteAndRegisterReplica() throws Exception { |
| raceConditionOnDeleteAndRegisterReplica("false"); |
| CollectionAdminRequest.setClusterProperty(ZkStateReader.LEGACY_CLOUD, null).process(cluster.getSolrClient()); |
| } |
| |
| @Test |
| @Slow |
| // commented out on: 17-Feb-2019 @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // annotated on: 24-Dec-2018 |
| public void raceConditionOnDeleteAndRegisterReplicaLegacy() throws Exception { |
| raceConditionOnDeleteAndRegisterReplica("true"); |
| CollectionAdminRequest.setClusterProperty(ZkStateReader.LEGACY_CLOUD, null).process(cluster.getSolrClient()); |
| } |
| |
| // commented out on: 17-Feb-2019 @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // annotated on: 24-Dec-2018 |
| public void raceConditionOnDeleteAndRegisterReplica(String legacyCloud) throws Exception { |
| |
| CollectionAdminRequest.setClusterProperty(ZkStateReader.LEGACY_CLOUD, legacyCloud).process(cluster.getSolrClient()); |
| final String collectionName = "raceDeleteReplica_"+legacyCloud; |
| CollectionAdminRequest.createCollection(collectionName, "conf", 1, 2) |
| .process(cluster.getSolrClient()); |
| |
| cluster.waitForActiveCollection(collectionName, 1, 2); |
| |
| waitForState("Expected 1x2 collections", collectionName, clusterShape(1, 2)); |
| |
| Slice shard1 = getCollectionState(collectionName).getSlice("shard1"); |
| Replica leader = shard1.getLeader(); |
| JettySolrRunner leaderJetty = getJettyForReplica(leader); |
| Replica replica1 = shard1.getReplicas(replica -> !replica.getName().equals(leader.getName())).get(0); |
| assertFalse(replica1.getName().equals(leader.getName())); |
| |
| JettySolrRunner replica1Jetty = getJettyForReplica(replica1); |
| |
| String replica1JettyNodeName = replica1Jetty.getNodeName(); |
| |
| Semaphore waitingForReplicaGetDeleted = new Semaphore(0); |
| // for safety, we only want this hook get triggered one time |
| AtomicInteger times = new AtomicInteger(0); |
| ZkContainer.testing_beforeRegisterInZk = cd -> { |
| if (cd.getCloudDescriptor() == null) return false; |
| if (replica1.getName().equals(cd.getCloudDescriptor().getCoreNodeName()) |
| && collectionName.equals(cd.getCloudDescriptor().getCollectionName())) { |
| if (times.incrementAndGet() > 1) { |
| return false; |
| } |
| log.info("Running delete core {}",cd); |
| |
| try { |
| ZkNodeProps m = new ZkNodeProps( |
| Overseer.QUEUE_OPERATION, OverseerAction.DELETECORE.toLower(), |
| ZkStateReader.CORE_NAME_PROP, replica1.getCoreName(), |
| ZkStateReader.NODE_NAME_PROP, replica1.getNodeName(), |
| ZkStateReader.COLLECTION_PROP, collectionName, |
| ZkStateReader.CORE_NODE_NAME_PROP, replica1.getName()); |
| cluster.getOpenOverseer().getStateUpdateQueue().offer(Utils.toJSON(m)); |
| |
| boolean replicaDeleted = false; |
| TimeOut timeOut = new TimeOut(20, TimeUnit.SECONDS, TimeSource.NANO_TIME); |
| while (!timeOut.hasTimedOut()) { |
| try { |
| ZkStateReader stateReader = replica1Jetty.getCoreContainer().getZkController().getZkStateReader(); |
| stateReader.forceUpdateCollection(collectionName); |
| Slice shard = stateReader.getClusterState().getCollection(collectionName).getSlice("shard1"); |
| if (shard.getReplicas().size() == 1) { |
| replicaDeleted = true; |
| waitingForReplicaGetDeleted.release(); |
| break; |
| } |
| Thread.sleep(500); |
| } catch (NullPointerException | SolrException e) { |
| e.printStackTrace(); |
| Thread.sleep(500); |
| } |
| } |
| if (!replicaDeleted) { |
| fail("Timeout for waiting replica get deleted"); |
| } |
| } catch (Exception e) { |
| e.printStackTrace(); |
| fail("Failed to delete replica"); |
| } finally { |
| //avoiding deadlock |
| waitingForReplicaGetDeleted.release(); |
| } |
| return true; |
| } |
| return false; |
| }; |
| |
| try { |
| replica1Jetty.stop(); |
| waitForNodeLeave(replica1JettyNodeName); |
| waitForState("Expected replica:"+replica1+" get down", collectionName, (liveNodes, collectionState) |
| -> collectionState.getSlice("shard1").getReplica(replica1.getName()).getState() == DOWN); |
| replica1Jetty.start(); |
| waitingForReplicaGetDeleted.acquire(); |
| } finally { |
| ZkContainer.testing_beforeRegisterInZk = null; |
| } |
| |
| TimeOut timeOut = new TimeOut(30, TimeUnit.SECONDS, TimeSource.NANO_TIME); |
| timeOut.waitFor("Timeout adding replica to shard", () -> { |
| try { |
| CollectionAdminRequest.addReplicaToShard(collectionName, "shard1") |
| .process(cluster.getSolrClient()); |
| return true; |
| } catch (Exception e) { |
| // expected, when the node is not fully started |
| return false; |
| } |
| }); |
| waitForState("Expected 1x2 collections", collectionName, clusterShape(1, 2)); |
| |
| shard1 = getCollectionState(collectionName).getSlice("shard1"); |
| Replica latestLeader = shard1.getLeader(); |
| leaderJetty = getJettyForReplica(latestLeader); |
| String leaderJettyNodeName = leaderJetty.getNodeName(); |
| leaderJetty.stop(); |
| waitForNodeLeave(leaderJettyNodeName); |
| |
| waitForState("Expected new active leader", collectionName, (liveNodes, collectionState) -> { |
| Slice shard = collectionState.getSlice("shard1"); |
| Replica newLeader = shard.getLeader(); |
| return newLeader != null && newLeader.getState() == Replica.State.ACTIVE && !newLeader.getName().equals(latestLeader.getName()); |
| }); |
| |
| leaderJetty.start(); |
| cluster.waitForActiveCollection(collectionName, 1, 2); |
| |
| CollectionAdminRequest.deleteCollection(collectionName).process(cluster.getSolrClient()); |
| } |
| |
| private JettySolrRunner getJettyForReplica(Replica replica) { |
| for (JettySolrRunner jetty : cluster.getJettySolrRunners()) { |
| String nodeName = jetty.getNodeName(); |
| if (nodeName != null && nodeName.equals(replica.getNodeName())) return jetty; |
| } |
| throw new IllegalArgumentException("Can not find jetty for replica "+ replica); |
| } |
| |
| |
| private void waitForNodeLeave(String lostNodeName) throws InterruptedException { |
| ZkStateReader reader = cluster.getSolrClient().getZkStateReader(); |
| TimeOut timeOut = new TimeOut(20, TimeUnit.SECONDS, TimeSource.NANO_TIME); |
| while (reader.getClusterState().getLiveNodes().contains(lostNodeName)) { |
| Thread.sleep(100); |
| if (timeOut.hasTimedOut()) fail("Wait for " + lostNodeName + " to leave failed!"); |
| } |
| } |
| |
| @Test |
| public void deleteReplicaOnIndexing() throws Exception { |
| final String collectionName = "deleteReplicaOnIndexing"; |
| CollectionAdminRequest.createCollection(collectionName, "conf", 1, 2) |
| .process(cluster.getSolrClient()); |
| waitForState("", collectionName, clusterShape(1, 2)); |
| AtomicBoolean closed = new AtomicBoolean(false); |
| Thread[] threads = new Thread[100]; |
| for (int i = 0; i < threads.length; i++) { |
| int finalI = i; |
| threads[i] = new Thread(() -> { |
| int doc = finalI * 10000; |
| while (!closed.get()) { |
| try { |
| cluster.getSolrClient().add(collectionName, new SolrInputDocument("id", String.valueOf(doc++))); |
| } catch (Exception e) { |
| log.error("Failed on adding document to {}", collectionName, e); |
| } |
| } |
| }); |
| threads[i].start(); |
| } |
| |
| Slice shard1 = getCollectionState(collectionName).getSlice("shard1"); |
| Replica nonLeader = shard1.getReplicas(rep -> !rep.getName().equals(shard1.getLeader().getName())).get(0); |
| CollectionAdminRequest.deleteReplica(collectionName, "shard1", nonLeader.getName()).process(cluster.getSolrClient()); |
| closed.set(true); |
| for (int i = 0; i < threads.length; i++) { |
| threads[i].join(); |
| } |
| |
| try { |
| cluster.getSolrClient().waitForState(collectionName, 20, TimeUnit.SECONDS, (liveNodes, collectionState) -> collectionState.getReplicas().size() == 1); |
| } catch (TimeoutException e) { |
| if (log.isInfoEnabled()) { |
| log.info("Timeout wait for state {}", getCollectionState(collectionName)); |
| } |
| throw e; |
| } |
| } |
| |
| /** |
| * Helper method for counting the number of instances of <code>UnloadCoreOnDeletedWatcher</code> |
| * that exist on a given node. |
| * |
| * This is useful for verifying that deleting a replica correctly removed it's watchers. |
| * |
| * (Note: tests should not assert specific values, since multiple replicas may exist on the same |
| * node. Instead tests should only assert that the number of watchers has decreased by 1 per known |
| * replica removed) |
| */ |
| private static final long countUnloadCoreOnDeletedWatchers(final Set<DocCollectionWatcher> watchers) { |
| return watchers.stream().filter(w -> w instanceof ZkController.UnloadCoreOnDeletedWatcher).count(); |
| } |
| } |
| |
