| /* |
| * 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.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Queue; |
| import java.util.Set; |
| import java.util.concurrent.ArrayBlockingQueue; |
| import java.util.concurrent.TimeUnit; |
| |
| import org.apache.http.client.HttpClient; |
| import org.apache.solr.SolrTestCaseJ4; |
| import org.apache.solr.client.solrj.SolrResponse; |
| import org.apache.solr.client.solrj.cloud.DistribStateManager; |
| import org.apache.solr.client.solrj.cloud.SolrCloudManager; |
| import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig; |
| import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData; |
| import org.apache.solr.client.solrj.impl.ClusterStateProvider; |
| import org.apache.solr.client.solrj.impl.Http2SolrClient; |
| import org.apache.solr.cloud.Overseer.LeaderStatus; |
| import org.apache.solr.cloud.OverseerTaskQueue.QueueEvent; |
| import org.apache.solr.cloud.api.collections.OverseerCollectionMessageHandler; |
| import org.apache.solr.common.cloud.Aliases; |
| import org.apache.solr.common.cloud.ClusterState; |
| import org.apache.solr.common.cloud.DocCollection; |
| import org.apache.solr.common.cloud.DocRouter; |
| 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.ZkNodeProps; |
| import org.apache.solr.common.cloud.ZkStateReader; |
| import org.apache.solr.common.params.CollectionParams; |
| import org.apache.solr.common.params.CoreAdminParams; |
| import org.apache.solr.common.params.CoreAdminParams.CoreAdminAction; |
| import org.apache.solr.common.params.ModifiableSolrParams; |
| import org.apache.solr.common.util.ObjectCache; |
| import org.apache.solr.common.util.StrUtils; |
| import org.apache.solr.common.util.TimeSource; |
| import org.apache.solr.common.util.Utils; |
| import org.apache.solr.core.CoreContainer; |
| import org.apache.solr.handler.component.HttpShardHandler; |
| import org.apache.solr.handler.component.HttpShardHandlerFactory; |
| import org.apache.solr.handler.component.ShardRequest; |
| import org.apache.solr.metrics.SolrMetricsContext; |
| import org.apache.solr.update.UpdateShardHandler; |
| import org.apache.solr.util.TimeOut; |
| import org.apache.zookeeper.CreateMode; |
| import org.apache.zookeeper.KeeperException; |
| import org.apache.zookeeper.data.Stat; |
| import org.junit.After; |
| import org.junit.AfterClass; |
| import org.junit.Before; |
| import org.junit.BeforeClass; |
| import org.junit.Test; |
| import org.mockito.ArgumentCaptor; |
| import org.mockito.Mockito; |
| import org.mockito.invocation.InvocationOnMock; |
| import org.mockito.stubbing.Answer; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import static org.apache.solr.common.params.CollectionAdminParams.CLUSTER; |
| import static org.apache.solr.common.params.CollectionAdminParams.DEFAULTS; |
| import static org.apache.solr.common.params.CollectionAdminParams.USE_LEGACY_REPLICA_ASSIGNMENT; |
| import static org.mockito.Mockito.any; |
| import static org.mockito.Mockito.anyBoolean; |
| import static org.mockito.Mockito.anyInt; |
| import static org.mockito.Mockito.anyLong; |
| import static org.mockito.Mockito.anyString; |
| import static org.mockito.Mockito.doAnswer; |
| import static org.mockito.Mockito.isNull; |
| import static org.mockito.Mockito.mock; |
| import static org.mockito.Mockito.reset; |
| import static org.mockito.Mockito.times; |
| import static org.mockito.Mockito.verify; |
| import static org.mockito.Mockito.when; |
| |
| public class OverseerCollectionConfigSetProcessorTest extends SolrTestCaseJ4 { |
| |
| private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); |
| |
| private static final String ADMIN_PATH = "/admin/cores"; |
| private static final String COLLECTION_NAME = "mycollection"; |
| private static final String CONFIG_NAME = "myconfig"; |
| |
| private static OverseerTaskQueue workQueueMock; |
| private static OverseerTaskQueue stateUpdateQueueMock; |
| private static Overseer overseerMock; |
| private static ZkController zkControllerMock; |
| private static SolrCloudManager cloudDataProviderMock; |
| private static ClusterStateProvider clusterStateProviderMock; |
| private static DistributedMap runningMapMock; |
| private static DistributedMap completedMapMock; |
| private static DistributedMap failureMapMock; |
| private static HttpShardHandlerFactory shardHandlerFactoryMock; |
| private static HttpShardHandler shardHandlerMock; |
| private static ZkStateReader zkStateReaderMock; |
| private static ClusterState clusterStateMock; |
| private static SolrZkClient solrZkClientMock; |
| private static DistribStateManager stateManagerMock; |
| private static SolrCloudManager cloudManagerMock; |
| private static DistribStateManager distribStateManagerMock; |
| private static CoreContainer coreContainerMock; |
| private static UpdateShardHandler updateShardHandlerMock; |
| private static HttpClient httpClientMock; |
| private static SolrMetricsContext solrMetricsContextMock; |
| |
| private static ObjectCache objectCache; |
| private static AutoScalingConfig autoScalingConfig = new AutoScalingConfig(Collections.emptyMap()); |
| private Map<String, byte[]> zkClientData = new HashMap<>(); |
| private final Map<String, ClusterState.CollectionRef> collectionsSet = new HashMap<>(); |
| private final List<ZkNodeProps> replicas = new ArrayList<>(); |
| private SolrResponse lastProcessMessageResult; |
| |
| |
| private OverseerCollectionConfigSetProcessorToBeTested underTest; |
| |
| private Thread thread; |
| private Queue<QueueEvent> queue = new ArrayBlockingQueue<>(10); |
| |
| private static class OverseerCollectionConfigSetProcessorToBeTested extends |
| OverseerCollectionConfigSetProcessor { |
| |
| |
| public OverseerCollectionConfigSetProcessorToBeTested(ZkStateReader zkStateReader, |
| String myId, HttpShardHandlerFactory shardHandlerFactory, |
| String adminPath, |
| OverseerTaskQueue workQueue, DistributedMap runningMap, |
| Overseer overseer, |
| DistributedMap completedMap, |
| DistributedMap failureMap, |
| SolrMetricsContext solrMetricsContext) { |
| super(zkStateReader, myId, shardHandlerFactory, adminPath, new Stats(), overseer, new OverseerNodePrioritizer(zkStateReader, overseer.getStateUpdateQueue(), adminPath, shardHandlerFactory, null), workQueue, runningMap, completedMap, failureMap, solrMetricsContext); |
| } |
| |
| @Override |
| protected LeaderStatus amILeader() { |
| return LeaderStatus.YES; |
| } |
| |
| } |
| |
| @BeforeClass |
| public static void setUpOnce() throws Exception { |
| assumeWorkingMockito(); |
| |
| workQueueMock = mock(OverseerTaskQueue.class); |
| stateUpdateQueueMock = mock(OverseerTaskQueue.class); |
| runningMapMock = mock(DistributedMap.class); |
| completedMapMock = mock(DistributedMap.class); |
| failureMapMock = mock(DistributedMap.class); |
| shardHandlerFactoryMock = mock(HttpShardHandlerFactory.class); |
| shardHandlerMock = mock(HttpShardHandler.class); |
| zkStateReaderMock = mock(ZkStateReader.class); |
| clusterStateMock = mock(ClusterState.class); |
| solrZkClientMock = mock(SolrZkClient.class); |
| overseerMock = mock(Overseer.class); |
| zkControllerMock = mock(ZkController.class); |
| cloudDataProviderMock = mock(SolrCloudManager.class); |
| objectCache = new ObjectCache(); |
| clusterStateProviderMock = mock(ClusterStateProvider.class); |
| stateManagerMock = mock(DistribStateManager.class); |
| cloudManagerMock = mock(SolrCloudManager.class); |
| distribStateManagerMock = mock(DistribStateManager.class); |
| coreContainerMock = mock(CoreContainer.class); |
| updateShardHandlerMock = mock(UpdateShardHandler.class); |
| httpClientMock = mock(HttpClient.class); |
| solrMetricsContextMock = mock(SolrMetricsContext.class); |
| } |
| |
| @AfterClass |
| public static void tearDownOnce() { |
| workQueueMock = null; |
| stateUpdateQueueMock = null; |
| runningMapMock = null; |
| completedMapMock = null; |
| failureMapMock = null; |
| shardHandlerFactoryMock = null; |
| shardHandlerMock = null; |
| zkStateReaderMock = null; |
| clusterStateMock = null; |
| solrZkClientMock = null; |
| overseerMock = null; |
| zkControllerMock = null; |
| cloudDataProviderMock = null; |
| clusterStateProviderMock = null; |
| stateManagerMock = null;; |
| cloudManagerMock = null; |
| distribStateManagerMock = null; |
| coreContainerMock = null; |
| updateShardHandlerMock = null; |
| httpClientMock = null; |
| solrMetricsContextMock = null; |
| } |
| |
| @Before |
| public void setUp() throws Exception { |
| super.setUp(); |
| queue.clear(); |
| reset(workQueueMock); |
| reset(stateUpdateQueueMock); |
| reset(runningMapMock); |
| reset(completedMapMock); |
| reset(failureMapMock); |
| reset(shardHandlerFactoryMock); |
| reset(shardHandlerMock); |
| reset(zkStateReaderMock); |
| reset(clusterStateMock); |
| reset(solrZkClientMock); |
| reset(overseerMock); |
| reset(zkControllerMock); |
| reset(cloudDataProviderMock); |
| objectCache.clear(); |
| when(cloudDataProviderMock.getObjectCache()).thenReturn(objectCache); |
| when(cloudDataProviderMock.getTimeSource()).thenReturn(TimeSource.NANO_TIME); |
| reset(clusterStateProviderMock); |
| reset(stateManagerMock); |
| reset(cloudManagerMock); |
| reset(distribStateManagerMock); |
| reset(coreContainerMock); |
| reset(updateShardHandlerMock); |
| reset(httpClientMock); |
| reset(solrMetricsContextMock); |
| |
| zkClientData.clear(); |
| collectionsSet.clear(); |
| replicas.clear(); |
| } |
| |
| @After |
| public void tearDown() throws Exception { |
| stopComponentUnderTest(); |
| super.tearDown(); |
| } |
| |
| protected Set<String> commonMocks(int liveNodesCount) throws Exception { |
| when(shardHandlerFactoryMock.getShardHandler()).thenReturn(shardHandlerMock); |
| when(shardHandlerFactoryMock.getShardHandler(any(Http2SolrClient.class))).thenReturn(shardHandlerMock); |
| when(shardHandlerFactoryMock.getShardHandler(any(HttpClient.class))).thenReturn(shardHandlerMock); |
| when(workQueueMock.peekTopN(anyInt(), any(), anyLong())).thenAnswer(invocation -> { |
| Object result; |
| int count = 0; |
| while ((result = queue.peek()) == null) { |
| Thread.sleep(1000); |
| count++; |
| if (count > 1) return null; |
| } |
| |
| return Arrays.asList(result); |
| }); |
| |
| when(workQueueMock.getTailId()).thenAnswer(invocation -> { |
| Object result = null; |
| @SuppressWarnings({"rawtypes"}) |
| Iterator iter = queue.iterator(); |
| while(iter.hasNext()) { |
| result = iter.next(); |
| } |
| return result==null ? null : ((QueueEvent)result).getId(); |
| }); |
| |
| when(workQueueMock.peek(true)).thenAnswer(invocation -> { |
| Object result; |
| while ((result = queue.peek()) == null) { |
| Thread.sleep(1000); |
| } |
| return result; |
| }); |
| |
| doAnswer(invocation -> { |
| queue.remove(invocation.getArgument(0)); |
| return null; |
| }).when(workQueueMock).remove(any(QueueEvent.class)); |
| |
| when(workQueueMock.poll()).thenAnswer(invocation -> { |
| queue.poll(); |
| return null; |
| }); |
| |
| when(zkStateReaderMock.getZkClient()).thenReturn(solrZkClientMock); |
| when(zkStateReaderMock.getClusterState()).thenReturn(clusterStateMock); |
| when(zkStateReaderMock.getAutoScalingConfig()).thenReturn(autoScalingConfig); |
| when(zkStateReaderMock.getAliases()).thenReturn(Aliases.EMPTY); |
| |
| when(clusterStateMock.getCollection(anyString())).thenAnswer(invocation -> { |
| String key = invocation.getArgument(0); |
| if (!collectionsSet.containsKey(key)) return null; |
| DocCollection docCollection = collectionsSet.get(key).get(); |
| Map<String, Map<String, Replica>> slices = new HashMap<>(); |
| for (ZkNodeProps replica : replicas) { |
| if (!key.equals(replica.getStr(ZkStateReader.COLLECTION_PROP))) continue; |
| |
| String slice = replica.getStr(ZkStateReader.SHARD_ID_PROP); |
| if (!slices.containsKey(slice)) slices.put(slice, new HashMap<>()); |
| String replicaName = replica.getStr(ZkStateReader.CORE_NAME_PROP); |
| slices.get(slice).put(replicaName, new Replica(replicaName, replica.getProperties(), docCollection.getName(), slice)); |
| } |
| |
| Map<String, Slice> slicesMap = new HashMap<>(); |
| for (Map.Entry<String, Map<String, Replica>> entry : slices.entrySet()) { |
| slicesMap.put(entry.getKey(), new Slice(entry.getKey(), entry.getValue(), null,docCollection.getName())); |
| } |
| |
| return docCollection.copyWithSlices(slicesMap); |
| }); |
| final Set<String> liveNodes = new HashSet<>(); |
| for (int i = 0; i < liveNodesCount; i++) { |
| final String address = "localhost:" + (8963 + i) + "_solr"; |
| liveNodes.add(address); |
| |
| when(zkStateReaderMock.getBaseUrlForNodeName(address)).thenAnswer(invocation -> address.replaceAll("_", "/")); |
| } |
| |
| when(zkStateReaderMock.getClusterProperty("legacyCloud", "false")).thenReturn("false"); |
| |
| when(solrZkClientMock.getZkClientTimeout()).thenReturn(30000); |
| |
| when(clusterStateMock.hasCollection(anyString())).thenAnswer(invocation -> { |
| String key = invocation.getArgument(0); |
| return collectionsSet.containsKey(key); |
| }); |
| |
| when(clusterStateMock.getLiveNodes()).thenReturn(liveNodes); |
| |
| when(solrZkClientMock.setData(anyString(), any(), anyInt(), anyBoolean())).then(invocation -> { |
| System.out.println("set data: " + invocation.getArgument(0) + " " + invocation.getArgument(1)); |
| if (invocation.getArgument(1) == null) { |
| zkClientData.put(invocation.getArgument(0), new byte[0]); |
| } else { |
| zkClientData.put(invocation.getArgument(0), invocation.getArgument(1)); |
| } |
| return null; |
| }); |
| |
| when(solrZkClientMock.getData(anyString(), any(), any(), anyBoolean())).thenAnswer(invocation -> { |
| byte[] data = zkClientData.get(invocation.getArgument(0)); |
| if (data == null || data.length == 0) { |
| return null; |
| } |
| return data; |
| }); |
| |
| when(solrZkClientMock.create(any(), any(), any(), anyBoolean())).thenAnswer(invocation -> { |
| zkClientData.put(invocation.getArgument(0), invocation.getArgument(1)); |
| return invocation.getArgument(0); |
| }); |
| |
| when(solrZkClientMock.exists(any(String.class), anyBoolean())).thenAnswer(invocation -> { |
| String key = invocation.getArgument(0); |
| return zkClientData.containsKey(key); |
| }); |
| |
| when(overseerMock.getZkController()).thenReturn(zkControllerMock); |
| when(overseerMock.getSolrCloudManager()).thenReturn(cloudDataProviderMock); |
| when(overseerMock.getCoreContainer()).thenReturn(coreContainerMock); |
| when(coreContainerMock.getUpdateShardHandler()).thenReturn(updateShardHandlerMock); |
| when(updateShardHandlerMock.getDefaultHttpClient()).thenReturn(httpClientMock); |
| |
| when(zkControllerMock.getSolrCloudManager()).thenReturn(cloudDataProviderMock); |
| when(cloudDataProviderMock.getClusterStateProvider()).thenReturn(clusterStateProviderMock); |
| when(clusterStateProviderMock.getClusterState()).thenReturn(clusterStateMock); |
| when(clusterStateProviderMock.getLiveNodes()).thenReturn(liveNodes); |
| when(clusterStateProviderMock.getClusterProperties()).thenReturn(Utils.makeMap(DEFAULTS, Utils.makeMap(CLUSTER, Utils.makeMap(USE_LEGACY_REPLICA_ASSIGNMENT, true)))); |
| when(cloudDataProviderMock.getDistribStateManager()).thenReturn(stateManagerMock); |
| when(cloudManagerMock.getDistribStateManager()).thenReturn(distribStateManagerMock); |
| when(distribStateManagerMock.getAutoScalingConfig()).thenReturn(new AutoScalingConfig(Collections.emptyMap())); |
| |
| Mockito.doAnswer( |
| new Answer<Void>() { |
| public Void answer(InvocationOnMock invocation) { |
| System.out.println("set data: " + invocation.getArgument(0) + " " + invocation.getArgument(1)); |
| if (invocation.getArgument(1) == null) { |
| zkClientData.put(invocation.getArgument(0), new byte[0]); |
| } else { |
| zkClientData.put(invocation.getArgument(0), invocation.getArgument(1)); |
| } |
| |
| return null; |
| }}).when(distribStateManagerMock).setData(anyString(), any(), anyInt()); |
| |
| when(distribStateManagerMock.getData(anyString(), any())).thenAnswer(invocation -> { |
| byte[] data = zkClientData.get(invocation.getArgument(0)); |
| if (data == null || data.length == 0) { |
| return null; |
| } |
| return new VersionedData(-1, data, CreateMode.PERSISTENT, ""); |
| |
| }); |
| |
| when(distribStateManagerMock.createData(any(), any(), any())).thenAnswer(invocation -> { |
| System.out.println("set data: " + invocation.getArgument(0) + " " + invocation.getArgument(1)); |
| if (invocation.getArgument(1) == null) { |
| zkClientData.put(invocation.getArgument(0), new byte[0]); |
| } else { |
| zkClientData.put(invocation.getArgument(0), invocation.getArgument(1)); |
| } |
| return null; |
| }); |
| |
| when(distribStateManagerMock.hasData(anyString())) |
| .then(invocation -> zkClientData.containsKey(invocation.getArgument(0)) && zkClientData.get(invocation.getArgument(0)).length > 0); |
| |
| Mockito.doAnswer( |
| new Answer<Void>() { |
| public Void answer(InvocationOnMock invocation) { |
| System.out.println("set data: " + invocation.getArgument(0) + " " + new byte[0]); |
| zkClientData.put(invocation.getArgument(0), new byte[0]); |
| return null; |
| }}).when(distribStateManagerMock).makePath(anyString()); |
| |
| when(solrZkClientMock.exists(any(String.class), isNull(), anyBoolean())).thenAnswer(invocation -> { |
| String key = invocation.getArgument(0); |
| if (zkClientData.containsKey(key)) { |
| return new Stat(); |
| } else { |
| return null; |
| } |
| }); |
| |
| when(cloudManagerMock.getClusterStateProvider()).thenReturn(clusterStateProviderMock); |
| when(cloudManagerMock.getTimeSource()).thenReturn(new TimeSource.NanoTimeSource()); |
| when(cloudManagerMock.getDistribStateManager()).thenReturn(distribStateManagerMock); |
| |
| when(overseerMock.getSolrCloudManager()).thenReturn(cloudManagerMock); |
| |
| when(overseerMock.getStateUpdateQueue(any())).thenReturn(stateUpdateQueueMock); |
| when(overseerMock.getStateUpdateQueue()).thenReturn(stateUpdateQueueMock); |
| |
| Mockito.doAnswer( |
| new Answer<Void>() { |
| public Void answer(InvocationOnMock invocation) { |
| try { |
| handleCreateCollMessage(invocation.getArgument(0)); |
| stateUpdateQueueMock.offer(invocation.getArgument(0)); |
| } catch (KeeperException e) { |
| throw new RuntimeException(e); |
| } catch (InterruptedException e) { |
| throw new RuntimeException(e); |
| } |
| return null; |
| }}).when(overseerMock).offerStateUpdate(any()); |
| |
| when(zkControllerMock.getZkClient()).thenReturn(solrZkClientMock); |
| |
| when(cloudManagerMock.getDistribStateManager()).thenReturn(distribStateManagerMock); |
| when(distribStateManagerMock.getAutoScalingConfig()).thenReturn(new AutoScalingConfig(Collections.emptyMap())); |
| |
| Mockito.doAnswer( |
| new Answer<Void>() { |
| public Void answer(InvocationOnMock invocation) { |
| System.out.println("set data: " + invocation.getArgument(0) + " " + invocation.getArgument(1)); |
| if (invocation.getArgument(1) == null) { |
| zkClientData.put(invocation.getArgument(0), new byte[0]); |
| } else { |
| zkClientData.put(invocation.getArgument(0), invocation.getArgument(1)); |
| } |
| |
| return null; |
| }}).when(distribStateManagerMock).setData(anyString(), any(), anyInt()); |
| |
| when(distribStateManagerMock.getData(anyString(), any())).thenAnswer(invocation -> { |
| byte[] data = zkClientData.get(invocation.getArgument(0)); |
| if (data == null || data.length == 0) { |
| return null; |
| } |
| return new VersionedData(-1, data, CreateMode.PERSISTENT, ""); |
| |
| }); |
| |
| when(distribStateManagerMock.createData(any(), any(), any())).thenAnswer(invocation -> { |
| System.out.println("set data: " + invocation.getArgument(0) + " " + invocation.getArgument(1)); |
| if (invocation.getArgument(1) == null) { |
| zkClientData.put(invocation.getArgument(0), new byte[0]); |
| } else { |
| zkClientData.put(invocation.getArgument(0), invocation.getArgument(1)); |
| } |
| return null; |
| }); |
| |
| when(distribStateManagerMock.hasData(anyString())) |
| .then(invocation -> zkClientData.containsKey(invocation.getArgument(0)) && zkClientData.get(invocation.getArgument(0)).length > 0); |
| |
| Mockito.doAnswer( |
| new Answer<Void>() { |
| public Void answer(InvocationOnMock invocation) { |
| System.out.println("set data: " + invocation.getArgument(0) + " " + new byte[0]); |
| zkClientData.put(invocation.getArgument(0), new byte[0]); |
| return null; |
| }}).when(distribStateManagerMock).makePath(anyString()); |
| |
| zkClientData.put("/configs/myconfig", new byte[1]); |
| |
| when(solrMetricsContextMock.getChildContext(any(Object.class))).thenReturn(solrMetricsContextMock); |
| |
| return liveNodes; |
| } |
| |
| private void handleCreateCollMessage(byte[] bytes) { |
| log.info("track created replicas / collections"); |
| try { |
| ZkNodeProps props = ZkNodeProps.load(bytes); |
| if (CollectionParams.CollectionAction.CREATE.isEqual(props.getStr("operation"))) { |
| String collName = props.getStr("name"); |
| if (collName != null) collectionsSet.put(collName, new ClusterState.CollectionRef( |
| new DocCollection(collName, new HashMap<>(), props.getProperties(), DocRouter.DEFAULT))); |
| } |
| if (CollectionParams.CollectionAction.ADDREPLICA.isEqual(props.getStr("operation"))) { |
| replicas.add(props); |
| } |
| } catch (Exception e) {} |
| } |
| |
| protected void startComponentUnderTest() { |
| thread = new Thread(underTest); |
| thread.start(); |
| } |
| |
| protected void stopComponentUnderTest() throws Exception { |
| if (null != underTest) { |
| underTest.close(); |
| underTest = null; |
| } |
| if (null != thread) { |
| thread.interrupt(); |
| thread.join(); |
| thread = null; |
| } |
| } |
| |
| protected void issueCreateJob(Integer numberOfSlices, |
| Integer replicationFactor, Integer maxShardsPerNode, List<String> createNodeList, boolean sendCreateNodeList, boolean createNodeSetShuffle) { |
| Map<String,Object> propMap = Utils.makeMap( |
| Overseer.QUEUE_OPERATION, CollectionParams.CollectionAction.CREATE.toLower(), |
| ZkStateReader.REPLICATION_FACTOR, replicationFactor.toString(), |
| "name", COLLECTION_NAME, |
| "collection.configName", CONFIG_NAME, |
| OverseerCollectionMessageHandler.NUM_SLICES, numberOfSlices.toString(), |
| ZkStateReader.MAX_SHARDS_PER_NODE, maxShardsPerNode.toString() |
| ); |
| if (sendCreateNodeList) { |
| propMap.put(OverseerCollectionMessageHandler.CREATE_NODE_SET, |
| (createNodeList != null)?StrUtils.join(createNodeList, ','):null); |
| if (OverseerCollectionMessageHandler.CREATE_NODE_SET_SHUFFLE_DEFAULT != createNodeSetShuffle || random().nextBoolean()) { |
| propMap.put(OverseerCollectionMessageHandler.CREATE_NODE_SET_SHUFFLE, createNodeSetShuffle); |
| } |
| } |
| |
| ZkNodeProps props = new ZkNodeProps(propMap); |
| QueueEvent qe = new QueueEvent("id", Utils.toJSON(props), null){ |
| @Override |
| public void setBytes(byte[] bytes) { |
| lastProcessMessageResult = OverseerSolrResponseSerializer.deserialize(bytes); |
| } |
| }; |
| queue.add(qe); |
| } |
| |
| protected void verifySubmitCaptures( |
| Integer numberOfSlices, Integer numberOfReplica, Collection<String> createNodes, boolean dontShuffleCreateNodeSet) { |
| List<String> coreNames = new ArrayList<>(); |
| Map<String,Map<String,Integer>> sliceToNodeUrlsWithoutProtocolPartToNumberOfShardsRunningMapMap = new HashMap<>(); |
| List<String> nodeUrlWithoutProtocolPartForLiveNodes = new ArrayList<>( |
| createNodes.size()); |
| for (String nodeName : createNodes) { |
| String nodeUrlWithoutProtocolPart = nodeName.replaceAll("_", "/"); |
| if (nodeUrlWithoutProtocolPart.startsWith("http://")) nodeUrlWithoutProtocolPart = nodeUrlWithoutProtocolPart |
| .substring(7); |
| nodeUrlWithoutProtocolPartForLiveNodes.add(nodeUrlWithoutProtocolPart); |
| } |
| final Map<String,String> coreName_TO_nodeUrlWithoutProtocolPartForLiveNodes_map = new HashMap<>(); |
| |
| ArgumentCaptor<ShardRequest> shardRequestCaptor = ArgumentCaptor.forClass(ShardRequest.class); |
| ArgumentCaptor<String> nodeUrlsWithoutProtocolPartCaptor = ArgumentCaptor.forClass(String.class); |
| ArgumentCaptor<ModifiableSolrParams> paramsCaptor = ArgumentCaptor.forClass(ModifiableSolrParams.class); |
| verify(shardHandlerMock, times(numberOfReplica * numberOfSlices)) |
| .submit(shardRequestCaptor.capture(), nodeUrlsWithoutProtocolPartCaptor.capture(), paramsCaptor.capture()); |
| |
| for (int i = 0; i < shardRequestCaptor.getAllValues().size(); i++) { |
| ShardRequest shardRequest = shardRequestCaptor.getAllValues().get(i); |
| String nodeUrlsWithoutProtocolPartCapture = nodeUrlsWithoutProtocolPartCaptor.getAllValues().get(i); |
| ModifiableSolrParams params = paramsCaptor.getAllValues().get(i); |
| assertEquals(CoreAdminAction.CREATE.toString(), |
| shardRequest.params.get(CoreAdminParams.ACTION)); |
| // assertEquals(shardRequest.params, submitCapture.params); |
| String coreName = shardRequest.params.get(CoreAdminParams.NAME); |
| assertFalse("Core with name " + coreName + " created twice", |
| coreNames.contains(coreName)); |
| coreNames.add(coreName); |
| assertEquals(CONFIG_NAME, |
| shardRequest.params.get("collection.configName")); |
| assertEquals(COLLECTION_NAME, |
| shardRequest.params.get(CoreAdminParams.COLLECTION)); |
| assertEquals(numberOfSlices.toString(), |
| shardRequest.params.get(ZkStateReader.NUM_SHARDS_PROP)); |
| assertEquals(ADMIN_PATH, shardRequest.params.get("qt")); |
| assertEquals(1, shardRequest.purpose); |
| assertEquals(1, shardRequest.shards.length); |
| assertEquals(nodeUrlsWithoutProtocolPartCapture, |
| shardRequest.shards[0]); |
| assertTrue("Shard " + coreName + " created on wrong node " |
| + shardRequest.shards[0], |
| nodeUrlWithoutProtocolPartForLiveNodes |
| .contains(shardRequest.shards[0])); |
| coreName_TO_nodeUrlWithoutProtocolPartForLiveNodes_map.put(coreName, shardRequest.shards[0]); |
| assertEquals(shardRequest.shards, shardRequest.actualShards); |
| |
| String sliceName = shardRequest.params.get(CoreAdminParams.SHARD); |
| if (!sliceToNodeUrlsWithoutProtocolPartToNumberOfShardsRunningMapMap |
| .containsKey(sliceName)) { |
| sliceToNodeUrlsWithoutProtocolPartToNumberOfShardsRunningMapMap.put( |
| sliceName, new HashMap<String,Integer>()); |
| } |
| Map<String,Integer> nodeUrlsWithoutProtocolPartToNumberOfShardsRunningMap = sliceToNodeUrlsWithoutProtocolPartToNumberOfShardsRunningMapMap |
| .get(sliceName); |
| Integer existingCount; |
| nodeUrlsWithoutProtocolPartToNumberOfShardsRunningMap |
| .put( |
| shardRequest.shards[0], |
| ((existingCount = nodeUrlsWithoutProtocolPartToNumberOfShardsRunningMap |
| .get(shardRequest.shards[0])) == null) ? 1 |
| : (existingCount + 1)); |
| } |
| |
| assertEquals(numberOfSlices * numberOfReplica, coreNames.size()); |
| for (int i = 1; i <= numberOfSlices; i++) { |
| for (int j = 1; j <= numberOfReplica; j++) { |
| String coreName = coreNames.get((i-1) * numberOfReplica + (j-1)); |
| |
| if (dontShuffleCreateNodeSet) { |
| final String expectedNodeName = nodeUrlWithoutProtocolPartForLiveNodes.get((numberOfReplica * (i - 1) + (j - 1)) % nodeUrlWithoutProtocolPartForLiveNodes.size()); |
| assertFalse("expectedNodeName is null for coreName="+coreName, null == expectedNodeName); |
| |
| final String actualNodeName = coreName_TO_nodeUrlWithoutProtocolPartForLiveNodes_map.get(coreName); |
| assertFalse("actualNodeName is null for coreName="+coreName, null == actualNodeName); |
| |
| assertTrue("node name mismatch for coreName="+coreName+" ( actual="+actualNodeName+" versus expected="+expectedNodeName+" )", actualNodeName.equals(expectedNodeName)); |
| } |
| } |
| } |
| |
| assertEquals(numberOfSlices.intValue(), |
| sliceToNodeUrlsWithoutProtocolPartToNumberOfShardsRunningMapMap.size()); |
| for (int i = 1; i <= numberOfSlices; i++) { |
| sliceToNodeUrlsWithoutProtocolPartToNumberOfShardsRunningMapMap.keySet() |
| .contains("shard" + i); |
| } |
| int minShardsPerSlicePerNode = numberOfReplica / createNodes.size(); |
| int numberOfNodesSupposedToRunMaxShards = numberOfReplica |
| % createNodes.size(); |
| int numberOfNodesSupposedToRunMinShards = createNodes.size() |
| - numberOfNodesSupposedToRunMaxShards; |
| int maxShardsPerSlicePerNode = (minShardsPerSlicePerNode + 1); |
| if (numberOfNodesSupposedToRunMaxShards == 0) { |
| numberOfNodesSupposedToRunMaxShards = numberOfNodesSupposedToRunMinShards; |
| maxShardsPerSlicePerNode = minShardsPerSlicePerNode; |
| } |
| boolean diffBetweenMinAndMaxShardsPerSlicePerNode = (maxShardsPerSlicePerNode != minShardsPerSlicePerNode); |
| |
| for (Entry<String,Map<String,Integer>> sliceToNodeUrlsWithoutProtocolPartToNumberOfShardsRunningMapMapEntry : sliceToNodeUrlsWithoutProtocolPartToNumberOfShardsRunningMapMap |
| .entrySet()) { |
| int numberOfShardsRunning = 0; |
| int numberOfNodesRunningMinShards = 0; |
| int numberOfNodesRunningMaxShards = 0; |
| int numberOfNodesRunningAtLeastOneShard = 0; |
| for (String nodeUrlsWithoutProtocolPart : sliceToNodeUrlsWithoutProtocolPartToNumberOfShardsRunningMapMapEntry |
| .getValue().keySet()) { |
| int numberOfShardsRunningOnThisNode = sliceToNodeUrlsWithoutProtocolPartToNumberOfShardsRunningMapMapEntry |
| .getValue().get(nodeUrlsWithoutProtocolPart); |
| numberOfShardsRunning += numberOfShardsRunningOnThisNode; |
| numberOfNodesRunningAtLeastOneShard++; |
| assertTrue( |
| "Node " |
| + nodeUrlsWithoutProtocolPart |
| + " is running wrong number of shards. Supposed to run " |
| + minShardsPerSlicePerNode |
| + (diffBetweenMinAndMaxShardsPerSlicePerNode ? (" or " + maxShardsPerSlicePerNode) |
| : ""), |
| (numberOfShardsRunningOnThisNode == minShardsPerSlicePerNode) |
| || (numberOfShardsRunningOnThisNode == maxShardsPerSlicePerNode)); |
| if (numberOfShardsRunningOnThisNode == minShardsPerSlicePerNode) numberOfNodesRunningMinShards++; |
| if (numberOfShardsRunningOnThisNode == maxShardsPerSlicePerNode) numberOfNodesRunningMaxShards++; |
| } |
| if (minShardsPerSlicePerNode == 0) numberOfNodesRunningMinShards = (createNodes |
| .size() - numberOfNodesRunningAtLeastOneShard); |
| assertEquals( |
| "Too many shards are running under slice " |
| + sliceToNodeUrlsWithoutProtocolPartToNumberOfShardsRunningMapMapEntry |
| .getKey(), |
| numberOfReplica.intValue(), numberOfShardsRunning); |
| assertEquals(numberOfNodesSupposedToRunMinShards, |
| numberOfNodesRunningMinShards); |
| assertEquals(numberOfNodesSupposedToRunMaxShards, |
| numberOfNodesRunningMaxShards); |
| } |
| } |
| |
| protected void waitForEmptyQueue(long maxWait) throws Exception { |
| final TimeOut timeout = new TimeOut(maxWait, TimeUnit.MILLISECONDS, TimeSource.NANO_TIME); |
| while (queue.peek() != null) { |
| if (timeout.hasTimedOut()) |
| fail("Queue not empty within " + maxWait + " ms"); |
| Thread.sleep(100); |
| } |
| } |
| |
| protected enum CreateNodeListOptions { |
| SEND, |
| DONT_SEND, |
| SEND_NULL |
| } |
| protected void testTemplate(Integer numberOfNodes, Integer numberOfNodesToCreateOn, CreateNodeListOptions createNodeListOption, Integer replicationFactor, |
| Integer numberOfSlices, Integer maxShardsPerNode, |
| boolean collectionExceptedToBeCreated) throws Exception { |
| assertTrue("Wrong usage of testTemplate. numberOfNodesToCreateOn " + numberOfNodesToCreateOn + " is not allowed to be higher than numberOfNodes " + numberOfNodes, numberOfNodes.intValue() >= numberOfNodesToCreateOn.intValue()); |
| assertTrue("Wrong usage of testTemplage. createNodeListOption has to be " + CreateNodeListOptions.SEND + " when numberOfNodes and numberOfNodesToCreateOn are unequal", ((createNodeListOption == CreateNodeListOptions.SEND) || (numberOfNodes.intValue() == numberOfNodesToCreateOn.intValue()))); |
| |
| Set<String> liveNodes = commonMocks(numberOfNodes); |
| List<String> createNodeList = new ArrayList<>(); |
| int i = 0; |
| for (String node : liveNodes) { |
| if (i++ < numberOfNodesToCreateOn) { |
| createNodeList.add(node); |
| } |
| } |
| |
| if (random().nextBoolean()) Collections.shuffle(createNodeList, random()); |
| |
| underTest = new OverseerCollectionConfigSetProcessorToBeTested(zkStateReaderMock, |
| "1234", shardHandlerFactoryMock, ADMIN_PATH, workQueueMock, runningMapMock, |
| overseerMock, completedMapMock, failureMapMock, solrMetricsContextMock); |
| |
| |
| if (log.isInfoEnabled()) { |
| log.info("clusterstate {}", clusterStateMock.hashCode()); |
| } |
| |
| startComponentUnderTest(); |
| |
| final List<String> createNodeListToSend = ((createNodeListOption != CreateNodeListOptions.SEND_NULL) ? createNodeList : null); |
| final boolean sendCreateNodeList = (createNodeListOption != CreateNodeListOptions.DONT_SEND); |
| final boolean dontShuffleCreateNodeSet = (createNodeListToSend != null) && sendCreateNodeList && random().nextBoolean(); |
| issueCreateJob(numberOfSlices, replicationFactor, maxShardsPerNode, createNodeListToSend, sendCreateNodeList, !dontShuffleCreateNodeSet); |
| waitForEmptyQueue(10000); |
| |
| if (collectionExceptedToBeCreated) { |
| assertNotNull(lastProcessMessageResult.getResponse().toString(), lastProcessMessageResult); |
| } |
| |
| if (collectionExceptedToBeCreated) { |
| verifySubmitCaptures(numberOfSlices, replicationFactor, |
| createNodeList, dontShuffleCreateNodeSet); |
| } |
| } |
| @Test |
| public void testNoReplicationEqualNumberOfSlicesPerNode() throws Exception { |
| Integer numberOfNodes = 4; |
| Integer numberOfNodesToCreateOn = 4; |
| CreateNodeListOptions createNodeListOptions = CreateNodeListOptions.DONT_SEND; |
| Integer replicationFactor = 1; |
| Integer numberOfSlices = 8; |
| Integer maxShardsPerNode = 2; |
| testTemplate(numberOfNodes, numberOfNodesToCreateOn, createNodeListOptions, replicationFactor, numberOfSlices, |
| maxShardsPerNode, true); |
| } |
| |
| @Test |
| public void testReplicationEqualNumberOfSlicesPerNode() throws Exception { |
| Integer numberOfNodes = 4; |
| Integer numberOfNodesToCreateOn = 4; |
| CreateNodeListOptions createNodeListOptions = CreateNodeListOptions.DONT_SEND; |
| Integer replicationFactor = 2; |
| Integer numberOfSlices = 4; |
| Integer maxShardsPerNode = 2; |
| testTemplate(numberOfNodes, numberOfNodesToCreateOn, createNodeListOptions, replicationFactor, numberOfSlices, |
| maxShardsPerNode, true); |
| } |
| |
| @Test |
| public void testNoReplicationEqualNumberOfSlicesPerNodeSendCreateNodesEqualToLiveNodes() throws Exception { |
| Integer numberOfNodes = 4; |
| Integer numberOfNodesToCreateOn = 4; |
| CreateNodeListOptions createNodeListOptions = CreateNodeListOptions.SEND; |
| Integer replicationFactor = 1; |
| Integer numberOfSlices = 8; |
| Integer maxShardsPerNode = 2; |
| testTemplate(numberOfNodes, numberOfNodesToCreateOn, createNodeListOptions, replicationFactor, numberOfSlices, |
| maxShardsPerNode, true); |
| } |
| |
| @Test |
| public void testReplicationEqualNumberOfSlicesPerNodeSendCreateNodesEqualToLiveNodes() throws Exception { |
| Integer numberOfNodes = 4; |
| Integer numberOfNodesToCreateOn = 4; |
| CreateNodeListOptions createNodeListOptions = CreateNodeListOptions.SEND; |
| Integer replicationFactor = 2; |
| Integer numberOfSlices = 4; |
| Integer maxShardsPerNode = 2; |
| testTemplate(numberOfNodes, numberOfNodesToCreateOn, createNodeListOptions, replicationFactor, numberOfSlices, |
| maxShardsPerNode, true); |
| } |
| |
| @Test |
| public void testNoReplicationEqualNumberOfSlicesPerNodeSendNullCreateNodes() throws Exception { |
| Integer numberOfNodes = 4; |
| Integer numberOfNodesToCreateOn = 4; |
| CreateNodeListOptions createNodeListOptions = CreateNodeListOptions.SEND_NULL; |
| Integer replicationFactor = 1; |
| Integer numberOfSlices = 8; |
| Integer maxShardsPerNode = 2; |
| testTemplate(numberOfNodes, numberOfNodesToCreateOn, createNodeListOptions, replicationFactor, numberOfSlices, |
| maxShardsPerNode, true); |
| } |
| |
| @Test |
| public void testReplicationEqualNumberOfSlicesPerNodeSendNullCreateNodes() throws Exception { |
| Integer numberOfNodes = 4; |
| Integer numberOfNodesToCreateOn = 4; |
| CreateNodeListOptions createNodeListOptions = CreateNodeListOptions.SEND_NULL; |
| Integer replicationFactor = 2; |
| Integer numberOfSlices = 4; |
| Integer maxShardsPerNode = 2; |
| testTemplate(numberOfNodes, numberOfNodesToCreateOn, createNodeListOptions, replicationFactor, numberOfSlices, |
| maxShardsPerNode, true); |
| } |
| |
| @Test |
| public void testNoReplicationUnequalNumberOfSlicesPerNode() throws Exception { |
| Integer numberOfNodes = 4; |
| Integer numberOfNodesToCreateOn = 4; |
| CreateNodeListOptions createNodeListOptions = CreateNodeListOptions.DONT_SEND; |
| Integer replicationFactor = 1; |
| Integer numberOfSlices = 6; |
| Integer maxShardsPerNode = 2; |
| testTemplate(numberOfNodes, numberOfNodesToCreateOn, createNodeListOptions, replicationFactor, numberOfSlices, |
| maxShardsPerNode, true); |
| } |
| |
| @Test |
| public void testReplicationUnequalNumberOfSlicesPerNode() throws Exception { |
| Integer numberOfNodes = 4; |
| Integer numberOfNodesToCreateOn = 4; |
| CreateNodeListOptions createNodeListOptions = CreateNodeListOptions.DONT_SEND; |
| Integer replicationFactor = 2; |
| Integer numberOfSlices = 3; |
| Integer maxShardsPerNode = 2; |
| testTemplate(numberOfNodes, numberOfNodesToCreateOn, createNodeListOptions, replicationFactor, numberOfSlices, |
| maxShardsPerNode, true); |
| } |
| |
| @Test |
| public void testNoReplicationCollectionNotCreatedDueToMaxShardsPerNodeLimit() |
| throws Exception { |
| Integer numberOfNodes = 4; |
| Integer numberOfNodesToCreateOn = 4; |
| CreateNodeListOptions createNodeListOptions = CreateNodeListOptions.DONT_SEND; |
| Integer replicationFactor = 1; |
| Integer numberOfSlices = 6; |
| Integer maxShardsPerNode = 1; |
| testTemplate(numberOfNodes, numberOfNodesToCreateOn, createNodeListOptions, replicationFactor, numberOfSlices, |
| maxShardsPerNode, false); |
| } |
| |
| @Test |
| public void testReplicationCollectionNotCreatedDueToMaxShardsPerNodeLimit() |
| throws Exception { |
| Integer numberOfNodes = 4; |
| Integer numberOfNodesToCreateOn = 4; |
| CreateNodeListOptions createNodeListOptions = CreateNodeListOptions.DONT_SEND; |
| Integer replicationFactor = 2; |
| Integer numberOfSlices = 3; |
| Integer maxShardsPerNode = 1; |
| testTemplate(numberOfNodes, numberOfNodesToCreateOn, createNodeListOptions, replicationFactor, numberOfSlices, |
| maxShardsPerNode, false); |
| } |
| |
| @Test |
| public void testNoReplicationLimitedNodesToCreateOn() |
| throws Exception { |
| Integer numberOfNodes = 4; |
| Integer numberOfNodesToCreateOn = 2; |
| CreateNodeListOptions createNodeListOptions = CreateNodeListOptions.SEND; |
| Integer replicationFactor = 1; |
| Integer numberOfSlices = 6; |
| Integer maxShardsPerNode = 3; |
| testTemplate(numberOfNodes, numberOfNodesToCreateOn, createNodeListOptions, replicationFactor, numberOfSlices, |
| maxShardsPerNode, true); |
| } |
| |
| @Test |
| public void testReplicationLimitedNodesToCreateOn() |
| throws Exception { |
| Integer numberOfNodes = 4; |
| Integer numberOfNodesToCreateOn = 2; |
| CreateNodeListOptions createNodeListOptions = CreateNodeListOptions.SEND; |
| Integer replicationFactor = 2; |
| Integer numberOfSlices = 3; |
| Integer maxShardsPerNode = 3; |
| testTemplate(numberOfNodes, numberOfNodesToCreateOn, createNodeListOptions, replicationFactor, numberOfSlices, |
| maxShardsPerNode, true); |
| } |
| |
| @Test |
| public void testNoReplicationCollectionNotCreatedDueToMaxShardsPerNodeAndNodesToCreateOnLimits() |
| throws Exception { |
| Integer numberOfNodes = 4; |
| Integer numberOfNodesToCreateOn = 3; |
| CreateNodeListOptions createNodeListOptions = CreateNodeListOptions.SEND; |
| Integer replicationFactor = 1; |
| Integer numberOfSlices = 8; |
| Integer maxShardsPerNode = 2; |
| testTemplate(numberOfNodes, numberOfNodesToCreateOn, createNodeListOptions, replicationFactor, numberOfSlices, |
| maxShardsPerNode, false); |
| } |
| |
| @Test |
| public void testReplicationCollectionNotCreatedDueToMaxShardsPerNodeAndNodesToCreateOnLimits() |
| throws Exception { |
| Integer numberOfNodes = 4; |
| Integer numberOfNodesToCreateOn = 3; |
| CreateNodeListOptions createNodeListOptions = CreateNodeListOptions.SEND; |
| Integer replicationFactor = 2; |
| Integer numberOfSlices = 4; |
| Integer maxShardsPerNode = 2; |
| testTemplate(numberOfNodes, numberOfNodesToCreateOn, createNodeListOptions, replicationFactor, numberOfSlices, |
| maxShardsPerNode, false); |
| } |
| |
| } |
