| /* |
| * 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 static org.apache.solr.cloud.AbstractDistribZkTestBase.verifyReplicaStatus; |
| import static org.mockito.Mockito.anyBoolean; |
| import static org.mockito.Mockito.anyInt; |
| import static org.mockito.Mockito.anyString; |
| import static org.mockito.Mockito.doNothing; |
| import static org.mockito.Mockito.doReturn; |
| import static org.mockito.Mockito.mock; |
| import static org.mockito.Mockito.when; |
| |
| import java.io.IOException; |
| import java.lang.invoke.MethodHandles; |
| import java.nio.file.Path; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Locale; |
| import java.util.Map; |
| import java.util.Optional; |
| import java.util.Set; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.TimeoutException; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicInteger; |
| |
| import javax.xml.parsers.ParserConfigurationException; |
| |
| import org.apache.lucene.util.LuceneTestCase.Slow; |
| import org.apache.solr.SolrTestCaseJ4; |
| import org.apache.solr.client.solrj.SolrClient; |
| import org.apache.solr.client.solrj.cloud.DistributedQueue; |
| import org.apache.solr.client.solrj.cloud.SolrCloudManager; |
| import org.apache.solr.client.solrj.impl.CloudSolrClient; |
| import org.apache.solr.client.solrj.impl.SolrClientCloudManager; |
| import org.apache.solr.cloud.overseer.NodeMutator; |
| import org.apache.solr.cloud.overseer.OverseerAction; |
| import org.apache.solr.cloud.overseer.ZkWriteCommand; |
| import org.apache.solr.common.AlreadyClosedException; |
| 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.SolrZkClient; |
| import org.apache.solr.common.cloud.ZkCoreNodeProps; |
| import org.apache.solr.common.cloud.ZkNodeProps; |
| import org.apache.solr.common.cloud.ZkStateReader; |
| import org.apache.solr.common.params.CollectionParams; |
| import org.apache.solr.common.util.ExecutorUtil; |
| import org.apache.solr.common.util.IOUtils; |
| import org.apache.solr.common.util.SolrNamedThreadFactory; |
| import org.apache.solr.common.util.TimeSource; |
| import org.apache.solr.common.util.Utils; |
| import org.apache.solr.core.CloudConfig; |
| import org.apache.solr.core.CoreContainer; |
| import org.apache.solr.core.PluginInfo; |
| import org.apache.solr.core.SolrResourceLoader; |
| import org.apache.solr.handler.component.HttpShardHandler; |
| import org.apache.solr.handler.component.HttpShardHandlerFactory; |
| import org.apache.solr.metrics.SolrMetricManager; |
| import org.apache.solr.update.UpdateShardHandler; |
| import org.apache.solr.update.UpdateShardHandlerConfig; |
| import org.apache.solr.util.TimeOut; |
| import org.apache.zookeeper.CreateMode; |
| import org.apache.zookeeper.KeeperException; |
| import org.apache.zookeeper.KeeperException.NoNodeException; |
| import org.apache.zookeeper.KeeperException.NodeExistsException; |
| import org.apache.zookeeper.KeeperException.SessionExpiredException; |
| import org.apache.zookeeper.WatchedEvent; |
| import org.apache.zookeeper.Watcher.Event; |
| import org.apache.zookeeper.data.Stat; |
| import org.apache.zookeeper.proto.WatcherEvent; |
| import org.junit.After; |
| import org.junit.AfterClass; |
| import org.junit.Before; |
| import org.junit.BeforeClass; |
| import org.junit.Ignore; |
| import org.junit.Test; |
| import org.mockito.Mockito; |
| import org.mockito.internal.util.reflection.FieldSetter; |
| import org.mockito.invocation.InvocationOnMock; |
| import org.mockito.stubbing.Answer; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.xml.sax.SAXException; |
| |
| import com.codahale.metrics.Snapshot; |
| import com.codahale.metrics.Timer; |
| |
| @Slow |
| @SolrTestCaseJ4.SuppressSSL |
| public class OverseerTest extends SolrTestCaseJ4 { |
| |
| private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); |
| |
| static final int TIMEOUT = 30000; |
| |
| private static ZkTestServer server; |
| |
| private static SolrZkClient zkClient; |
| |
| |
| private volatile boolean testDone = false; |
| |
| private final List<ZkController> zkControllers = Collections.synchronizedList(new ArrayList<>()); |
| private final List<Overseer> overseers = Collections.synchronizedList(new ArrayList<>()); |
| private final List<ZkStateReader> readers = Collections.synchronizedList(new ArrayList<>()); |
| private final List<SolrZkClient> zkClients = Collections.synchronizedList(new ArrayList<>()); |
| private final List<HttpShardHandlerFactory> httpShardHandlerFactorys = Collections.synchronizedList(new ArrayList<>()); |
| private final List<UpdateShardHandler> updateShardHandlers = Collections.synchronizedList(new ArrayList<>()); |
| private final List<CloudSolrClient> solrClients = Collections.synchronizedList(new ArrayList<>()); |
| |
| private static final String COLLECTION = SolrTestCaseJ4.DEFAULT_TEST_COLLECTION_NAME; |
| |
| public static class MockZKController{ |
| |
| private final SolrZkClient zkClient; |
| private final ZkStateReader zkStateReader; |
| private final String nodeName; |
| private final Map<String, ElectionContext> electionContext = Collections.synchronizedMap(new HashMap<String, ElectionContext>()); |
| private List<Overseer> overseers; |
| |
| public MockZKController(String zkAddress, String nodeName, List<Overseer> overseers) throws InterruptedException, TimeoutException, IOException, KeeperException { |
| this.overseers = overseers; |
| this.nodeName = nodeName; |
| zkClient = new SolrZkClient(zkAddress, TIMEOUT); |
| |
| ZkController.createClusterZkNodes(zkClient); |
| |
| zkStateReader = new ZkStateReader(zkClient); |
| zkStateReader.createClusterStateWatchersAndUpdate(); |
| |
| // live node |
| final String nodePath = ZkStateReader.LIVE_NODES_ZKNODE + "/" + nodeName; |
| zkClient.makePath(nodePath, CreateMode.EPHEMERAL, true); |
| } |
| |
| private void deleteNode(final String path) { |
| |
| try { |
| zkClient.delete(path, -1, true); |
| } catch (NoNodeException e) { |
| // fine |
| log.warn("cancelElection did not find election node to remove"); |
| } catch (KeeperException e) { |
| fail("Unexpected KeeperException!" + e); |
| } catch (InterruptedException e) { |
| fail("Unexpected InterruptedException!" + e); |
| } |
| } |
| |
| public void close() { |
| for (ElectionContext ec : electionContext.values()) { |
| try { |
| ec.cancelElection(); |
| } catch (Exception e) { |
| log.warn(String.format(Locale.ROOT, "Error cancelling election for %s", ec.id), e); |
| } |
| } |
| deleteNode(ZkStateReader.LIVE_NODES_ZKNODE + "/" + nodeName); |
| zkClient.close(); |
| zkStateReader.close(); |
| } |
| |
| public void createCollection(String collection, int numShards) throws Exception { |
| |
| ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, CollectionParams.CollectionAction.CREATE.toLower(), |
| "name", collection, |
| ZkStateReader.REPLICATION_FACTOR, "1", |
| ZkStateReader.NUM_SHARDS_PROP, numShards+"", |
| "createNodeSet", ""); |
| ZkDistributedQueue q = MiniSolrCloudCluster.getOpenOverseer(overseers).getStateUpdateQueue(); |
| q.offer(Utils.toJSON(m)); |
| |
| } |
| |
| public String publishState(String collection, String coreName, String coreNodeName, String shard, Replica.State stateName, int numShards, boolean startElection, Overseer overseer) |
| throws Exception { |
| if (stateName == null) { |
| ElectionContext ec = electionContext.remove(coreName); |
| if (ec != null) { |
| ec.cancelElection(); |
| } |
| ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.DELETECORE.toLower(), |
| ZkStateReader.NODE_NAME_PROP, nodeName, |
| ZkStateReader.CORE_NAME_PROP, coreName, |
| ZkStateReader.CORE_NODE_NAME_PROP, coreNodeName, |
| ZkStateReader.COLLECTION_PROP, collection); |
| ZkDistributedQueue q = overseer.getStateUpdateQueue(); |
| q.offer(Utils.toJSON(m)); |
| return null; |
| } else { |
| ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.STATE.toLower(), |
| ZkStateReader.STATE_PROP, stateName.toString(), |
| ZkStateReader.NODE_NAME_PROP, nodeName, |
| ZkStateReader.CORE_NAME_PROP, coreName, |
| ZkStateReader.CORE_NODE_NAME_PROP, coreNodeName, |
| ZkStateReader.COLLECTION_PROP, collection, |
| ZkStateReader.SHARD_ID_PROP, shard, |
| ZkStateReader.NUM_SHARDS_PROP, Integer.toString(numShards)); |
| ZkDistributedQueue q = overseer.getStateUpdateQueue(); |
| q.offer(Utils.toJSON(m)); |
| } |
| |
| if (startElection && collection.length() > 0) { |
| zkStateReader.waitForState(collection, 45000, TimeUnit.MILLISECONDS, |
| (liveNodes, collectionState) -> getShardId(collectionState, coreNodeName) != null); |
| String shardId = getShardId(collection, coreNodeName); |
| if (shardId != null) { |
| ElectionContext prevContext = electionContext.get(coreName); |
| if (prevContext != null) { |
| prevContext.cancelElection(); |
| } |
| |
| try { |
| zkClient.makePath("/collections/" + collection + "/leader_elect/" |
| + shardId + "/election", true); |
| } catch (NodeExistsException nee) {} |
| ZkNodeProps props = new ZkNodeProps(ZkStateReader.NODE_NAME_PROP, nodeName, |
| ZkStateReader.CORE_NAME_PROP, coreName, |
| ZkStateReader.SHARD_ID_PROP, shardId, |
| ZkStateReader.COLLECTION_PROP, collection, |
| ZkStateReader.CORE_NODE_NAME_PROP, coreNodeName); |
| LeaderElector elector = new LeaderElector(zkClient); |
| ShardLeaderElectionContextBase ctx = new ShardLeaderElectionContextBase( |
| elector, shardId, collection, nodeName + coreName, props, |
| MockSolrSource.makeSimpleMock(overseer, zkStateReader, null)); |
| elector.setup(ctx); |
| electionContext.put(coreName, ctx); |
| elector.joinElection(ctx, false); |
| return shardId; |
| } |
| } |
| return null; |
| } |
| |
| private String getShardId(String collection, String coreNodeName) { |
| DocCollection dc = zkStateReader.getClusterState().getCollectionOrNull(collection); |
| return getShardId(dc, coreNodeName); |
| } |
| |
| private String getShardId(DocCollection collection, String coreNodeName) { |
| if (collection == null) return null; |
| Map<String,Slice> slices = collection.getSlicesMap(); |
| if (slices != null) { |
| for (Slice slice : slices.values()) { |
| for (Replica replica : slice.getReplicas()) { |
| String cnn = replica.getName(); |
| if (coreNodeName.equals(cnn)) { |
| return slice.getName(); |
| } |
| } |
| } |
| } |
| return null; |
| } |
| |
| |
| public ZkStateReader getZkReader() { |
| return zkStateReader; |
| } |
| } |
| |
| @BeforeClass |
| public static void beforeClass() throws Exception { |
| assumeWorkingMockito(); |
| |
| System.setProperty("solr.zkclienttimeout", "30000"); |
| |
| Path zkDir = createTempDir("zkData"); |
| |
| server = new ZkTestServer(zkDir); |
| server.run(); |
| |
| zkClient = server.getZkClient(); |
| |
| initCore(); |
| } |
| |
| |
| @Before |
| public void setUp() throws Exception { |
| testDone = false; |
| super.setUp(); |
| } |
| |
| @AfterClass |
| public static void afterClass() throws Exception { |
| if (null != zkClient) { |
| zkClient.printLayoutToStdOut(); |
| } |
| |
| System.clearProperty("solr.zkclienttimeout"); |
| |
| if (null != server) { |
| server.shutdown(); |
| } |
| |
| server = null; |
| } |
| |
| @After |
| public void tearDown() throws Exception { |
| testDone = true; |
| |
| ExecutorService customThreadPool = ExecutorUtil.newMDCAwareCachedThreadPool(new SolrNamedThreadFactory("closeThreadPool")); |
| |
| for (ZkController zkController : zkControllers) { |
| customThreadPool.submit(zkController::close); |
| } |
| |
| for (HttpShardHandlerFactory httpShardHandlerFactory : httpShardHandlerFactorys) { |
| customThreadPool.submit(httpShardHandlerFactory::close); |
| } |
| |
| for (UpdateShardHandler updateShardHandler : updateShardHandlers) { |
| customThreadPool.submit(updateShardHandler::close); |
| } |
| |
| for (SolrClient solrClient : solrClients) { |
| customThreadPool.submit( () -> IOUtils.closeQuietly(solrClient)); |
| } |
| |
| for (ZkStateReader reader : readers) { |
| customThreadPool.submit(reader::close); |
| } |
| |
| for (SolrZkClient solrZkClient : zkClients) { |
| customThreadPool.submit( () -> IOUtils.closeQuietly(solrZkClient)); |
| } |
| |
| ExecutorUtil.shutdownAndAwaitTermination(customThreadPool); |
| |
| customThreadPool = ExecutorUtil.newMDCAwareCachedThreadPool(new SolrNamedThreadFactory("closeThreadPool")); |
| |
| |
| for (Overseer overseer : overseers) { |
| customThreadPool.submit(overseer::close); |
| } |
| |
| ExecutorUtil.shutdownAndAwaitTermination(customThreadPool); |
| |
| overseers.clear(); |
| zkControllers.clear(); |
| httpShardHandlerFactorys.clear(); |
| updateShardHandlers.clear(); |
| solrClients.clear(); |
| readers.clear(); |
| zkClients.clear(); |
| |
| server.tryCleanSolrZkNode(); |
| server.makeSolrZkNode(); |
| |
| super.tearDown(); |
| } |
| |
| @Test |
| public void testShardAssignment() throws Exception { |
| |
| MockZKController mockController = null; |
| SolrZkClient overseerClient = null; |
| |
| try { |
| |
| |
| ZkController.createClusterZkNodes(zkClient); |
| |
| overseerClient = electNewOverseer(server.getZkAddress()); |
| |
| try (ZkStateReader reader = new ZkStateReader(zkClient)) { |
| reader.createClusterStateWatchersAndUpdate(); |
| |
| mockController = new MockZKController(server.getZkAddress(), "127.0.0.1:8983_solr", overseers); |
| |
| final int numShards = 6; |
| |
| ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, CollectionParams.CollectionAction.CREATE.toLower(), |
| "name", COLLECTION, |
| ZkStateReader.REPLICATION_FACTOR, "1", |
| ZkStateReader.NUM_SHARDS_PROP, "3", |
| "createNodeSet", ""); |
| ZkDistributedQueue q = overseers.get(0).getStateUpdateQueue(); |
| q.offer(Utils.toJSON(m)); |
| |
| for (int i = 0; i < numShards; i++) { |
| assertNotNull("shard got no id?", mockController.publishState(COLLECTION, "core" + (i + 1), "node" + (i + 1), "shard" + ((i % 3) + 1), Replica.State.ACTIVE, 3, true, overseers.get(0))); |
| } |
| |
| reader.waitForState(COLLECTION, 30, TimeUnit.SECONDS, MiniSolrCloudCluster.expectedShardsAndActiveReplicas(3, 6)); |
| |
| final Map<String, Replica> rmap = reader.getClusterState().getCollection(COLLECTION).getSlice("shard1").getReplicasMap(); |
| assertEquals(rmap.toString(), 2, rmap.size()); |
| assertEquals(rmap.toString(), 2, reader.getClusterState().getCollection(COLLECTION).getSlice("shard2").getReplicasMap().size()); |
| assertEquals(rmap.toString(), 2, reader.getClusterState().getCollection(COLLECTION).getSlice("shard3").getReplicasMap().size()); |
| |
| //make sure leaders are in cloud state |
| assertNotNull(reader.getLeaderUrl(COLLECTION, "shard1", 15000)); |
| assertNotNull(reader.getLeaderUrl(COLLECTION, "shard2", 15000)); |
| assertNotNull(reader.getLeaderUrl(COLLECTION, "shard3", 15000)); |
| } |
| } finally { |
| if (mockController != null) { |
| mockController.close(); |
| } |
| close(overseerClient); |
| } |
| } |
| |
| @Test |
| public void testBadQueueItem() throws Exception { |
| |
| MockZKController mockController = null; |
| SolrZkClient overseerClient = null; |
| |
| try { |
| ZkController.createClusterZkNodes(zkClient); |
| |
| overseerClient = electNewOverseer(server.getZkAddress()); |
| |
| try (ZkStateReader reader = new ZkStateReader(zkClient)) { |
| reader.createClusterStateWatchersAndUpdate(); |
| |
| mockController = new MockZKController(server.getZkAddress(), "127.0.0.1:8983_solr", overseers); |
| |
| final int numShards = 3; |
| mockController.createCollection(COLLECTION, 3); |
| for (int i = 0; i < numShards; i++) { |
| assertNotNull("shard got no id?", mockController.publishState(COLLECTION, "core" + (i + 1), |
| "node" + (i + 1), "shard" + ((i % 3) + 1), Replica.State.ACTIVE, 3, true, overseers.get(0))); |
| } |
| |
| reader.waitForState(COLLECTION, 30, TimeUnit.SECONDS, MiniSolrCloudCluster.expectedShardsAndActiveReplicas(3, 3)); |
| |
| assertEquals(1, reader.getClusterState().getCollection(COLLECTION).getSlice("shard1").getReplicasMap().size()); |
| assertEquals(1, reader.getClusterState().getCollection(COLLECTION).getSlice("shard2").getReplicasMap().size()); |
| assertEquals(1, reader.getClusterState().getCollection(COLLECTION).getSlice("shard3").getReplicasMap().size()); |
| |
| //make sure leaders are in cloud state |
| assertNotNull(reader.getLeaderUrl(COLLECTION, "shard1", 15000)); |
| assertNotNull(reader.getLeaderUrl(COLLECTION, "shard2", 15000)); |
| assertNotNull(reader.getLeaderUrl(COLLECTION, "shard3", 15000)); |
| |
| // publish a bad queue item |
| String emptyCollectionName = ""; |
| mockController.publishState(emptyCollectionName, "core0", "node0", "shard1", Replica.State.ACTIVE, 1, true, overseers.get(0)); |
| mockController.publishState(emptyCollectionName, "core0", "node0", "shard1", null, 1, true, overseers.get(0)); |
| |
| mockController.createCollection("collection2", 3); |
| // make sure the Overseer is still processing items |
| for (int i = 0; i < numShards; i++) { |
| assertNotNull("shard got no id?", mockController.publishState("collection2", |
| "core" + (i + 1), "node" + (i + 1), "shard" + ((i % 3) + 1), Replica.State.ACTIVE, 3, true, overseers.get(0))); |
| } |
| |
| reader.waitForState("collection2", 30, TimeUnit.SECONDS, MiniSolrCloudCluster.expectedShardsAndActiveReplicas(3, 3)); |
| |
| assertEquals(1, reader.getClusterState().getCollection("collection2").getSlice("shard1").getReplicasMap().size()); |
| assertEquals(1, reader.getClusterState().getCollection("collection2").getSlice("shard2").getReplicasMap().size()); |
| assertEquals(1, reader.getClusterState().getCollection("collection2").getSlice("shard3").getReplicasMap().size()); |
| |
| //make sure leaders are in cloud state |
| assertNotNull(reader.getLeaderUrl("collection2", "shard1", 15000)); |
| assertNotNull(reader.getLeaderUrl("collection2", "shard2", 15000)); |
| assertNotNull(reader.getLeaderUrl("collection2", "shard3", 15000)); |
| } |
| |
| } finally { |
| if (mockController != null) { |
| mockController.close(); |
| } |
| close(overseerClient); |
| } |
| } |
| |
| @Test |
| @SuppressWarnings({"try"}) |
| public void testDownNodeFailover() throws Exception { |
| MockZKController mockController = null; |
| SolrZkClient overseerClient = null; |
| |
| try { |
| |
| ZkController.createClusterZkNodes(zkClient); |
| |
| overseerClient = electNewOverseer(server.getZkAddress()); |
| |
| try (ZkStateReader reader = new ZkStateReader(zkClient)) { |
| reader.createClusterStateWatchersAndUpdate(); |
| |
| mockController = new MockZKController(server.getZkAddress(), "127.0.0.1:8983_solr", overseers); |
| |
| try (ZkController zkController = createMockZkController(server.getZkAddress(), zkClient, reader)) { |
| |
| for (int i = 0; i < 5; i++) { |
| mockController.createCollection("collection" + i, 1); |
| assertNotNull("shard got no id?", mockController.publishState("collection" + i, "core1", |
| "core_node1", "shard1", Replica.State.ACTIVE, 1, true, overseers.get(0))); |
| } |
| } |
| ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.DOWNNODE.toLower(), |
| ZkStateReader.NODE_NAME_PROP, "127.0.0.1:8983_solr"); |
| List<ZkWriteCommand> commands = new NodeMutator(null).downNode(reader.getClusterState(), m); |
| |
| ZkDistributedQueue q = overseers.get(0).getStateUpdateQueue(); |
| |
| q.offer(Utils.toJSON(m)); |
| |
| verifyReplicaStatus(reader, commands.get(0).name, "shard1", "core_node1", Replica.State.DOWN); |
| overseerClient.close(); |
| |
| overseerClient = electNewOverseer(server.getZkAddress()); |
| for (int i = 0; i < 5; i++) { |
| verifyReplicaStatus(reader, "collection" + i, "shard1", "core_node1", Replica.State.DOWN); |
| } |
| } |
| } finally { |
| if (mockController != null) { |
| mockController.close(); |
| } |
| close(overseerClient); |
| } |
| } |
| |
| //wait until collections are available |
| private void waitForCollections(ZkStateReader stateReader, String... collections) throws InterruptedException, KeeperException, TimeoutException { |
| int maxIterations = 100; |
| while (0 < maxIterations--) { |
| |
| final ClusterState state = stateReader.getClusterState(); |
| Set<String> availableCollections = state.getCollectionsMap().keySet(); |
| int availableCount = 0; |
| for(String requiredCollection: collections) { |
| stateReader.waitForState(requiredCollection, 30000, TimeUnit.MILLISECONDS, (liveNodes, collectionState) -> collectionState != null); |
| if(availableCollections.contains(requiredCollection)) { |
| availableCount++; |
| } |
| if(availableCount == collections.length) return; |
| |
| } |
| } |
| log.warn("Timeout waiting for collections: {} state: {}" |
| , Arrays.asList(collections), stateReader.getClusterState()); |
| } |
| |
| @Test |
| public void testStateChange() throws Exception { |
| |
| ZkStateReader reader = null; |
| SolrZkClient overseerClient = null; |
| |
| try { |
| |
| ZkController.createClusterZkNodes(zkClient); |
| |
| reader = new ZkStateReader(zkClient); |
| reader.createClusterStateWatchersAndUpdate(); |
| |
| overseerClient = electNewOverseer(server.getZkAddress()); |
| |
| ZkDistributedQueue q = overseers.get(0).getStateUpdateQueue(); |
| |
| ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, CollectionParams.CollectionAction.CREATE.toLower(), |
| "name", COLLECTION, |
| ZkStateReader.REPLICATION_FACTOR, "1", |
| ZkStateReader.NUM_SHARDS_PROP, "1", |
| "createNodeSet", ""); |
| q.offer(Utils.toJSON(m)); |
| |
| m = new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.STATE.toLower(), |
| ZkStateReader.NODE_NAME_PROP, "node1:8983_", |
| ZkStateReader.COLLECTION_PROP, COLLECTION, |
| ZkStateReader.SHARD_ID_PROP, "shard1", |
| ZkStateReader.CORE_NAME_PROP, "core1", |
| ZkStateReader.CORE_NODE_NAME_PROP, "core_node1", |
| ZkStateReader.ROLES_PROP, "", |
| ZkStateReader.STATE_PROP, Replica.State.RECOVERING.toString()); |
| |
| q.offer(Utils.toJSON(m)); |
| |
| waitForCollections(reader, COLLECTION); |
| verifyReplicaStatus(reader, "collection1", "shard1", "core_node1", Replica.State.RECOVERING); |
| |
| //publish node state (active) |
| m = new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.STATE.toLower(), |
| ZkStateReader.NODE_NAME_PROP, "node1:8983_", |
| ZkStateReader.COLLECTION_PROP, COLLECTION, |
| ZkStateReader.SHARD_ID_PROP, "shard1", |
| ZkStateReader.CORE_NAME_PROP, "core1", |
| ZkStateReader.ROLES_PROP, "", |
| ZkStateReader.STATE_PROP, Replica.State.ACTIVE.toString()); |
| |
| q.offer(Utils.toJSON(m)); |
| |
| verifyReplicaStatus(reader, "collection1", "shard1", "core_node1", Replica.State.ACTIVE); |
| |
| } finally { |
| |
| close(overseerClient); |
| |
| close(reader); |
| } |
| } |
| |
| private void verifyShardLeader(ZkStateReader reader, String collection, String shard, String expectedCore) |
| throws InterruptedException, KeeperException, TimeoutException { |
| |
| reader.waitForState(collection, 15000, TimeUnit.MILLISECONDS, |
| (liveNodes, collectionState) -> collectionState != null |
| && expectedCore.equals((collectionState.getLeader(shard) != null) |
| ? collectionState.getLeader(shard).getStr(ZkStateReader.CORE_NAME_PROP) : null)); |
| |
| DocCollection docCollection = reader.getClusterState().getCollection(collection); |
| assertEquals("Unexpected shard leader coll:" + collection + " shard:" + shard, expectedCore, |
| (docCollection.getLeader(shard) != null) ? docCollection.getLeader(shard).getStr(ZkStateReader.CORE_NAME_PROP) |
| : null); |
| } |
| |
| private Overseer getOpenOverseer() { |
| return MiniSolrCloudCluster.getOpenOverseer(overseers); |
| } |
| |
| @Test |
| public void testOverseerFailure() throws Exception { |
| |
| SolrZkClient overseerClient = null; |
| ZkStateReader reader = null; |
| MockZKController mockController = null; |
| |
| try { |
| |
| final String core = "core1"; |
| final String core_node = "core_node1"; |
| final String shard = "shard1"; |
| final int numShards = 1; |
| |
| ZkController.createClusterZkNodes(zkClient); |
| |
| reader = new ZkStateReader(zkClient); |
| reader.createClusterStateWatchersAndUpdate(); |
| |
| mockController = new MockZKController(server.getZkAddress(), "127.0.0.1:8983_solr", overseers); |
| |
| overseerClient = electNewOverseer(server.getZkAddress()); |
| |
| mockController.createCollection(COLLECTION, 1); |
| |
| ZkController zkController = createMockZkController(server.getZkAddress(), zkClient, reader); |
| |
| mockController.publishState(COLLECTION, core, core_node, "shard1", |
| Replica.State.RECOVERING, numShards, true, overseers.get(0)); |
| |
| waitForCollections(reader, COLLECTION); |
| verifyReplicaStatus(reader, COLLECTION, "shard1", "core_node1", Replica.State.RECOVERING); |
| |
| mockController.publishState(COLLECTION, core, core_node, "shard1", Replica.State.ACTIVE, |
| numShards, true, overseers.get(0)); |
| |
| verifyReplicaStatus(reader, COLLECTION, "shard1", "core_node1", Replica.State.ACTIVE); |
| |
| mockController.publishState(COLLECTION, core, core_node, "shard1", |
| Replica.State.RECOVERING, numShards, true, overseers.get(0)); |
| |
| overseerClient.close(); |
| |
| overseerClient = electNewOverseer(server.getZkAddress()); |
| |
| verifyReplicaStatus(reader, COLLECTION, "shard1", "core_node1", Replica.State.RECOVERING); |
| |
| assertEquals("Live nodes count does not match", 1, reader |
| .getClusterState().getLiveNodes().size()); |
| assertEquals(shard+" replica count does not match", 1, reader.getClusterState() |
| .getCollection(COLLECTION).getSlice(shard).getReplicasMap().size()); |
| mockController.publishState(COLLECTION, core, core_node, "shard1", null, numShards, true, overseers.get(1)); |
| |
| reader.waitForState(COLLECTION, 5000, |
| TimeUnit.MILLISECONDS, (liveNodes, collectionState) -> collectionState != null && collectionState.getReplica(core_node) == null); |
| |
| reader.forceUpdateCollection(COLLECTION); |
| // as of SOLR-5209 core removal does not cascade to remove the slice and collection |
| assertTrue(COLLECTION +" should remain after removal of the last core", |
| reader.getClusterState().hasCollection(COLLECTION)); |
| assertTrue(core_node+" should be gone after publishing the null state", |
| null == reader.getClusterState().getCollection(COLLECTION).getReplica(core_node)); |
| } finally { |
| close(mockController); |
| close(overseerClient); |
| close(reader); |
| } |
| } |
| |
| @Test |
| public void testOverseerStatsReset() throws Exception { |
| ZkStateReader reader = null; |
| MockZKController mockController = null; |
| |
| try { |
| |
| ZkController.createClusterZkNodes(zkClient); |
| |
| reader = new ZkStateReader(zkClient); |
| reader.createClusterStateWatchersAndUpdate(); |
| |
| mockController = new MockZKController(server.getZkAddress(), "127.0.0.1:8983_solr", overseers); |
| |
| LeaderElector overseerElector = new LeaderElector(zkClient); |
| if (overseers.size() > 0) { |
| overseers.get(overseers.size() -1).close(); |
| overseers.get(overseers.size() -1).getZkStateReader().getZkClient().close(); |
| } |
| ZkController zkController = createMockZkController(server.getZkAddress(), zkClient, reader); |
| |
| UpdateShardHandler updateShardHandler = new UpdateShardHandler(UpdateShardHandlerConfig.DEFAULT); |
| updateShardHandlers.add(updateShardHandler); |
| HttpShardHandlerFactory httpShardHandlerFactory = new HttpShardHandlerFactory(); |
| httpShardHandlerFactory.init(new PluginInfo("shardHandlerFactory", Collections.emptyMap())); |
| httpShardHandlerFactorys.add(httpShardHandlerFactory); |
| Overseer overseer = new Overseer((HttpShardHandler) httpShardHandlerFactory.getShardHandler(), updateShardHandler, "/admin/cores", reader, zkController, |
| new CloudConfig.CloudConfigBuilder("127.0.0.1", 8983, "").build()); |
| overseers.add(overseer); |
| ElectionContext ec = new OverseerElectionContext(zkClient, overseer, |
| server.getZkAddress().replaceAll("/", "_")); |
| overseerElector.setup(ec); |
| overseerElector.joinElection(ec, false); |
| |
| mockController.createCollection(COLLECTION, 1); |
| |
| mockController.publishState(COLLECTION, "core1", "core_node1", "shard1", Replica.State.ACTIVE, 1, true, overseers.get(0)); |
| |
| assertNotNull(overseer.getStats()); |
| assertTrue((overseer.getStats().getSuccessCount(OverseerAction.STATE.toLower())) > 0); |
| |
| // shut it down |
| overseer.close(); |
| ec.cancelElection(); |
| |
| // start it again |
| overseerElector.setup(ec); |
| overseerElector.joinElection(ec, false); |
| assertNotNull(overseer.getStats()); |
| assertEquals(0, (overseer.getStats().getSuccessCount(OverseerAction.STATE.toLower()))); |
| |
| } finally { |
| close(mockController); |
| close(reader); |
| } |
| } |
| |
| private AtomicInteger killCounter = new AtomicInteger(); |
| |
| private class OverseerRestarter implements Runnable{ |
| SolrZkClient overseerClient = null; |
| public volatile boolean run = true; |
| private final String zkAddress; |
| |
| public OverseerRestarter(String zkAddress) { |
| this.zkAddress = zkAddress; |
| } |
| |
| @Override |
| public void run() { |
| try { |
| overseerClient = electNewOverseer(zkAddress); |
| while (run) { |
| if (killCounter.get()>0) { |
| try { |
| killCounter.decrementAndGet(); |
| log.info("Killing overseer."); |
| overseerClient.close(); |
| overseerClient = electNewOverseer(zkAddress); |
| } catch (Throwable e) { |
| // e.printStackTrace(); |
| } |
| } |
| try { |
| Thread.sleep(100); |
| } catch (Throwable e) { |
| // e.printStackTrace(); |
| } |
| } |
| } catch (Throwable t) { |
| // ignore |
| } finally { |
| if (overseerClient != null) { |
| try { |
| // overseerClient.close(); |
| } catch (Throwable t) { |
| // ignore |
| } |
| } |
| } |
| } |
| } |
| |
| @Test |
| public void testExceptionWhenFlushClusterState() throws Exception { |
| |
| SolrZkClient overseerClient = null; |
| ZkStateReader reader = null; |
| |
| try { |
| |
| ZkController.createClusterZkNodes(zkClient); |
| |
| reader = new ZkStateReader(zkClient); |
| reader.createClusterStateWatchersAndUpdate(); |
| |
| // We did not create /collections -> this message will cause exception when Overseer try to flush the clusterstate |
| ZkNodeProps badMessage = new ZkNodeProps(Overseer.QUEUE_OPERATION, CollectionParams.CollectionAction.CREATE.toLower(), |
| "name", "collection1", |
| ZkStateReader.REPLICATION_FACTOR, "1", |
| ZkStateReader.NUM_SHARDS_PROP, "1", |
| DocCollection.STATE_FORMAT, "2", |
| "createNodeSet", ""); |
| ZkNodeProps goodMessage = new ZkNodeProps(Overseer.QUEUE_OPERATION, CollectionParams.CollectionAction.CREATE.toLower(), |
| "name", "collection2", |
| ZkStateReader.REPLICATION_FACTOR, "1", |
| ZkStateReader.NUM_SHARDS_PROP, "1", |
| DocCollection.STATE_FORMAT, "1", |
| "createNodeSet", ""); |
| ZkDistributedQueue workQueue = Overseer.getInternalWorkQueue(zkClient, new Stats()); |
| workQueue.offer(Utils.toJSON(badMessage)); |
| workQueue.offer(Utils.toJSON(goodMessage)); |
| overseerClient = electNewOverseer(server.getZkAddress()); |
| waitForCollections(reader, "collection2"); |
| |
| ZkDistributedQueue q = getOpenOverseer().getStateUpdateQueue(); |
| q.offer(Utils.toJSON(badMessage)); |
| q.offer(Utils.toJSON(goodMessage.plus("name", "collection3"))); |
| waitForCollections(reader, "collection2", "collection3"); |
| assertNotNull(reader.getClusterState().getCollectionOrNull("collection2")); |
| assertNotNull(reader.getClusterState().getCollectionOrNull("collection3")); |
| |
| TimeOut timeOut = new TimeOut(10, TimeUnit.SECONDS, TimeSource.NANO_TIME); |
| while(!timeOut.hasTimedOut()) { |
| if (q.peek() == null) { |
| break; |
| } |
| Thread.sleep(50); |
| } |
| |
| assertTrue(showQpeek(workQueue), workQueue.peek() == null); |
| assertTrue(showQpeek(q), q.peek() == null); |
| } finally { |
| close(overseerClient); |
| close(reader); |
| } |
| } |
| |
| private String showQpeek(ZkDistributedQueue q) throws KeeperException, InterruptedException { |
| if (q == null) { |
| return ""; |
| } |
| byte[] bytes = q.peek(); |
| if (bytes == null) { |
| return ""; |
| } |
| |
| ZkNodeProps json = ZkNodeProps.load(bytes); |
| return json.toString(); |
| } |
| |
| |
| @Test |
| public void testShardLeaderChange() throws Exception { |
| ZkStateReader reader = null; |
| MockZKController mockController = null; |
| MockZKController mockController2 = null; |
| OverseerRestarter killer = null; |
| Thread killerThread = null; |
| |
| try { |
| ZkController.createClusterZkNodes(zkClient); |
| |
| killer = new OverseerRestarter(server.getZkAddress()); |
| killerThread = new Thread(killer); |
| killerThread.start(); |
| |
| reader = new ZkStateReader(zkClient); |
| reader.createClusterStateWatchersAndUpdate(); |
| |
| UpdateShardHandler updateShardHandler = new UpdateShardHandler(UpdateShardHandlerConfig.DEFAULT); |
| updateShardHandlers.add(updateShardHandler); |
| HttpShardHandlerFactory httpShardHandlerFactory = new HttpShardHandlerFactory(); |
| httpShardHandlerFactorys.add(httpShardHandlerFactory); |
| |
| electNewOverseer(server.getZkAddress()); |
| |
| for (int i = 0; i < atLeast(4); i++) { |
| killCounter.incrementAndGet(); // for each round allow 1 kill |
| |
| mockController = new MockZKController(server.getZkAddress(), "node1:8983_", overseers); |
| |
| TimeOut timeout = new TimeOut(10, TimeUnit.SECONDS, TimeSource.NANO_TIME); |
| while (!timeout.hasTimedOut()) { |
| try { |
| mockController.createCollection(COLLECTION, 1); |
| break; |
| } catch (SolrException | KeeperException | AlreadyClosedException e) { |
| e.printStackTrace(); |
| } |
| } |
| |
| timeout = new TimeOut(10, TimeUnit.SECONDS, TimeSource.NANO_TIME); |
| while (!timeout.hasTimedOut()) { |
| try { |
| mockController.publishState(COLLECTION, "core1", "node1", "shard1", Replica.State.ACTIVE, |
| 1, true, getOpenOverseer()); |
| break; |
| } catch (SolrException | KeeperException | AlreadyClosedException e) { |
| e.printStackTrace(); |
| } |
| } |
| |
| if (mockController2 != null) { |
| mockController2.close(); |
| mockController2 = null; |
| } |
| |
| Thread.sleep(100); |
| |
| timeout = new TimeOut(1, TimeUnit.SECONDS, TimeSource.NANO_TIME); |
| while (!timeout.hasTimedOut()) { |
| try { |
| mockController.publishState(COLLECTION, "core1", "node1", "shard1", |
| Replica.State.RECOVERING, 1, true, getOpenOverseer()); |
| break; |
| } catch (SolrException | AlreadyClosedException e) { |
| e.printStackTrace(); |
| } |
| } |
| |
| mockController2 = new MockZKController(server.getZkAddress(), "node2:8984_", overseers); |
| |
| timeout = new TimeOut(10, TimeUnit.SECONDS, TimeSource.NANO_TIME); |
| while (!timeout.hasTimedOut()) { |
| try { |
| mockController.publishState(COLLECTION, "core1", "node1", "shard1", Replica.State.ACTIVE, |
| 1, true, getOpenOverseer()); |
| break; |
| } catch (SolrException | AlreadyClosedException e) { |
| e.printStackTrace(); |
| } |
| } |
| |
| verifyShardLeader(reader, COLLECTION, "shard1", "core1"); |
| |
| |
| timeout = new TimeOut(10, TimeUnit.SECONDS, TimeSource.NANO_TIME); |
| while (!timeout.hasTimedOut()) { |
| try { |
| mockController2.publishState(COLLECTION, "core4", "node2", "shard1", Replica.State.ACTIVE, |
| 1, true, getOpenOverseer()); |
| break; |
| } catch (SolrException | AlreadyClosedException e) { |
| e.printStackTrace(); |
| } |
| } |
| |
| |
| mockController.close(); |
| mockController = null; |
| |
| ZkController zkController = createMockZkController(server.getZkAddress(), null, reader); |
| zkControllers.add(zkController); |
| |
| TimeOut timeOut = new TimeOut(10, TimeUnit.SECONDS, TimeSource.NANO_TIME); |
| timeOut.waitFor("Timed out waiting to see core4 as leader", () -> { |
| |
| ZkCoreNodeProps leaderProps; |
| try { |
| leaderProps = zkController.getLeaderProps(COLLECTION, "shard1", 1000, false); |
| } catch (SolrException e) { |
| return false; |
| } catch (InterruptedException e) { |
| throw new RuntimeException(e); |
| } catch (SessionExpiredException e) { |
| return false; |
| } |
| if (leaderProps.getCoreName().equals("core4")) { |
| return true; |
| } |
| return false; |
| |
| }); |
| |
| } |
| |
| } finally { |
| if (killer != null) { |
| killer.run = false; |
| if (killerThread != null) { |
| killerThread.join(); |
| } |
| } |
| close(mockController); |
| |
| close(mockController2); |
| close(reader); |
| } |
| } |
| |
| @Test |
| public void testDoubleAssignment() throws Exception { |
| |
| SolrZkClient overseerClient = null; |
| ZkStateReader reader = null; |
| MockZKController mockController = null; |
| |
| try { |
| |
| ZkController.createClusterZkNodes(zkClient); |
| |
| reader = new ZkStateReader(zkClient); |
| reader.createClusterStateWatchersAndUpdate(); |
| |
| mockController = new MockZKController(server.getZkAddress(), "127.0.0.1:8983_solr", overseers); |
| |
| overseerClient = electNewOverseer(server.getZkAddress()); |
| |
| mockController.createCollection(COLLECTION, 1); |
| |
| ZkController zkController = createMockZkController(server.getZkAddress(), zkClient, reader); |
| |
| mockController.publishState(COLLECTION, "core1", "core_node1", "shard1", Replica.State.RECOVERING, 1, true, overseers.get(0)); |
| |
| waitForCollections(reader, COLLECTION); |
| |
| verifyReplicaStatus(reader, COLLECTION, "shard1", "core_node1", Replica.State.RECOVERING); |
| |
| mockController.close(); |
| |
| mockController = new MockZKController(server.getZkAddress(), "127.0.0.1:8983_solr", overseers); |
| |
| mockController.publishState(COLLECTION, "core1", "core_node1","shard1", Replica.State.RECOVERING, 1, true, overseers.get(0)); |
| |
| reader.forceUpdateCollection(COLLECTION); |
| ClusterState state = reader.getClusterState(); |
| |
| int numFound = 0; |
| Map<String, DocCollection> collectionsMap = state.getCollectionsMap(); |
| for (Map.Entry<String, DocCollection> entry : collectionsMap.entrySet()) { |
| DocCollection collection = entry.getValue(); |
| for (Slice slice : collection.getSlices()) { |
| if (slice.getReplicasMap().get("core_node1") != null) { |
| numFound++; |
| } |
| } |
| } |
| assertEquals("Shard was found more than once in ClusterState", 1, |
| numFound); |
| } finally { |
| close(overseerClient); |
| close(mockController); |
| close(reader); |
| } |
| } |
| |
| @Test |
| @Ignore |
| public void testPerformance() throws Exception { |
| |
| SolrZkClient overseerClient = null; |
| ZkStateReader reader = null; |
| MockZKController mockController = null; |
| |
| try { |
| |
| ZkController.createClusterZkNodes(zkClient); |
| |
| reader = new ZkStateReader(zkClient); |
| reader.createClusterStateWatchersAndUpdate(); |
| |
| mockController = new MockZKController(server.getZkAddress(), "127.0.0.1:8983_solr", overseers); |
| |
| final int MAX_COLLECTIONS = 10, MAX_CORES = 10, MAX_STATE_CHANGES = 20000, STATE_FORMAT = 2; |
| |
| for (int i=0; i<MAX_COLLECTIONS; i++) { |
| ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, CollectionParams.CollectionAction.CREATE.toLower(), |
| "name", "perf" + i, |
| ZkStateReader.NUM_SHARDS_PROP, "1", |
| "stateFormat", String.valueOf(STATE_FORMAT), |
| ZkStateReader.REPLICATION_FACTOR, "1", |
| ZkStateReader.MAX_SHARDS_PER_NODE, "1" |
| ); |
| ZkDistributedQueue q = overseers.get(0).getStateUpdateQueue(); |
| q.offer(Utils.toJSON(m)); |
| zkClient.makePath("/collections/perf" + i, true); |
| } |
| |
| for (int i = 0, j = 0, k = 0; i < MAX_STATE_CHANGES; i++, j++, k++) { |
| ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.STATE.toLower(), |
| ZkStateReader.STATE_PROP, Replica.State.RECOVERING.toString(), |
| ZkStateReader.NODE_NAME_PROP, "node1", |
| ZkStateReader.CORE_NAME_PROP, "core" + k, |
| ZkStateReader.CORE_NODE_NAME_PROP, "node1", |
| ZkStateReader.COLLECTION_PROP, "perf" + j, |
| ZkStateReader.NUM_SHARDS_PROP, "1"); |
| ZkDistributedQueue q = overseers.get(0).getStateUpdateQueue(); |
| q.offer(Utils.toJSON(m)); |
| if (j >= MAX_COLLECTIONS - 1) j = 0; |
| if (k >= MAX_CORES - 1) k = 0; |
| if (i > 0 && i % 100 == 0) log.info("Published {} items", i); |
| } |
| |
| // let's publish a sentinel collection which we'll use to wait for overseer to complete operations |
| ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.STATE.toLower(), |
| ZkStateReader.STATE_PROP, Replica.State.ACTIVE.toString(), |
| ZkStateReader.NODE_NAME_PROP, "node1", |
| ZkStateReader.CORE_NAME_PROP, "core1", |
| ZkStateReader.CORE_NODE_NAME_PROP, "node1", |
| ZkStateReader.COLLECTION_PROP, "perf_sentinel", |
| ZkStateReader.NUM_SHARDS_PROP, "1", |
| ZkStateReader.BASE_URL_PROP, "http://" + "node1" |
| + "/solr/"); |
| ZkDistributedQueue q = overseers.get(0).getStateUpdateQueue(); |
| q.offer(Utils.toJSON(m)); |
| |
| Timer t = new Timer(); |
| Timer.Context context = t.time(); |
| try { |
| overseerClient = electNewOverseer(server.getZkAddress()); |
| assertTrue(overseers.size() > 0); |
| |
| reader.waitForState("perf_sentinel", 15000, TimeUnit.MILLISECONDS, (liveNodes, collectionState) -> collectionState != null); |
| |
| } finally { |
| context.stop(); |
| } |
| |
| log.info("Overseer loop finished processing: "); |
| printTimingStats(t); |
| |
| Overseer overseer = overseers.get(0); |
| Stats stats = overseer.getStats(); |
| |
| String[] interestingOps = {"state", "update_state", "am_i_leader", ""}; |
| Arrays.sort(interestingOps); |
| for (Map.Entry<String, Stats.Stat> entry : stats.getStats().entrySet()) { |
| String op = entry.getKey(); |
| if (Arrays.binarySearch(interestingOps, op) < 0) |
| continue; |
| Stats.Stat stat = entry.getValue(); |
| if (log.isInfoEnabled()) { |
| log.info("op: {}, success: {}, failure: {}", op, stat.success.get(), stat.errors.get()); |
| } |
| Timer timer = stat.requestTime; |
| printTimingStats(timer); |
| } |
| |
| } finally { |
| close(overseerClient); |
| close(mockController); |
| close(reader); |
| } |
| } |
| |
| private void printTimingStats(Timer timer) { |
| Snapshot snapshot = timer.getSnapshot(); |
| if (log.isInfoEnabled()) { |
| log.info("\t avgRequestsPerSecond: {}", timer.getMeanRate()); |
| log.info("\t 5minRateRequestsPerSecond: {}", timer.getFiveMinuteRate()); // nowarn |
| log.info("\t 15minRateRequestsPerSecond: {}", timer.getFifteenMinuteRate()); // nowarn |
| log.info("\t avgTimePerRequest: {}", nsToMs(snapshot.getMean())); // nowarn |
| log.info("\t medianRequestTime: {}", nsToMs(snapshot.getMedian())); // nowarn |
| log.info("\t 75thPcRequestTime: {}", nsToMs(snapshot.get75thPercentile())); // nowarn |
| log.info("\t 95thPcRequestTime: {}", nsToMs(snapshot.get95thPercentile())); // nowarn |
| log.info("\t 99thPcRequestTime: {}", nsToMs(snapshot.get99thPercentile())); // nowarn |
| log.info("\t 999thPcRequestTime: {}", nsToMs(snapshot.get999thPercentile())); // nowarn |
| } |
| } |
| |
| private static long nsToMs(double ns) { |
| return TimeUnit.MILLISECONDS.convert((long)ns, TimeUnit.NANOSECONDS); |
| } |
| |
| private void close(MockZKController mockController) { |
| if (mockController != null) { |
| mockController.close(); |
| } |
| } |
| |
| |
| @Test |
| public void testReplay() throws Exception{ |
| |
| SolrZkClient overseerClient = null; |
| ZkStateReader reader = null; |
| |
| try { |
| |
| ZkController.createClusterZkNodes(zkClient); |
| |
| reader = new ZkStateReader(zkClient); |
| reader.createClusterStateWatchersAndUpdate(); |
| //prepopulate work queue with some items to emulate previous overseer died before persisting state |
| DistributedQueue queue = Overseer.getInternalWorkQueue(zkClient, new Stats()); |
| |
| ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, CollectionParams.CollectionAction.CREATE.toLower(), |
| "name", COLLECTION, |
| ZkStateReader.REPLICATION_FACTOR, "1", |
| ZkStateReader.NUM_SHARDS_PROP, "1", |
| "createNodeSet", ""); |
| queue.offer(Utils.toJSON(m)); |
| m = new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.STATE.toLower(), |
| ZkStateReader.NODE_NAME_PROP, "127.0.0.1:8983_solr", |
| ZkStateReader.SHARD_ID_PROP, "shard1", |
| ZkStateReader.COLLECTION_PROP, COLLECTION, |
| ZkStateReader.CORE_NAME_PROP, "core1", |
| ZkStateReader.ROLES_PROP, "", |
| ZkStateReader.STATE_PROP, Replica.State.RECOVERING.toString()); |
| queue.offer(Utils.toJSON(m)); |
| m = new ZkNodeProps(Overseer.QUEUE_OPERATION, "state", |
| ZkStateReader.NODE_NAME_PROP, "node1:8983_", |
| ZkStateReader.SHARD_ID_PROP, "shard1", |
| ZkStateReader.COLLECTION_PROP, COLLECTION, |
| ZkStateReader.CORE_NAME_PROP, "core2", |
| ZkStateReader.ROLES_PROP, "", |
| ZkStateReader.STATE_PROP, Replica.State.RECOVERING.toString()); |
| queue.offer(Utils.toJSON(m)); |
| |
| overseerClient = electNewOverseer(server.getZkAddress()); |
| |
| //submit to proper queue |
| queue = overseers.get(0).getStateUpdateQueue(); |
| m = new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.STATE.toLower(), |
| ZkStateReader.NODE_NAME_PROP, "127.0.0.1:8983_solr", |
| ZkStateReader.SHARD_ID_PROP, "shard1", |
| ZkStateReader.COLLECTION_PROP, COLLECTION, |
| ZkStateReader.CORE_NAME_PROP, "core3", |
| ZkStateReader.ROLES_PROP, "", |
| ZkStateReader.STATE_PROP, Replica.State.RECOVERING.toString()); |
| queue.offer(Utils.toJSON(m)); |
| |
| reader.waitForState(COLLECTION, 1000, TimeUnit.MILLISECONDS, |
| (liveNodes, collectionState) -> collectionState != null && collectionState.getSlice("shard1") != null |
| && collectionState.getSlice("shard1").getReplicas().size() == 3); |
| |
| assertNotNull(reader.getClusterState().getCollection(COLLECTION).getSlice("shard1")); |
| assertEquals(3, reader.getClusterState().getCollection(COLLECTION).getSlice("shard1").getReplicasMap().size()); |
| } finally { |
| close(overseerClient); |
| close(reader); |
| } |
| } |
| |
| @Test |
| public void testExternalClusterStateChangeBehavior() throws Exception { |
| |
| ZkStateReader reader = null; |
| SolrZkClient overseerClient = null; |
| |
| try { |
| |
| ZkController.createClusterZkNodes(zkClient); |
| |
| zkClient.create("/collections/test", null, CreateMode.PERSISTENT, true); |
| |
| reader = new ZkStateReader(zkClient); |
| reader.createClusterStateWatchersAndUpdate(); |
| |
| overseerClient = electNewOverseer(server.getZkAddress()); |
| |
| ZkDistributedQueue q = overseers.get(0).getStateUpdateQueue(); |
| |
| |
| ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, CollectionParams.CollectionAction.CREATE.toLower(), |
| "name", "c1", |
| ZkStateReader.REPLICATION_FACTOR, "1", |
| ZkStateReader.NUM_SHARDS_PROP, "1", |
| "createNodeSet", ""); |
| q.offer(Utils.toJSON(m)); |
| |
| m = new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.STATE.toLower(), |
| ZkStateReader.SHARD_ID_PROP, "shard1", |
| ZkStateReader.NODE_NAME_PROP, "127.0.0.1:8983_solr", |
| ZkStateReader.COLLECTION_PROP, "c1", |
| ZkStateReader.CORE_NAME_PROP, "core1", |
| ZkStateReader.CORE_NODE_NAME_PROP, "core_node1", |
| ZkStateReader.ROLES_PROP, "", |
| ZkStateReader.STATE_PROP, Replica.State.DOWN.toString()); |
| |
| q.offer(Utils.toJSON(m)); |
| |
| waitForCollections(reader, "c1"); |
| verifyReplicaStatus(reader, "c1", "shard1", "core_node1", Replica.State.DOWN); |
| |
| m = new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.STATE.toLower(), |
| ZkStateReader.SHARD_ID_PROP, "shard1", |
| ZkStateReader.NODE_NAME_PROP, "127.0.0.1:8983_solr", |
| ZkStateReader.COLLECTION_PROP, "c1", |
| ZkStateReader.CORE_NAME_PROP, "core1", |
| ZkStateReader.ROLES_PROP, "", |
| ZkStateReader.STATE_PROP, Replica.State.RECOVERING.toString()); |
| |
| q.offer(Utils.toJSON(m)); |
| |
| |
| m = new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.STATE.toLower(), |
| ZkStateReader.SHARD_ID_PROP, "shard1", |
| ZkStateReader.NODE_NAME_PROP, "127.0.0.1:8983_solr", |
| ZkStateReader.COLLECTION_PROP, "c1", |
| ZkStateReader.CORE_NAME_PROP, "core1", |
| ZkStateReader.ROLES_PROP, "", |
| ZkStateReader.STATE_PROP, Replica.State.ACTIVE.toString()); |
| |
| q.offer(Utils.toJSON(m)); |
| |
| Stat stat = new Stat(); |
| byte[] data = zkClient.getData("/clusterstate.json", null, stat, true); |
| // Simulate an external modification |
| zkClient.setData("/clusterstate.json", data, true); |
| |
| m = new ZkNodeProps(Overseer.QUEUE_OPERATION, CollectionParams.CollectionAction.CREATE.toLower(), |
| "name", "test", |
| ZkStateReader.NUM_SHARDS_PROP, "1", |
| ZkStateReader.REPLICATION_FACTOR, "1", |
| DocCollection.STATE_FORMAT, "2" |
| ); |
| q.offer(Utils.toJSON(m)); |
| |
| m = new ZkNodeProps(Overseer.QUEUE_OPERATION, CollectionParams.CollectionAction.CREATESHARD.toLower(), |
| "collection", "test", |
| ZkStateReader.SHARD_ID_PROP, "x", |
| ZkStateReader.REPLICATION_FACTOR, "1" |
| ); |
| q.offer(Utils.toJSON(m)); |
| |
| m = new ZkNodeProps(Overseer.QUEUE_OPERATION, CollectionParams.CollectionAction.ADDREPLICA.toLower(), |
| "collection", "test", |
| ZkStateReader.SHARD_ID_PROP, "x", |
| ZkStateReader.CORE_NODE_NAME_PROP, "core_node1", |
| ZkStateReader.NODE_NAME_PROP, "127.0.0.1:8983_solr", |
| ZkStateReader.CORE_NAME_PROP, "core1", |
| ZkStateReader.STATE_PROP, Replica.State.DOWN.toString() |
| ); |
| q.offer(Utils.toJSON(m)); |
| |
| waitForCollections(reader, "test"); |
| verifyReplicaStatus(reader, "test", "x", "core_node1", Replica.State.DOWN); |
| |
| waitForCollections(reader, "c1"); |
| verifyReplicaStatus(reader, "c1", "shard1", "core_node1", Replica.State.ACTIVE); |
| |
| } finally { |
| close(overseerClient); |
| close(reader); |
| } |
| } |
| |
| private void close(ZkStateReader reader) { |
| if (reader != null) { |
| reader.close(); |
| } |
| } |
| |
| private void close(SolrZkClient client) throws InterruptedException { |
| if (client != null) { |
| client.close(); |
| } |
| } |
| |
| private SolrZkClient electNewOverseer(String address) |
| throws InterruptedException, TimeoutException, IOException, |
| KeeperException, ParserConfigurationException, SAXException, NoSuchFieldException, SecurityException { |
| SolrZkClient zkClient = new SolrZkClient(address, TIMEOUT); |
| zkClients.add(zkClient); |
| ZkStateReader reader = new ZkStateReader(zkClient); |
| readers.add(reader); |
| LeaderElector overseerElector = new LeaderElector(zkClient); |
| if (overseers.size() > 0) { |
| overseers.get(0).close(); |
| overseers.get(0).getZkStateReader().getZkClient().close(); |
| } |
| UpdateShardHandler updateShardHandler = new UpdateShardHandler(UpdateShardHandlerConfig.DEFAULT); |
| updateShardHandlers.add(updateShardHandler); |
| HttpShardHandlerFactory httpShardHandlerFactory = new HttpShardHandlerFactory(); |
| httpShardHandlerFactory.init(new PluginInfo("shardHandlerFactory", Collections.emptyMap())); |
| httpShardHandlerFactorys.add(httpShardHandlerFactory); |
| |
| ZkController zkController = createMockZkController(address, null, reader); |
| zkControllers.add(zkController); |
| Overseer overseer = new Overseer((HttpShardHandler) httpShardHandlerFactory.getShardHandler(), updateShardHandler, "/admin/cores", reader, zkController, |
| new CloudConfig.CloudConfigBuilder("127.0.0.1", 8983, "").build()); |
| overseers.add(overseer); |
| ElectionContext ec = new OverseerElectionContext(zkClient, overseer, |
| address.replaceAll("/", "_")); |
| overseerElector.setup(ec); |
| overseerElector.joinElection(ec, false); |
| return zkClient; |
| } |
| |
| private ZkController createMockZkController(String zkAddress, SolrZkClient zkClient, ZkStateReader reader) throws InterruptedException, NoSuchFieldException, SecurityException, SessionExpiredException { |
| ZkController zkController = mock(ZkController.class); |
| |
| if (zkClient == null) { |
| SolrZkClient newZkClient = new SolrZkClient(server.getZkAddress(), AbstractZkTestCase.TIMEOUT); |
| Mockito.doAnswer( |
| new Answer<Void>() { |
| public Void answer(InvocationOnMock invocation) { |
| newZkClient.close(); |
| return null; |
| }}).when(zkController).close(); |
| zkClient = newZkClient; |
| } else { |
| doNothing().when(zkController).close(); |
| } |
| |
| CoreContainer mockAlwaysUpCoreContainer = mock(CoreContainer.class, |
| Mockito.withSettings().defaultAnswer(Mockito.CALLS_REAL_METHODS)); |
| SolrMetricManager mockMetricManager = mock(SolrMetricManager.class); |
| when(mockAlwaysUpCoreContainer.getMetricManager()).thenReturn(mockMetricManager); |
| when(mockAlwaysUpCoreContainer.isShutDown()).thenReturn(testDone); // Allow retry on session expiry |
| when(mockAlwaysUpCoreContainer.getResourceLoader()).thenReturn(new SolrResourceLoader(createTempDir())); |
| FieldSetter.setField(zkController, ZkController.class.getDeclaredField("zkClient"), zkClient); |
| FieldSetter.setField(zkController, ZkController.class.getDeclaredField("cc"), mockAlwaysUpCoreContainer); |
| when(zkController.getCoreContainer()).thenReturn(mockAlwaysUpCoreContainer); |
| when(zkController.getZkClient()).thenReturn(zkClient); |
| when(zkController.getZkStateReader()).thenReturn(reader); |
| |
| when(zkController.getLeaderProps(anyString(), anyString(), anyInt())).thenCallRealMethod(); |
| when(zkController.getLeaderProps(anyString(), anyString(), anyInt(), anyBoolean())).thenCallRealMethod(); |
| doReturn(getCloudDataProvider(zkAddress, zkClient, reader)) |
| .when(zkController).getSolrCloudManager(); |
| return zkController; |
| } |
| |
| private SolrCloudManager getCloudDataProvider(String zkAddress, SolrZkClient zkClient, ZkStateReader reader) { |
| CloudSolrClient client = new CloudSolrClient.Builder(Collections.singletonList(zkAddress), Optional.empty()).withSocketTimeout(30000).withConnectionTimeout(15000).build(); |
| solrClients.add(client); |
| SolrClientCloudManager sccm = new SolrClientCloudManager(new ZkDistributedQueueFactory(zkClient), client); |
| sccm.getClusterStateProvider().connect(); |
| return sccm; |
| } |
| |
| @Test |
| public void testRemovalOfLastReplica() throws Exception { |
| |
| final Integer numReplicas = 1+random().nextInt(4); // between 1 and 4 replicas |
| final Integer numShards = 1+random().nextInt(4); // between 1 and 4 shards |
| |
| ZkStateReader zkStateReader = null; |
| SolrZkClient overseerClient = null; |
| try { |
| |
| ZkController.createClusterZkNodes(zkClient); |
| |
| zkStateReader = new ZkStateReader(zkClient); |
| zkStateReader.createClusterStateWatchersAndUpdate(); |
| |
| overseerClient = electNewOverseer(server.getZkAddress()); |
| |
| ZkDistributedQueue q = overseers.get(0).getStateUpdateQueue(); |
| |
| // create collection |
| { |
| final Integer maxShardsPerNode = numReplicas * numShards; |
| ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, CollectionParams.CollectionAction.CREATE.toLower(), |
| "name", COLLECTION, |
| ZkStateReader.NUM_SHARDS_PROP, numShards.toString(), |
| ZkStateReader.REPLICATION_FACTOR, "1", |
| ZkStateReader.MAX_SHARDS_PER_NODE, maxShardsPerNode.toString() |
| ); |
| q.offer(Utils.toJSON(m)); |
| } |
| waitForCollections(zkStateReader, COLLECTION); |
| |
| // create nodes with state recovering |
| for (int rr = 1; rr <= numReplicas; ++rr) { |
| for (int ss = 1; ss <= numShards; ++ss) { |
| final int N = (numReplicas-rr)*numShards + ss; |
| ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.STATE.toLower(), |
| ZkStateReader.SHARD_ID_PROP, "shard"+ss, |
| ZkStateReader.NODE_NAME_PROP, "127.0.0.1:8983_solr", |
| ZkStateReader.COLLECTION_PROP, COLLECTION, |
| ZkStateReader.CORE_NAME_PROP, "core"+N, |
| ZkStateReader.CORE_NODE_NAME_PROP, "core_node"+N, |
| ZkStateReader.ROLES_PROP, "", |
| ZkStateReader.STATE_PROP, Replica.State.RECOVERING.toString()); |
| |
| q.offer(Utils.toJSON(m)); |
| } |
| } |
| // verify recovering |
| for (int rr = 1; rr <= numReplicas; ++rr) { |
| for (int ss = 1; ss <= numShards; ++ss) { |
| final int N = (numReplicas-rr)*numShards + ss; |
| verifyReplicaStatus(zkStateReader, COLLECTION, "shard"+ss, "core_node"+N, Replica.State.RECOVERING); |
| } |
| } |
| |
| // publish node states (active) |
| for (int rr = 1; rr <= numReplicas; ++rr) { |
| for (int ss = 1; ss <= numShards; ++ss) { |
| final int N = (numReplicas-rr)*numShards + ss; |
| ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.STATE.toLower(), |
| ZkStateReader.SHARD_ID_PROP, "shard"+ss, |
| ZkStateReader.NODE_NAME_PROP, "127.0.0.1:8983_solr", |
| ZkStateReader.COLLECTION_PROP, COLLECTION, |
| ZkStateReader.CORE_NAME_PROP, "core"+N, |
| ZkStateReader.ROLES_PROP, "", |
| ZkStateReader.STATE_PROP, Replica.State.ACTIVE.toString()); |
| |
| q.offer(Utils.toJSON(m)); |
| } |
| } |
| // verify active |
| for (int rr = 1; rr <= numReplicas; ++rr) { |
| for (int ss = 1; ss <= numShards; ++ss) { |
| final int N = (numReplicas-rr)*numShards + ss; |
| verifyReplicaStatus(zkStateReader, COLLECTION, "shard"+ss, "core_node"+N, Replica.State.ACTIVE); |
| } |
| } |
| |
| // delete node |
| for (int rr = 1; rr <= numReplicas; ++rr) { |
| for (int ss = 1; ss <= numShards; ++ss) { |
| final int N = (numReplicas-rr)*numShards + ss; |
| ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.DELETECORE.toLower(), |
| ZkStateReader.COLLECTION_PROP, COLLECTION, |
| ZkStateReader.CORE_NODE_NAME_PROP, "core_node"+N); |
| |
| q.offer(Utils.toJSON(m)); |
| |
| { |
| String shard = "shard"+ss; |
| zkStateReader.waitForState(COLLECTION, 15000, TimeUnit.MILLISECONDS, (liveNodes, collectionState) -> collectionState != null && (collectionState.getSlice(shard) == null || collectionState.getSlice(shard).getReplicasMap().get("core_node"+N) == null)); |
| } |
| |
| final DocCollection docCollection = zkStateReader.getClusterState().getCollection(COLLECTION); |
| assertTrue("found no "+ COLLECTION, (null != docCollection)); |
| |
| final Slice slice = docCollection.getSlice("shard"+ss); |
| assertTrue("found no "+ COLLECTION +" shard"+ss+" slice after removal of replica "+rr+" of "+numReplicas, (null != slice)); |
| |
| final Collection<Replica> replicas = slice.getReplicas(); |
| assertEquals("wrong number of "+ COLLECTION +" shard"+ss+" replicas left, replicas="+replicas, numReplicas-rr, replicas.size()); |
| } |
| } |
| |
| } finally { |
| |
| close(overseerClient); |
| close(zkStateReader); |
| } |
| } |
| |
| @Test |
| public void testLatchWatcher() throws InterruptedException { |
| OverseerTaskQueue.LatchWatcher latch1 = new OverseerTaskQueue.LatchWatcher(); |
| long before = System.nanoTime(); |
| latch1.await(100); |
| long after = System.nanoTime(); |
| assertTrue(TimeUnit.NANOSECONDS.toMillis(after-before) > 50); |
| assertTrue(TimeUnit.NANOSECONDS.toMillis(after-before) < 500);// Mostly to make sure the millis->nanos->millis is not broken |
| latch1.process(new WatchedEvent(new WatcherEvent(1, 1, "/foo/bar"))); |
| before = System.nanoTime(); |
| latch1.await(10000);// Expecting no wait |
| after = System.nanoTime(); |
| assertTrue(TimeUnit.NANOSECONDS.toMillis(after-before) < 1000); |
| |
| final AtomicBoolean expectedEventProcessed = new AtomicBoolean(false); |
| final AtomicBoolean doneWaiting = new AtomicBoolean(false); |
| final OverseerTaskQueue.LatchWatcher latch2 = new OverseerTaskQueue.LatchWatcher(Event.EventType.NodeCreated); |
| Thread t = new Thread(()->{ |
| //Process an event of a different type first, this shouldn't release the latch |
| latch2.process(new WatchedEvent(new WatcherEvent(Event.EventType.NodeDeleted.getIntValue(), 1, "/foo/bar"))); |
| |
| assertFalse("Latch shouldn't have been released", doneWaiting.get()); |
| // Now process the correct type of event |
| expectedEventProcessed.set(true); |
| latch2.process(new WatchedEvent(new WatcherEvent(Event.EventType.NodeCreated.getIntValue(), 1, "/foo/bar"))); |
| }); |
| t.start(); |
| before = System.nanoTime(); |
| latch2.await(10000); // It shouldn't wait this long, t should notify the lock |
| after = System.nanoTime(); |
| doneWaiting.set(true); |
| assertTrue(expectedEventProcessed.get()); |
| assertTrue(TimeUnit.NANOSECONDS.toMillis(after-before) < 1000); |
| } |
| |
| } |
