You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by cp...@apache.org on 2017/05/25 17:39:38 UTC
[2/6] lucene-solr:jira/solr-8668: SOLR-10233: Add support for replica
types
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/2fc41d56/solr/core/src/test/org/apache/solr/cloud/TestTlogReplica.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestTlogReplica.java b/solr/core/src/test/org/apache/solr/cloud/TestTlogReplica.java
new file mode 100644
index 0000000..034a8bf
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/TestTlogReplica.java
@@ -0,0 +1,845 @@
+/*
+ * 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.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import org.apache.http.client.HttpClient;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.util.LuceneTestCase.Slow;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrQuery;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.embedded.JettySolrRunner;
+import org.apache.solr.client.solrj.impl.CloudSolrClient;
+import org.apache.solr.client.solrj.impl.HttpSolrClient;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.client.solrj.response.CollectionAdminResponse;
+import org.apache.solr.client.solrj.response.QueryResponse;
+import org.apache.solr.common.SolrDocument;
+import org.apache.solr.common.SolrDocumentList;
+import org.apache.solr.common.SolrException;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.cloud.CollectionStatePredicate;
+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.ZkStateReader;
+import org.apache.solr.common.util.NamedList;
+import org.apache.solr.core.SolrCore;
+import org.apache.solr.update.DirectUpdateHandler2;
+import org.apache.solr.update.SolrIndexWriter;
+import org.apache.solr.update.UpdateHandler;
+import org.apache.solr.update.UpdateLog;
+import org.apache.solr.util.RefCounted;
+import org.apache.solr.util.TestInjection;
+import org.apache.solr.util.TimeOut;
+import org.apache.zookeeper.KeeperException;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.carrotsearch.randomizedtesting.annotations.Repeat;
+
+@Slow
+public class TestTlogReplica extends SolrCloudTestCase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private String collectionName = null;
+ private final static int REPLICATION_TIMEOUT_SECS = 10;
+
+ private String suggestedCollectionName() {
+ return (getTestClass().getSimpleName().replace("Test", "") + "_" + getTestName().split(" ")[0]).replaceAll("(.)(\\p{Upper})", "$1_$2").toLowerCase(Locale.ROOT);
+ }
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ TestInjection.waitForReplicasInSync = null; // We'll be explicit about this in this test
+ configureCluster(2) // 2 + random().nextInt(3)
+ .addConfig("conf", configset("cloud-minimal-inplace-updates"))
+ .configure();
+ Boolean useLegacyCloud = rarely();
+ LOG.info("Using legacyCloud?: {}", useLegacyCloud);
+ CollectionAdminRequest.ClusterProp clusterPropRequest = CollectionAdminRequest.setClusterProperty(ZkStateReader.LEGACY_CLOUD, String.valueOf(useLegacyCloud));
+ CollectionAdminResponse response = clusterPropRequest.process(cluster.getSolrClient());
+ assertEquals(0, response.getStatus());
+ }
+
+ @AfterClass
+ public static void tearDownCluster() {
+ TestInjection.reset();
+ }
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ collectionName = suggestedCollectionName();
+ expectThrows(SolrException.class, () -> getCollectionState(collectionName));
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ for (JettySolrRunner jetty:cluster.getJettySolrRunners()) {
+ if (!jetty.isRunning()) {
+ LOG.warn("Jetty {} not running, probably some bad test. Starting it", jetty.getLocalPort());
+ ChaosMonkey.start(jetty);
+ }
+ }
+ if (cluster.getSolrClient().getZkStateReader().getClusterState().getCollectionOrNull(collectionName) != null) {
+ LOG.info("tearDown deleting collection");
+ CollectionAdminRequest.deleteCollection(collectionName).process(cluster.getSolrClient());
+ waitForDeletion(collectionName);
+ }
+ super.tearDown();
+ }
+
+ /**
+ * Asserts that Update logs exist for replicas of type {@link org.apache.solr.common.cloud.Replica.Type#NRT}, but not
+ * for replicas of type {@link org.apache.solr.common.cloud.Replica.Type#PULL}
+ */
+ private void assertUlogPresence(DocCollection collection) {
+ for (Slice s:collection.getSlices()) {
+ for (Replica r:s.getReplicas()) {
+ SolrCore core = null;
+ try {
+ core = cluster.getReplicaJetty(r).getCoreContainer().getCore(r.getCoreName());
+ assertNotNull(core);
+ assertTrue("Update log should exist for replicas of type Append",
+ new java.io.File(core.getUlogDir()).exists());
+ } finally {
+ core.close();
+ }
+ }
+ }
+ }
+
+ @Repeat(iterations=2) // 2 times to make sure cleanup is complete and we can create the same collection
+ public void testCreateDelete() throws Exception {
+ try {
+ CollectionAdminRequest.createCollection(collectionName, "conf", 2, 0, 4, 0)
+ .setMaxShardsPerNode(100)
+ .process(cluster.getSolrClient());
+ boolean reloaded = false;
+ while (true) {
+ DocCollection docCollection = getCollectionState(collectionName);
+ assertNotNull(docCollection);
+ assertEquals("Expecting 2 shards",
+ 2, docCollection.getSlices().size());
+ assertEquals("Expecting 4 relpicas per shard",
+ 8, docCollection.getReplicas().size());
+ assertEquals("Expecting 8 tlog replicas, 4 per shard",
+ 8, docCollection.getReplicas(EnumSet.of(Replica.Type.TLOG)).size());
+ assertEquals("Expecting no nrt replicas",
+ 0, docCollection.getReplicas(EnumSet.of(Replica.Type.NRT)).size());
+ assertEquals("Expecting no pull replicas",
+ 0, docCollection.getReplicas(EnumSet.of(Replica.Type.PULL)).size());
+ for (Slice s:docCollection.getSlices()) {
+ assertTrue(s.getLeader().getType() == Replica.Type.TLOG);
+ List<String> shardElectionNodes = cluster.getZkClient().getChildren(ZkStateReader.getShardLeadersElectPath(collectionName, s.getName()), null, true);
+ assertEquals("Unexpected election nodes for Shard: " + s.getName() + ": " + Arrays.toString(shardElectionNodes.toArray()),
+ 4, shardElectionNodes.size());
+ }
+ assertUlogPresence(docCollection);
+ if (reloaded) {
+ break;
+ } else {
+ // reload
+ CollectionAdminResponse response = CollectionAdminRequest.reloadCollection(collectionName)
+ .process(cluster.getSolrClient());
+ assertEquals(0, response.getStatus());
+ reloaded = true;
+ }
+ }
+ } finally {
+ zkClient().printLayoutToStdOut();
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void testAddDocs() throws Exception {
+ int numTlogReplicas = 1 + random().nextInt(3);
+ DocCollection docCollection = createAndWaitForCollection(1, 0, numTlogReplicas, 0);
+ assertEquals(1, docCollection.getSlices().size());
+
+ cluster.getSolrClient().add(collectionName, new SolrInputDocument("id", "1", "foo", "bar"));
+ cluster.getSolrClient().commit(collectionName);
+
+ Slice s = docCollection.getSlices().iterator().next();
+ try (HttpSolrClient leaderClient = getHttpSolrClient(s.getLeader().getCoreUrl())) {
+ assertEquals(1, leaderClient.query(new SolrQuery("*:*")).getResults().getNumFound());
+ }
+
+ TimeOut t = new TimeOut(REPLICATION_TIMEOUT_SECS, TimeUnit.SECONDS);
+ for (Replica r:s.getReplicas(EnumSet.of(Replica.Type.TLOG))) {
+ //TODO: assert replication < REPLICATION_TIMEOUT_SECS
+ try (HttpSolrClient tlogReplicaClient = getHttpSolrClient(r.getCoreUrl())) {
+ while (true) {
+ try {
+ assertEquals("Replica " + r.getName() + " not up to date after 10 seconds",
+ 1, tlogReplicaClient.query(new SolrQuery("*:*")).getResults().getNumFound());
+ // Append replicas process all updates
+ SolrQuery req = new SolrQuery(
+ "qt", "/admin/plugins",
+ "stats", "true");
+ QueryResponse statsResponse = tlogReplicaClient.query(req);
+ assertEquals("Append replicas should recive all updates. Replica: " + r + ", response: " + statsResponse,
+ 1L, ((Map<String, Object>)((NamedList<Object>)statsResponse.getResponse()).findRecursive("plugins", "UPDATE", "updateHandler", "stats")).get("UPDATE.updateHandler.cumulativeAdds.count"));
+ break;
+ } catch (AssertionError e) {
+ if (t.hasTimedOut()) {
+ throw e;
+ } else {
+ Thread.sleep(100);
+ }
+ }
+ }
+ }
+ }
+ assertUlogPresence(docCollection);
+ }
+
+ public void testAddRemoveTlogReplica() throws Exception {
+ DocCollection docCollection = createAndWaitForCollection(2, 0, 1, 0);
+ assertEquals(2, docCollection.getSlices().size());
+
+ CollectionAdminRequest.addReplicaToShard(collectionName, "shard1", Replica.Type.TLOG).process(cluster.getSolrClient());
+ docCollection = assertNumberOfReplicas(0, 3, 0, true, false);
+ CollectionAdminRequest.addReplicaToShard(collectionName, "shard2", Replica.Type.TLOG).process(cluster.getSolrClient());
+ docCollection = assertNumberOfReplicas(0, 4, 0, true, false);
+
+ waitForState("Expecting collection to have 2 shards and 2 replica each", collectionName, clusterShape(2, 2));
+
+ //Delete tlog replica from shard1
+ CollectionAdminRequest.deleteReplica(
+ collectionName,
+ "shard1",
+ docCollection.getSlice("shard1").getReplicas(EnumSet.of(Replica.Type.TLOG)).get(0).getName())
+ .process(cluster.getSolrClient());
+ assertNumberOfReplicas(0, 3, 0, true, true);
+ }
+
+ public void testRemoveLeader() throws Exception {
+ doReplaceLeader(true);
+ }
+
+ public void testKillLeader() throws Exception {
+ doReplaceLeader(false);
+ }
+
+ public void testRealTimeGet() throws SolrServerException, IOException, KeeperException, InterruptedException {
+ // should be redirected to Replica.Type.REALTIME
+ int numReplicas = random().nextBoolean()?1:2;
+ int numNrtReplicas = random().nextBoolean()?0:2;
+ CollectionAdminRequest.createCollection(collectionName, "conf", 1, numNrtReplicas, numReplicas, 0)
+ .setMaxShardsPerNode(100)
+ .process(cluster.getSolrClient());
+ waitForState("Unexpected replica count", collectionName, activeReplicaCount(numNrtReplicas, numReplicas, 0));
+ DocCollection docCollection = assertNumberOfReplicas(numNrtReplicas, numReplicas, 0, false, true);
+ HttpClient httpClient = cluster.getSolrClient().getHttpClient();
+ int id = 0;
+ Slice slice = docCollection.getSlice("shard1");
+ List<String> ids = new ArrayList<>(slice.getReplicas().size());
+ for (Replica rAdd:slice.getReplicas()) {
+ try (HttpSolrClient client = getHttpSolrClient(rAdd.getCoreUrl(), httpClient)) {
+ client.add(new SolrInputDocument("id", String.valueOf(id), "foo_s", "bar"));
+ }
+ SolrDocument docCloudClient = cluster.getSolrClient().getById(collectionName, String.valueOf(id));
+ assertEquals("bar", docCloudClient.getFieldValue("foo_s"));
+ for (Replica rGet:slice.getReplicas()) {
+ try (HttpSolrClient client = getHttpSolrClient(rGet.getCoreUrl(), httpClient)) {
+ SolrDocument doc = client.getById(String.valueOf(id));
+ assertEquals("bar", doc.getFieldValue("foo_s"));
+ }
+ }
+ ids.add(String.valueOf(id));
+ id++;
+ }
+ SolrDocumentList previousAllIdsResult = null;
+ for (Replica rAdd:slice.getReplicas()) {
+ try (HttpSolrClient client = getHttpSolrClient(rAdd.getCoreUrl(), httpClient)) {
+ SolrDocumentList allIdsResult = client.getById(ids);
+ if (previousAllIdsResult != null) {
+ assertTrue(compareSolrDocumentList(previousAllIdsResult, allIdsResult));
+ } else {
+ // set the first response here
+ previousAllIdsResult = allIdsResult;
+ assertEquals("Unexpected number of documents", ids.size(), allIdsResult.getNumFound());
+ }
+ }
+ id++;
+ }
+ }
+
+ /*
+ * validate leader election and that replication still happens on a new leader
+ */
+ private void doReplaceLeader(boolean removeReplica) throws Exception {
+ DocCollection docCollection = createAndWaitForCollection(1, 0, 2, 0);
+
+ // Add a document and commit
+ cluster.getSolrClient().add(collectionName, new SolrInputDocument("id", "1", "foo", "bar"));
+ cluster.getSolrClient().commit(collectionName);
+ Slice s = docCollection.getSlices().iterator().next();
+ try (HttpSolrClient leaderClient = getHttpSolrClient(s.getLeader().getCoreUrl())) {
+ assertEquals(1, leaderClient.query(new SolrQuery("*:*")).getResults().getNumFound());
+ }
+
+ waitForNumDocsInAllReplicas(1, docCollection.getReplicas(EnumSet.of(Replica.Type.TLOG)), REPLICATION_TIMEOUT_SECS);
+
+ // Delete leader replica from shard1
+ JettySolrRunner leaderJetty = null;
+ if (removeReplica) {
+ CollectionAdminRequest.deleteReplica(
+ collectionName,
+ "shard1",
+ s.getLeader().getName())
+ .process(cluster.getSolrClient());
+ } else {
+ leaderJetty = cluster.getReplicaJetty(s.getLeader());
+ ChaosMonkey.kill(leaderJetty);
+ waitForState("Leader replica not removed", collectionName, clusterShape(1, 1));
+ // Wait for cluster state to be updated
+ waitForState("Replica state not updated in cluster state",
+ collectionName, clusterStateReflectsActiveAndDownReplicas());
+ }
+ docCollection = assertNumberOfReplicas(0, 1, 0, true, true);
+
+ // Wait until a new leader is elected
+ TimeOut t = new TimeOut(30, TimeUnit.SECONDS);
+ while (!t.hasTimedOut()) {
+ docCollection = getCollectionState(collectionName);
+ Replica leader = docCollection.getSlice("shard1").getLeader();
+ if (leader != null && leader.isActive(cluster.getSolrClient().getZkStateReader().getClusterState().getLiveNodes())) {
+ break;
+ }
+ Thread.sleep(500);
+ }
+ assertFalse("Timeout waiting for a new leader to be elected", t.hasTimedOut());
+
+ // There is a new leader, I should be able to add and commit
+ cluster.getSolrClient().add(collectionName, new SolrInputDocument("id", "2", "foo", "zoo"));
+ cluster.getSolrClient().commit(collectionName);
+
+ // Queries should still work
+ waitForNumDocsInAllReplicas(2, docCollection.getReplicas(EnumSet.of(Replica.Type.TLOG)), REPLICATION_TIMEOUT_SECS);
+ // Start back the node
+ if (removeReplica) {
+ CollectionAdminRequest.addReplicaToShard(collectionName, "shard1", Replica.Type.TLOG).process(cluster.getSolrClient());
+ } else {
+ ChaosMonkey.start(leaderJetty);
+ }
+ waitForState("Expected collection to be 1x2", collectionName, clusterShape(1, 2));
+ // added replica should replicate from the leader
+ waitForNumDocsInAllReplicas(2, docCollection.getReplicas(EnumSet.of(Replica.Type.TLOG)), REPLICATION_TIMEOUT_SECS);
+ }
+
+ public void testKillTlogReplica() throws Exception {
+ DocCollection docCollection = createAndWaitForCollection(1, 0, 2, 0);
+
+ waitForNumDocsInAllActiveReplicas(0);
+ cluster.getSolrClient().add(collectionName, new SolrInputDocument("id", "1", "foo", "bar"));
+ cluster.getSolrClient().commit(collectionName);
+ waitForNumDocsInAllActiveReplicas(1);
+
+ JettySolrRunner pullReplicaJetty = cluster.getReplicaJetty(docCollection.getSlice("shard1").getReplicas(EnumSet.of(Replica.Type.TLOG)).get(0));
+ ChaosMonkey.kill(pullReplicaJetty);
+ waitForState("Replica not removed", collectionName, activeReplicaCount(0, 1, 0));
+// // Also wait for the replica to be placed in state="down"
+// waitForState("Didn't update state", collectionName, clusterStateReflectsActiveAndDownReplicas());
+
+ cluster.getSolrClient().add(collectionName, new SolrInputDocument("id", "2", "foo", "bar"));
+ cluster.getSolrClient().commit(collectionName);
+ waitForNumDocsInAllActiveReplicas(2);
+
+ ChaosMonkey.start(pullReplicaJetty);
+ waitForState("Replica not added", collectionName, activeReplicaCount(0, 2, 0));
+ waitForNumDocsInAllActiveReplicas(2);
+ }
+
+ public void testOnlyLeaderIndexes() throws Exception {
+ createAndWaitForCollection(1, 0, 2, 0);
+
+ CloudSolrClient cloudClient = cluster.getSolrClient();
+ new UpdateRequest()
+ .add(sdoc("id", "1"))
+ .add(sdoc("id", "2"))
+ .add(sdoc("id", "3"))
+ .add(sdoc("id", "4"))
+ .process(cloudClient, collectionName);
+
+ {
+ UpdateHandler updateHandler = getSolrCore(true).get(0).getUpdateHandler();
+ RefCounted<IndexWriter> iwRef = updateHandler.getSolrCoreState().getIndexWriter(null);
+ assertTrue("IndexWriter at leader must see updates ", iwRef.get().hasUncommittedChanges());
+ iwRef.decref();
+ }
+
+ for (SolrCore solrCore : getSolrCore(false)) {
+ RefCounted<IndexWriter> iwRef = solrCore.getUpdateHandler().getSolrCoreState().getIndexWriter(null);
+ assertFalse("IndexWriter at replicas must not see updates ", iwRef.get().hasUncommittedChanges());
+ iwRef.decref();
+ }
+
+ checkRTG(1, 4, cluster.getJettySolrRunners());
+
+ new UpdateRequest()
+ .deleteById("1")
+ .deleteByQuery("id:2")
+ .process(cloudClient, collectionName);
+
+ // The DBQ is not processed at replicas, so we still can get doc2 and other docs by RTG
+ checkRTG(2,4, getSolrRunner(false));
+
+ new UpdateRequest()
+ .commit(cloudClient, collectionName);
+
+ waitForNumDocsInAllActiveReplicas(2);
+
+ // Update log roll over
+ for (SolrCore solrCore : getSolrCore(false)) {
+ UpdateLog updateLog = solrCore.getUpdateHandler().getUpdateLog();
+ assertFalse(updateLog.hasUncommittedChanges());
+ }
+
+ // UpdateLog copy over old updates
+ for (int i = 15; i <= 150; i++) {
+ cloudClient.add(collectionName, sdoc("id",String.valueOf(i)));
+ if (random().nextInt(100) < 15 & i != 150) {
+ cloudClient.commit(collectionName);
+ }
+ }
+ checkRTG(120,150, cluster.getJettySolrRunners());
+ waitForReplicasCatchUp(20);
+ }
+
+ @SuppressWarnings("unchecked")
+ public void testRecovery() throws Exception {
+ boolean useKill = random().nextBoolean();
+ createAndWaitForCollection(1, 0, 2, 0);
+
+ CloudSolrClient cloudClient = cluster.getSolrClient();
+ new UpdateRequest()
+ .add(sdoc("id", "3"))
+ .add(sdoc("id", "4"))
+ .commit(cloudClient, collectionName);
+ new UpdateRequest()
+ .add(sdoc("id", "5"))
+ .process(cloudClient, collectionName);
+ JettySolrRunner solrRunner = getSolrRunner(false).get(0);
+ if (useKill) {
+ ChaosMonkey.kill(solrRunner);
+ } else {
+ ChaosMonkey.stop(solrRunner);
+ }
+ waitForState("Replica still up", collectionName, activeReplicaCount(0,1,0));
+ new UpdateRequest()
+ .add(sdoc("id", "6"))
+ .process(cloudClient, collectionName);
+ ChaosMonkey.start(solrRunner);
+ waitForState("Replica didn't recover", collectionName, activeReplicaCount(0,2,0));
+ // We skip peerSync, so replica will always trigger commit on leader
+ // We query only the non-leader replicas, since we haven't opened a new searcher on the leader yet
+ waitForNumDocsInAllReplicas(4, getNonLeaderReplias(collectionName), 0);// Should be immediate
+
+ // If I add the doc immediately, the leader fails to communicate with the follower with broken pipe.
+ // Options are, wait or retry...
+ for (int i = 0; i < 3; i++) {
+ UpdateRequest ureq = new UpdateRequest().add(sdoc("id", "7"));
+ ureq.setParam("collection", collectionName);
+ ureq.setParam(UpdateRequest.MIN_REPFACT, "2");
+ NamedList<Object> response = cloudClient.request(ureq);
+ if ((Integer)((NamedList<Object>)response.get("responseHeader")).get(UpdateRequest.REPFACT) >= 2) {
+ break;
+ }
+ LOG.info("Min RF not achieved yet. retrying");
+ }
+ checkRTG(3,7, cluster.getJettySolrRunners());
+ DirectUpdateHandler2.commitOnClose = false;
+ ChaosMonkey.stop(solrRunner);
+ waitForState("Replica still up", collectionName, activeReplicaCount(0,1,0));
+ DirectUpdateHandler2.commitOnClose = true;
+ ChaosMonkey.start(solrRunner);
+ waitForState("Replica didn't recover", collectionName, activeReplicaCount(0,2,0));
+ checkRTG(3,7, cluster.getJettySolrRunners());
+ waitForNumDocsInAllReplicas(5, getNonLeaderReplias(collectionName), 0);// Should be immediate
+ cluster.getSolrClient().commit(collectionName);
+
+ // Test replica recovery apply buffer updates
+ Semaphore waitingForBufferUpdates = new Semaphore(0);
+ Semaphore waitingForReplay = new Semaphore(0);
+ RecoveryStrategy.testing_beforeReplayBufferingUpdates = () -> {
+ try {
+ waitingForReplay.release();
+ waitingForBufferUpdates.acquire();
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ fail("Test interrupted: " + e.getMessage());
+ }
+ };
+ if (useKill) {
+ ChaosMonkey.kill(solrRunner);
+ } else {
+ ChaosMonkey.stop(solrRunner);
+ }
+ ChaosMonkey.start(solrRunner);
+ waitingForReplay.acquire();
+ new UpdateRequest()
+ .add(sdoc("id", "8"))
+ .add(sdoc("id", "9"))
+ .process(cloudClient, collectionName);
+ waitingForBufferUpdates.release();
+ RecoveryStrategy.testing_beforeReplayBufferingUpdates = null;
+ waitForState("Replica didn't recover", collectionName, activeReplicaCount(0,2,0));
+ checkRTG(3,9, cluster.getJettySolrRunners());
+ for (SolrCore solrCore : getSolrCore(false)) {
+ RefCounted<IndexWriter> iwRef = solrCore.getUpdateHandler().getSolrCoreState().getIndexWriter(null);
+ assertFalse("IndexWriter at replicas must not see updates ", iwRef.get().hasUncommittedChanges());
+ iwRef.decref();
+ }
+ }
+
+ private List<Replica> getNonLeaderReplias(String collectionName) {
+ return getCollectionState(collectionName).getReplicas().stream().filter(
+ (r)-> !r.getBool("leader", false)).collect(Collectors.toList());
+ }
+
+ public void testDeleteById() throws Exception{
+ createAndWaitForCollection(1,0,2,0);
+ CloudSolrClient cloudClient = cluster.getSolrClient();
+ new UpdateRequest()
+ .deleteByQuery("*:*")
+ .commit(cluster.getSolrClient(), collectionName);
+ new UpdateRequest()
+ .add(sdoc("id", "1"))
+ .commit(cloudClient, collectionName);
+ waitForNumDocsInAllActiveReplicas(1);
+ new UpdateRequest()
+ .deleteById("1")
+ .process(cloudClient, collectionName);
+ boolean successs = false;
+ try {
+ checkRTG(1, 1, cluster.getJettySolrRunners());
+ successs = true;
+ } catch (AssertionError e) {
+ //expected
+ }
+ assertFalse("Doc1 is deleted but it's still exist", successs);
+ }
+
+ public void testBasicLeaderElection() throws Exception {
+ createAndWaitForCollection(1,0,2,0);
+ CloudSolrClient cloudClient = cluster.getSolrClient();
+ new UpdateRequest()
+ .deleteByQuery("*:*")
+ .commit(cluster.getSolrClient(), collectionName);
+ new UpdateRequest()
+ .add(sdoc("id", "1"))
+ .add(sdoc("id", "2"))
+ .process(cloudClient, collectionName);
+ JettySolrRunner oldLeaderJetty = getSolrRunner(true).get(0);
+ ChaosMonkey.kill(oldLeaderJetty);
+ waitForState("Replica not removed", collectionName, activeReplicaCount(0, 1, 0));
+ new UpdateRequest()
+ .add(sdoc("id", "3"))
+ .add(sdoc("id", "4"))
+ .process(cloudClient, collectionName);
+ ChaosMonkey.start(oldLeaderJetty);
+ waitForState("Replica not added", collectionName, activeReplicaCount(0, 2, 0));
+ checkRTG(1,4, cluster.getJettySolrRunners());
+ new UpdateRequest()
+ .commit(cloudClient, collectionName);
+ waitForNumDocsInAllActiveReplicas(4, 0);
+ }
+
+ public void testOutOfOrderDBQWithInPlaceUpdates() throws Exception {
+ createAndWaitForCollection(1,0,2,0);
+ assertFalse(getSolrCore(true).get(0).getLatestSchema().getField("inplace_updatable_int").indexed());
+ assertFalse(getSolrCore(true).get(0).getLatestSchema().getField("inplace_updatable_int").stored());
+ assertTrue(getSolrCore(true).get(0).getLatestSchema().getField("inplace_updatable_int").hasDocValues());
+ List<UpdateRequest> updates = new ArrayList<>();
+ updates.add(simulatedUpdateRequest(null, "id", 1, "title_s", "title0_new", "inplace_updatable_int", 5, "_version_", 1L)); // full update
+ updates.add(simulatedDBQ("inplace_updatable_int:5", 3L));
+ updates.add(simulatedUpdateRequest(1L, "id", 1, "inplace_updatable_int", 6, "_version_", 2L));
+ for (JettySolrRunner solrRunner: getSolrRunner(false)) {
+ try (SolrClient client = solrRunner.newClient()) {
+ for (UpdateRequest up : updates) {
+ up.process(client, collectionName);
+ }
+ }
+ }
+ JettySolrRunner oldLeaderJetty = getSolrRunner(true).get(0);
+ ChaosMonkey.kill(oldLeaderJetty);
+ waitForState("Replica not removed", collectionName, activeReplicaCount(0, 1, 0));
+ ChaosMonkey.start(oldLeaderJetty);
+ waitForState("Replica not added", collectionName, activeReplicaCount(0, 2, 0));
+ checkRTG(1,1, cluster.getJettySolrRunners());
+ SolrDocument doc = cluster.getSolrClient().getById(collectionName,"1");
+ assertNotNull(doc.get("title_s"));
+ }
+
+ private UpdateRequest simulatedUpdateRequest(Long prevVersion, Object... fields) throws SolrServerException, IOException {
+ SolrInputDocument doc = sdoc(fields);
+
+ // get baseUrl of the leader
+ String baseUrl = getBaseUrl();
+
+ UpdateRequest ur = new UpdateRequest();
+ ur.add(doc);
+ ur.setParam("update.distrib", "FROMLEADER");
+ if (prevVersion != null) {
+ ur.setParam("distrib.inplace.prevversion", String.valueOf(prevVersion));
+ ur.setParam("distrib.inplace.update", "true");
+ }
+ ur.setParam("distrib.from", baseUrl);
+ return ur;
+ }
+
+ private UpdateRequest simulatedDBQ(String query, long version) throws SolrServerException, IOException {
+ String baseUrl = getBaseUrl();
+
+ UpdateRequest ur = new UpdateRequest();
+ ur.deleteByQuery(query);
+ ur.setParam("_version_", ""+version);
+ ur.setParam("update.distrib", "FROMLEADER");
+ ur.setParam("distrib.from", baseUrl);
+ return ur;
+ }
+
+ private String getBaseUrl() {
+ DocCollection collection = cluster.getSolrClient().getZkStateReader().getClusterState().getCollection(collectionName);
+ Slice slice = collection.getSlice("shard1");
+ return slice.getLeader().getCoreUrl();
+ }
+
+ private DocCollection createAndWaitForCollection(int numShards, int numNrtReplicas, int numTlogReplicas, int numPullReplicas) throws SolrServerException, IOException, KeeperException, InterruptedException {
+ CollectionAdminRequest.createCollection(collectionName, "conf", numShards, numNrtReplicas, numTlogReplicas, numPullReplicas)
+ .setMaxShardsPerNode(100)
+ .process(cluster.getSolrClient());
+ int numReplicasPerShard = numNrtReplicas + numTlogReplicas + numPullReplicas;
+ cluster.getSolrClient().getZkStateReader().registerCore(collectionName); //TODO: Why is this needed? see SOLR-9440
+ waitForState("Expected collection to be created with " + numShards + " shards and " + numReplicasPerShard + " replicas",
+ collectionName, clusterShape(numShards, numReplicasPerShard));
+ return assertNumberOfReplicas(numNrtReplicas*numShards, numTlogReplicas*numShards, numPullReplicas*numShards, false, true);
+ }
+
+ private void waitForNumDocsInAllActiveReplicas(int numDocs) throws IOException, SolrServerException, InterruptedException {
+ waitForNumDocsInAllActiveReplicas(numDocs, REPLICATION_TIMEOUT_SECS);
+ }
+
+ private void waitForNumDocsInAllActiveReplicas(int numDocs, int timeout) throws IOException, SolrServerException, InterruptedException {
+ DocCollection docCollection = getCollectionState(collectionName);
+ waitForNumDocsInAllReplicas(numDocs, docCollection.getReplicas().stream().filter(r -> r.getState() == Replica.State.ACTIVE).collect(Collectors.toList()), timeout);
+ }
+
+ private void waitForNumDocsInAllReplicas(int numDocs, Collection<Replica> replicas, int timeout) throws IOException, SolrServerException, InterruptedException {
+ waitForNumDocsInAllReplicas(numDocs, replicas, "*:*", timeout);
+ }
+
+ private void waitForNumDocsInAllReplicas(int numDocs, Collection<Replica> replicas, String query, int timeout) throws IOException, SolrServerException, InterruptedException {
+ TimeOut t = new TimeOut(timeout, TimeUnit.SECONDS);
+ for (Replica r:replicas) {
+ if (!r.isActive(cluster.getSolrClient().getZkStateReader().getClusterState().getLiveNodes())) {
+ continue;
+ }
+ try (HttpSolrClient replicaClient = getHttpSolrClient(r.getCoreUrl())) {
+ while (true) {
+ try {
+ assertEquals("Replica " + r.getName() + " not up to date after " + timeout + " seconds",
+ numDocs, replicaClient.query(new SolrQuery(query)).getResults().getNumFound());
+ break;
+ } catch (AssertionError e) {
+ if (t.hasTimedOut()) {
+ throw e;
+ } else {
+ Thread.sleep(100);
+ }
+ }
+ }
+ }
+ }
+ }
+
+ private void waitForDeletion(String collection) throws InterruptedException, KeeperException {
+ TimeOut t = new TimeOut(10, TimeUnit.SECONDS);
+ while (cluster.getSolrClient().getZkStateReader().getClusterState().hasCollection(collection)) {
+ try {
+ Thread.sleep(100);
+ if (t.hasTimedOut()) {
+ fail("Timed out waiting for collection " + collection + " to be deleted.");
+ }
+ cluster.getSolrClient().getZkStateReader().forceUpdateCollection(collection);
+ } catch(SolrException e) {
+ return;
+ }
+
+ }
+ }
+
+ private DocCollection assertNumberOfReplicas(int numNrtReplicas, int numTlogReplicas, int numPullReplicas, boolean updateCollection, boolean activeOnly) throws KeeperException, InterruptedException {
+ if (updateCollection) {
+ cluster.getSolrClient().getZkStateReader().forceUpdateCollection(collectionName);
+ }
+ DocCollection docCollection = getCollectionState(collectionName);
+ assertNotNull(docCollection);
+ assertEquals("Unexpected number of nrt replicas: " + docCollection, numNrtReplicas,
+ docCollection.getReplicas(EnumSet.of(Replica.Type.NRT)).stream().filter(r->!activeOnly || r.getState() == Replica.State.ACTIVE).count());
+ assertEquals("Unexpected number of pull replicas: " + docCollection, numPullReplicas,
+ docCollection.getReplicas(EnumSet.of(Replica.Type.PULL)).stream().filter(r->!activeOnly || r.getState() == Replica.State.ACTIVE).count());
+ assertEquals("Unexpected number of tlog replicas: " + docCollection, numTlogReplicas,
+ docCollection.getReplicas(EnumSet.of(Replica.Type.TLOG)).stream().filter(r->!activeOnly || r.getState() == Replica.State.ACTIVE).count());
+ return docCollection;
+ }
+
+ /*
+ * passes only if all replicas are active or down, and the "liveNodes" reflect the same status
+ */
+ private CollectionStatePredicate clusterStateReflectsActiveAndDownReplicas() {
+ return (liveNodes, collectionState) -> {
+ for (Replica r:collectionState.getReplicas()) {
+ if (r.getState() != Replica.State.DOWN && r.getState() != Replica.State.ACTIVE) {
+ return false;
+ }
+ if (r.getState() == Replica.State.DOWN && liveNodes.contains(r.getNodeName())) {
+ return false;
+ }
+ if (r.getState() == Replica.State.ACTIVE && !liveNodes.contains(r.getNodeName())) {
+ return false;
+ }
+ }
+ return true;
+ };
+ }
+
+
+ private CollectionStatePredicate activeReplicaCount(int numNrtReplicas, int numTlogReplicas, int numPullReplicas) {
+ return (liveNodes, collectionState) -> {
+ int nrtFound = 0, tlogFound = 0, pullFound = 0;
+ if (collectionState == null)
+ return false;
+ for (Slice slice : collectionState) {
+ for (Replica replica : slice) {
+ if (replica.isActive(liveNodes))
+ switch (replica.getType()) {
+ case TLOG:
+ tlogFound++;
+ break;
+ case PULL:
+ pullFound++;
+ break;
+ case NRT:
+ nrtFound++;
+ break;
+ default:
+ throw new AssertionError("Unexpected replica type");
+ }
+ }
+ }
+ return numNrtReplicas == nrtFound && numTlogReplicas == tlogFound && numPullReplicas == pullFound;
+ };
+ }
+
+ private List<SolrCore> getSolrCore(boolean isLeader) {
+ List<SolrCore> rs = new ArrayList<>();
+
+ CloudSolrClient cloudClient = cluster.getSolrClient();
+ DocCollection docCollection = cloudClient.getZkStateReader().getClusterState().getCollection(collectionName);
+
+ for (JettySolrRunner solrRunner : cluster.getJettySolrRunners()) {
+ if (solrRunner.getCoreContainer() == null) continue;
+ for (SolrCore solrCore : solrRunner.getCoreContainer().getCores()) {
+ CloudDescriptor cloudDescriptor = solrCore.getCoreDescriptor().getCloudDescriptor();
+ Slice slice = docCollection.getSlice(cloudDescriptor.getShardId());
+ Replica replica = docCollection.getReplica(cloudDescriptor.getCoreNodeName());
+ if (slice.getLeader().equals(replica) && isLeader) {
+ rs.add(solrCore);
+ } else if (!slice.getLeader().equals(replica) && !isLeader) {
+ rs.add(solrCore);
+ }
+ }
+ }
+ return rs;
+ }
+
+ private void checkRTG(int from, int to, List<JettySolrRunner> solrRunners) throws Exception{
+ for (JettySolrRunner solrRunner: solrRunners) {
+ try (SolrClient client = solrRunner.newClient()) {
+ for (int i = from; i <= to; i++) {
+ SolrQuery query = new SolrQuery();
+ query.set("distrib", false);
+ query.setRequestHandler("/get");
+ query.set("id",i);
+ QueryResponse res = client.query(collectionName, query);
+ assertNotNull("Can not find doc "+ i + " in " + solrRunner.getBaseUrl(),res.getResponse().get("doc"));
+ }
+ }
+ }
+ }
+
+ private List<JettySolrRunner> getSolrRunner(boolean isLeader) {
+ List<JettySolrRunner> rs = new ArrayList<>();
+ CloudSolrClient cloudClient = cluster.getSolrClient();
+ DocCollection docCollection = cloudClient.getZkStateReader().getClusterState().getCollection(collectionName);
+ for (JettySolrRunner solrRunner : cluster.getJettySolrRunners()) {
+ if (solrRunner.getCoreContainer() == null) continue;
+ for (SolrCore solrCore : solrRunner.getCoreContainer().getCores()) {
+ CloudDescriptor cloudDescriptor = solrCore.getCoreDescriptor().getCloudDescriptor();
+ Slice slice = docCollection.getSlice(cloudDescriptor.getShardId());
+ Replica replica = docCollection.getReplica(cloudDescriptor.getCoreNodeName());
+ if (slice.getLeader() == replica && isLeader) {
+ rs.add(solrRunner);
+ } else if (slice.getLeader() != replica && !isLeader) {
+ rs.add(solrRunner);
+ }
+ }
+ }
+ return rs;
+ }
+
+ private void waitForReplicasCatchUp(int numTry) throws IOException, InterruptedException {
+ String leaderTimeCommit = getSolrCore(true).get(0).getDeletionPolicy().getLatestCommit().getUserData().get(SolrIndexWriter.COMMIT_TIME_MSEC_KEY);
+ if (leaderTimeCommit == null) return;
+ for (int i = 0; i < numTry; i++) {
+ boolean inSync = true;
+ for (SolrCore solrCore : getSolrCore(false)) {
+ String replicateTimeCommit = solrCore.getDeletionPolicy().getLatestCommit().getUserData().get(SolrIndexWriter.COMMIT_TIME_MSEC_KEY);
+ if (!leaderTimeCommit.equals(replicateTimeCommit)) {
+ inSync = false;
+ Thread.sleep(500);
+ break;
+ }
+ }
+ if (inSync) return;
+ }
+
+ fail("Some replicas are not in sync with leader");
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/2fc41d56/solr/core/src/test/org/apache/solr/cloud/hdfs/HdfsBasicDistributedZkTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/hdfs/HdfsBasicDistributedZkTest.java b/solr/core/src/test/org/apache/solr/cloud/hdfs/HdfsBasicDistributedZkTest.java
index 1bba523..1534162 100644
--- a/solr/core/src/test/org/apache/solr/cloud/hdfs/HdfsBasicDistributedZkTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/hdfs/HdfsBasicDistributedZkTest.java
@@ -44,8 +44,8 @@ public class HdfsBasicDistributedZkTest extends BasicDistributedZkTest {
}
@Override
- protected int getRealtimeReplicas() {
- return -1;
+ protected boolean useTlogReplicas() {
+ return false;
}
@AfterClass
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/2fc41d56/solr/core/src/test/org/apache/solr/metrics/reporters/solr/SolrCloudReportersTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/metrics/reporters/solr/SolrCloudReportersTest.java b/solr/core/src/test/org/apache/solr/metrics/reporters/solr/SolrCloudReportersTest.java
index df7e642..4184817 100644
--- a/solr/core/src/test/org/apache/solr/metrics/reporters/solr/SolrCloudReportersTest.java
+++ b/solr/core/src/test/org/apache/solr/metrics/reporters/solr/SolrCloudReportersTest.java
@@ -78,7 +78,7 @@ public class SolrCloudReportersTest extends SolrCloudTestCase {
String coreName = core.getName();
String collectionName = core.getCoreDescriptor().getCollectionName();
String coreNodeName = core.getCoreDescriptor().getCloudDescriptor().getCoreNodeName();
- String replicaName = coreName.split("_")[3];
+ String replicaName = coreName.substring(coreName.indexOf("_replica_") + 1);
String shardId = core.getCoreDescriptor().getCloudDescriptor().getShardId();
assertEquals("solr.core." + collectionName + "." + shardId + "." + replicaName, registryName);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/2fc41d56/solr/core/src/test/org/apache/solr/update/TestInPlaceUpdatesDistrib.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/update/TestInPlaceUpdatesDistrib.java b/solr/core/src/test/org/apache/solr/update/TestInPlaceUpdatesDistrib.java
index f9f377c..44b4a4e 100644
--- a/solr/core/src/test/org/apache/solr/update/TestInPlaceUpdatesDistrib.java
+++ b/solr/core/src/test/org/apache/solr/update/TestInPlaceUpdatesDistrib.java
@@ -111,8 +111,8 @@ public class TestInPlaceUpdatesDistrib extends AbstractFullDistribZkTestBase {
}
@Override
- protected int getRealtimeReplicas() {
- return onlyLeaderIndexes? 1 : -1;
+ protected boolean useTlogReplicas() {
+ return onlyLeaderIndexes;
}
@After
@@ -270,7 +270,7 @@ public class TestInPlaceUpdatesDistrib extends AbstractFullDistribZkTestBase {
private void reorderedDBQIndividualReplicaTest() throws Exception {
if (onlyLeaderIndexes) {
- log.info("RTG with DBQs are not working in append replicas");
+ log.info("RTG with DBQs are not working in tlog replicas");
return;
}
clearIndex();
@@ -743,7 +743,7 @@ public class TestInPlaceUpdatesDistrib extends AbstractFullDistribZkTestBase {
*/
private void reorderedDBQsResurrectionTest() throws Exception {
if (onlyLeaderIndexes) {
- log.info("RTG with DBQs are not working in append replicas");
+ log.info("RTG with DBQs are not working in tlog replicas");
return;
}
clearIndex();
@@ -1145,7 +1145,7 @@ public class TestInPlaceUpdatesDistrib extends AbstractFullDistribZkTestBase {
*/
private void reorderedDBQsUsingUpdatedValueFromADroppedUpdate() throws Exception {
if (onlyLeaderIndexes) {
- log.info("RTG with DBQs are not working in append replicas");
+ log.info("RTG with DBQs are not working in tlog replicas");
return;
}
clearIndex();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/2fc41d56/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClient.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClient.java
index fa93503..bc37c13 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClient.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ConcurrentUpdateSolrClient.java
@@ -449,9 +449,12 @@ public class ConcurrentUpdateSolrClient extends SolrClient {
try {
Runner r = new Runner();
runners.add(r);
-
- scheduler.execute(r); // this can throw an exception if the scheduler has been shutdown, but that should be fine.
-
+ try {
+ scheduler.execute(r); // this can throw an exception if the scheduler has been shutdown, but that should be fine.
+ } catch (RuntimeException e) {
+ runners.remove(r);
+ throw e;
+ }
} finally {
MDC.remove("ConcurrentUpdateSolrClient.url");
}
@@ -640,9 +643,15 @@ public class ConcurrentUpdateSolrClient extends SolrClient {
}
private void waitForEmptyQueue() {
+ boolean threadInterrupted = Thread.currentThread().isInterrupted();
while (!queue.isEmpty()) {
if (log.isDebugEnabled()) emptyQueueLoops.incrementAndGet();
+ if (scheduler.isTerminated()) {
+ log.warn("The task queue still has elements but the update scheduler {} is terminated. Can't process any more tasks. "
+ + "Queue size: {}, Runners: {}. Current thread Interrupted? {}", scheduler, queue.size(), runners.size(), threadInterrupted);
+ break;
+ }
synchronized (runners) {
int queueSize = queue.size();
@@ -656,10 +665,16 @@ public class ConcurrentUpdateSolrClient extends SolrClient {
try {
queue.wait(250);
} catch (InterruptedException e) {
- Thread.currentThread().interrupt();
+ // If we set the thread as interrupted again, the next time the wait it's called i t's going to return immediately
+ threadInterrupted = true;
+ log.warn("Thread interrupted while waiting for update queue to be empty. There are still {} elements in the queue.",
+ queue.size());
}
}
}
+ if (threadInterrupted) {
+ Thread.currentThread().interrupt();
+ }
}
public void handleError(Throwable ex) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/2fc41d56/solr/solrj/src/java/org/apache/solr/client/solrj/request/CollectionAdminRequest.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/request/CollectionAdminRequest.java b/solr/solrj/src/java/org/apache/solr/client/solrj/request/CollectionAdminRequest.java
index 0c1116a..6c00b5c 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/request/CollectionAdminRequest.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/request/CollectionAdminRequest.java
@@ -16,6 +16,8 @@
*/
package org.apache.solr.client.solrj.request;
+import static org.apache.solr.common.params.CollectionAdminParams.COUNT_PROP;
+
import java.io.IOException;
import java.util.Collection;
import java.util.Map;
@@ -34,6 +36,7 @@ import org.apache.solr.client.solrj.util.SolrIdentifierValidator;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.ImplicitDocRouter;
+import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.CollectionAdminParams;
import org.apache.solr.common.params.CollectionParams.CollectionAction;
@@ -45,7 +48,6 @@ import org.apache.solr.common.params.SolrParams;
import org.apache.solr.common.util.ContentStream;
import org.apache.solr.common.util.NamedList;
-import static org.apache.solr.common.params.CollectionAdminParams.COUNT_PROP;
import static org.apache.solr.common.params.CollectionAdminParams.CREATE_NODE_SET_PARAM;
import static org.apache.solr.common.params.CollectionAdminParams.CREATE_NODE_SET_SHUFFLE_PARAM;
@@ -323,10 +325,23 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
* @param collection the collection name
* @param config the collection config
* @param numShards the number of shards in the collection
- * @param numReplicas the replication factor of the collection
+ * @param numNrtReplicas the number of {@link org.apache.solr.common.cloud.Replica.Type#NRT} replicas
+ * @param numTlogReplicas the number of {@link org.apache.solr.common.cloud.Replica.Type#TLOG} replicas
+ * @param numPullReplicas the number of {@link org.apache.solr.common.cloud.Replica.Type#PULL} replicas
+ */
+ public static Create createCollection(String collection, String config, int numShards, int numNrtReplicas, int numTlogReplicas, int numPullReplicas) {
+ return new Create(collection, config, numShards, numNrtReplicas, numTlogReplicas, numPullReplicas);
+ }
+
+ /**
+ * Returns a SolrRequest for creating a collection
+ * @param collection the collection name
+ * @param config the collection config
+ * @param numShards the number of shards in the collection
+ * @param numReplicas the replication factor of the collection (same as numNrtReplicas)
*/
public static Create createCollection(String collection, String config, int numShards, int numReplicas) {
- return new Create(collection, config, numShards, numReplicas);
+ return new Create(collection, config, numShards, numReplicas, 0, 0);
}
/**
@@ -340,7 +355,7 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
* @param numReplicas the replication factor of the collection
*/
public static Create createCollection(String collection, int numShards, int numReplicas) {
- return new Create(collection, numShards, numReplicas);
+ return new Create(collection, null, numShards, numReplicas, 0, 0);
}
/**
@@ -353,6 +368,22 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
public static Create createCollectionWithImplicitRouter(String collection, String config, String shards, int numReplicas) {
return new Create(collection, config, shards, numReplicas);
}
+
+ /**
+ * Returns a SolrRequest for creating a collection with the implicit router and specific types of replicas
+ * @param collection the collection name
+ * @param config the collection config
+ * @param shards a shard definition string
+ * @param numNrtReplicas the number of replicas of type {@link org.apache.solr.common.cloud.Replica.Type#NRT}
+ * @param numTlogReplicas the number of replicas of type {@link org.apache.solr.common.cloud.Replica.Type#TLOG}
+ * @param numPullReplicas the number of replicas of type {@link org.apache.solr.common.cloud.Replica.Type#PULL}
+ */
+ public static Create createCollectionWithImplicitRouter(String collection, String config, String shards, int numNrtReplicas, int numTlogReplicas, int numPullReplicas) {
+ Create createRequest = new Create(collection, config, shards, numNrtReplicas);
+ createRequest.tlogReplicas = numTlogReplicas;
+ createRequest.pullReplicas = numPullReplicas;
+ return createRequest;
+ }
// CREATE request
public static class Create extends AsyncCollectionSpecificAdminRequest {
@@ -364,39 +395,36 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
protected String routerField;
protected Integer numShards;
protected Integer maxShardsPerNode;
- protected Integer replicationFactor;
+ protected Integer nrtReplicas;
+ protected Integer pullReplicas;
+ protected Integer tlogReplicas;
private Properties properties;
protected Boolean autoAddReplicas;
- protected Integer realtimeReplicas;
protected Integer stateFormat;
private String[] rule , snitch;
/**
- * @deprecated Use {@link #createCollection(String, String, int, int)}
+ * @deprecated Use {@link #createCollection(String, String, int, int, int, int)}
*/
@Deprecated
public Create() {
super(CollectionAction.CREATE, null);
}
-
- private Create(String collection, String config, int numShards, int numReplicas) {
+
+ private Create(String collection, String config, int numShards, int numNrtReplicas, int numTlogReplicas, int numPullReplicas) { // TODO: maybe add other constructors
super(CollectionAction.CREATE, SolrIdentifierValidator.validateCollectionName(collection));
this.configName = config;
this.numShards = numShards;
- this.replicationFactor = numReplicas;
- }
-
- private Create(String collection, int numShards, int numReplicas) {
- super(CollectionAction.CREATE, SolrIdentifierValidator.validateCollectionName(collection));
- this.numShards = numShards;
- this.replicationFactor = numReplicas;
+ this.nrtReplicas = numNrtReplicas;
+ this.pullReplicas = numPullReplicas;
+ this.tlogReplicas = numTlogReplicas;
}
- private Create(String collection, String config, String shards, int numReplicas) {
+ private Create(String collection, String config, String shards, int numNrtReplicas) {
super(CollectionAction.CREATE, SolrIdentifierValidator.validateCollectionName(collection));
this.configName = config;
- this.replicationFactor = numReplicas;
+ this.nrtReplicas = numNrtReplicas;
this.shards = shards;
this.routerName = ImplicitDocRouter.NAME;
}
@@ -410,9 +438,11 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
public Create setNumShards(Integer numShards) {this.numShards = numShards; return this; }
public Create setMaxShardsPerNode(Integer numShards) { this.maxShardsPerNode = numShards; return this; }
public Create setAutoAddReplicas(boolean autoAddReplicas) { this.autoAddReplicas = autoAddReplicas; return this; }
- public Create setRealtimeReplicas(Integer realtimeReplicas) { this.realtimeReplicas = realtimeReplicas; return this;}
- @Deprecated
- public Create setReplicationFactor(Integer repl) { this.replicationFactor = repl; return this; }
+ public Create setNrtReplicas(Integer nrtReplicas) { this.nrtReplicas = nrtReplicas; return this;}
+ public Create setTlogReplicas(Integer tlogReplicas) { this.tlogReplicas = tlogReplicas; return this;}
+ public Create setPullReplicas(Integer pullReplicas) { this.pullReplicas = pullReplicas; return this;}
+
+ public Create setReplicationFactor(Integer repl) { this.nrtReplicas = repl; return this; }
public Create setStateFormat(Integer stateFormat) { this.stateFormat = stateFormat; return this; }
public Create setRule(String... s){ this.rule = s; return this; }
public Create setSnitch(String... s){ this.snitch = s; return this; }
@@ -423,9 +453,13 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
public String getShards() { return shards; }
public Integer getNumShards() { return numShards; }
public Integer getMaxShardsPerNode() { return maxShardsPerNode; }
- public Integer getReplicationFactor() { return replicationFactor; }
+
+ public Integer getReplicationFactor() { return getNumNrtReplicas(); }
+ public Integer getNumNrtReplicas() { return nrtReplicas; }
public Boolean getAutoAddReplicas() { return autoAddReplicas; }
- public Integer getRealtimeReplicas() { return realtimeReplicas; }
+ public Integer getNumTlogReplicas() {return tlogReplicas;}
+ public Integer getNumPullReplicas() {return pullReplicas;}
+
public Integer getStateFormat() { return stateFormat; }
/**
@@ -506,21 +540,25 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
if (routerField != null) {
params.set("router.field", routerField);
}
- if (replicationFactor != null) {
- params.set( "replicationFactor", replicationFactor);
+ if (nrtReplicas != null) {
+ params.set( "replicationFactor", nrtReplicas);// Keep both for compatibility?
+ params.set( ZkStateReader.NRT_REPLICAS, nrtReplicas);
}
if (autoAddReplicas != null) {
params.set(ZkStateReader.AUTO_ADD_REPLICAS, autoAddReplicas);
}
- if (realtimeReplicas != null) {
- params.set(ZkStateReader.REALTIME_REPLICAS, realtimeReplicas);
- }
if(properties != null) {
addProperties(params, properties);
}
if (stateFormat != null) {
params.set(DocCollection.STATE_FORMAT, stateFormat);
}
+ if (pullReplicas != null) {
+ params.set(ZkStateReader.PULL_REPLICAS, pullReplicas);
+ }
+ if (tlogReplicas != null) {
+ params.set(ZkStateReader.TLOG_REPLICAS, tlogReplicas);
+ }
if(rule != null) params.set("rule", rule);
if(snitch != null) params.set("snitch", snitch);
return params;
@@ -1641,19 +1679,26 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
}
-
+
/**
- * Returns a SolrRequest to add a replica to a shard in a collection
+ * Returns a SolrRequest to add a replica of type {@link org.apache.solr.common.cloud.Replica.Type#NRT} to a shard in a collection
*/
public static AddReplica addReplicaToShard(String collection, String shard) {
- return new AddReplica(collection, shard, null);
+ return addReplicaToShard(collection, shard, Replica.Type.NRT);
+ }
+
+ /**
+ * Returns a SolrRequest to add a replica of the specified type to a shard in a collection
+ */
+ public static AddReplica addReplicaToShard(String collection, String shard, Replica.Type replicaType) {
+ return new AddReplica(collection, shard, null, replicaType);
}
/**
* Returns a SolrRequest to add a replica to a collection using a route key
*/
public static AddReplica addReplicaByRouteKey(String collection, String routeKey) {
- return new AddReplica(collection, null, routeKey);
+ return new AddReplica(collection, null, routeKey, Replica.Type.NRT);
}
// ADDREPLICA request
@@ -1666,6 +1711,7 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
protected String instanceDir;
protected String dataDir;
protected Properties properties;
+ protected Replica.Type type;
/**
* @deprecated Use {@link #addReplicaByRouteKey(String, String)} or {@link #addReplicaToShard(String, String)}
@@ -1675,11 +1721,12 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
super(CollectionAction.ADDREPLICA);
}
- private AddReplica(String collection, String shard, String routeKey) {
+ private AddReplica(String collection, String shard, String routeKey, Replica.Type type) {
super(CollectionAction.ADDREPLICA);
this.collection = collection;
this.shard = shard;
this.routeKey = routeKey;
+ this.type = type;
}
public Properties getProperties() {
@@ -1753,6 +1800,11 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
this.asyncId = id;
return this;
}
+
+ public AddReplica setType(Replica.Type type) {
+ this.type = type;
+ return this;
+ }
@Override
public SolrParams getParams() {
@@ -1778,6 +1830,9 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
if (dataDir != null) {
params.add("dataDir", dataDir);
}
+ if (type != null) {
+ params.add(ZkStateReader.REPLICA_TYPE, type.name());
+ }
if (properties != null) {
addProperties(params, properties);
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/2fc41d56/solr/solrj/src/java/org/apache/solr/common/cloud/DocCollection.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/DocCollection.java b/solr/solrj/src/java/org/apache/solr/common/cloud/DocCollection.java
index d89b2f6..5c3f895 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/DocCollection.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/DocCollection.java
@@ -18,6 +18,7 @@ package org.apache.solr.common.cloud;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
@@ -33,8 +34,10 @@ import org.noggit.JSONWriter;
import static org.apache.solr.common.cloud.ZkStateReader.AUTO_ADD_REPLICAS;
import static org.apache.solr.common.cloud.ZkStateReader.MAX_SHARDS_PER_NODE;
-import static org.apache.solr.common.cloud.ZkStateReader.REALTIME_REPLICAS;
import static org.apache.solr.common.cloud.ZkStateReader.REPLICATION_FACTOR;
+import static org.apache.solr.common.cloud.ZkStateReader.NRT_REPLICAS;
+import static org.apache.solr.common.cloud.ZkStateReader.TLOG_REPLICAS;
+import static org.apache.solr.common.cloud.ZkStateReader.PULL_REPLICAS;
/**
* Models a Collection in zookeeper (but that Java name is obviously taken, hence "DocCollection")
@@ -58,10 +61,11 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
private final String znode;
private final Integer replicationFactor;
+ private final Integer numNrtReplicas;
+ private final Integer numTlogReplicas;
+ private final Integer numPullReplicas;
private final Integer maxShardsPerNode;
private final Boolean autoAddReplicas;
- private final Integer realtimeReplicas;
-
public DocCollection(String name, Map<String, Slice> slices, Map<String, Object> props, DocRouter router) {
this(name, slices, props, router, Integer.MAX_VALUE, ZkStateReader.CLUSTER_STATE);
@@ -83,14 +87,13 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
this.nodeNameLeaderReplicas = new HashMap<>();
this.nodeNameReplicas = new HashMap<>();
this.replicationFactor = (Integer) verifyProp(props, REPLICATION_FACTOR);
+ this.numNrtReplicas = (Integer) verifyProp(props, NRT_REPLICAS);
+ this.numTlogReplicas = (Integer) verifyProp(props, TLOG_REPLICAS);
+ this.numPullReplicas = (Integer) verifyProp(props, PULL_REPLICAS);
this.maxShardsPerNode = (Integer) verifyProp(props, MAX_SHARDS_PER_NODE);
Boolean autoAddReplicas = (Boolean) verifyProp(props, AUTO_ADD_REPLICAS);
this.autoAddReplicas = autoAddReplicas == null ? Boolean.FALSE : autoAddReplicas;
- Integer realtimeReplicas = (Integer) verifyProp(props, REALTIME_REPLICAS);
- this.realtimeReplicas = realtimeReplicas == null ? -1 : realtimeReplicas;
- if (this.realtimeReplicas != -1 && this.realtimeReplicas != 1) {
- throw new SolrException(ErrorCode.SERVER_ERROR, "Invalid realtimeReplicas must be 1 or -1, found:" + this.realtimeReplicas);
- }
+
verifyProp(props, RULE);
verifyProp(props, SNITCH);
Iterator<Map.Entry<String, Slice>> iter = slices.entrySet().iterator();
@@ -133,7 +136,9 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
switch (propName) {
case MAX_SHARDS_PER_NODE:
case REPLICATION_FACTOR:
- case REALTIME_REPLICAS:
+ case NRT_REPLICAS:
+ case PULL_REPLICAS:
+ case TLOG_REPLICAS:
return Integer.parseInt(o.toString());
case AUTO_ADD_REPLICAS:
return Boolean.parseBoolean(o.toString());
@@ -234,10 +239,6 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
return maxShardsPerNode;
}
- public int getRealtimeReplicas() {
- return realtimeReplicas;
- }
-
public String getZNode(){
return znode;
}
@@ -311,6 +312,14 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
}
return replicas;
}
+
+ public List<Replica> getReplicas(EnumSet<Replica.Type> s) {
+ List<Replica> replicas = new ArrayList<>();
+ for (Slice slice : this) {
+ replicas.addAll(slice.getReplicas(s));
+ }
+ return replicas;
+ }
/**
* Get the shardId of a core on a specific node
@@ -333,4 +342,25 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
return super.equals(that) && Objects.equals(this.znode, other.znode) && this.znodeVersion == other.znodeVersion;
}
+ /**
+ * @return the number of replicas of type {@link org.apache.solr.common.cloud.Replica.Type#NRT} this collection was created with
+ */
+ public Integer getNumNrtReplicas() {
+ return numNrtReplicas;
+ }
+
+ /**
+ * @return the number of replicas of type {@link org.apache.solr.common.cloud.Replica.Type#TLOG} this collection was created with
+ */
+ public Integer getNumTlogReplicas() {
+ return numTlogReplicas;
+ }
+
+ /**
+ * @return the number of replicas of type {@link org.apache.solr.common.cloud.Replica.Type#PULL} this collection was created with
+ */
+ public Integer getNumPullReplicas() {
+ return numPullReplicas;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/2fc41d56/solr/solrj/src/java/org/apache/solr/common/cloud/Replica.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/Replica.java b/solr/solrj/src/java/org/apache/solr/common/cloud/Replica.java
index 4968cf2..b7655be 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/Replica.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/Replica.java
@@ -84,9 +84,31 @@ public class Replica extends ZkNodeProps {
}
}
+ public enum Type {
+ /**
+ * Writes updates to transaction log and indexes locally. Replicas of type {@link Type#NRT} support NRT (soft commits) and RTG.
+ * Any {@link Type#NRT} replica can become a leader. A shard leader will forward updates to all active {@link Type#NRT} and
+ * {@link Type#TLOG} replicas.
+ */
+ NRT,
+ /**
+ * Writes to transaction log, but not to index, uses replication. Any {@link Type#TLOG} replica can become leader (by first
+ * applying all local transaction log elements). If a replica is of type {@link Type#TLOG} but is also the leader, it will behave
+ * as a {@link Type#NRT}. A shard leader will forward updates to all active {@link Type#NRT} and {@link Type#TLOG} replicas.
+ */
+ TLOG,
+ /**
+ * Doesn’t index or writes to transaction log. Just replicates from {@link Type#NRT} or {@link Type#TLOG} replicas. {@link Type#PULL}
+ * replicas can’t become shard leaders (i.e., if there are only pull replicas in the collection at some point, updates will fail
+ * same as if there is no leaders, queries continue to work), so they don’t even participate in elections.
+ */
+ PULL
+ }
+
private final String name;
private final String nodeName;
private final State state;
+ private final Type type;
public Replica(String name, Map<String,Object> propMap) {
super(propMap);
@@ -98,6 +120,12 @@ public class Replica extends ZkNodeProps {
this.state = State.ACTIVE; //Default to ACTIVE
propMap.put(ZkStateReader.STATE_PROP, state.toString());
}
+ String typeString = (String)propMap.get(ZkStateReader.REPLICA_TYPE);
+ if (typeString == null) {
+ this.type = Type.NRT;
+ } else {
+ this.type = Type.valueOf(typeString);
+ }
}
@@ -129,6 +157,10 @@ public class Replica extends ZkNodeProps {
public boolean isActive(Set<String> liveNodes) {
return liveNodes.contains(this.nodeName) && this.state == State.ACTIVE;
}
+
+ public Type getType() {
+ return this.type;
+ }
@Override
public String toString() {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/2fc41d56/solr/solrj/src/java/org/apache/solr/common/cloud/Slice.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/Slice.java b/solr/solrj/src/java/org/apache/solr/common/cloud/Slice.java
index bd3bafd..477c1b1 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/Slice.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/Slice.java
@@ -18,6 +18,7 @@ package org.apache.solr.common.cloud;
import java.util.Collection;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
@@ -27,6 +28,7 @@ import java.util.Map;
import java.util.function.Predicate;
import java.util.stream.Collectors;
+import org.apache.solr.common.cloud.Replica.Type;
import org.noggit.JSONUtil;
import org.noggit.JSONWriter;
@@ -161,7 +163,7 @@ public class Slice extends ZkNodeProps implements Iterable<Replica> {
// add the replicas *after* the other properties (for aesthetics, so it's easy to find slice properties in the JSON output)
this.replicas = replicas != null ? replicas : makeReplicas((Map<String,Object>)propMap.get(REPLICAS));
propMap.put(REPLICAS, this.replicas);
-
+
Map<String, Object> rules = (Map<String, Object>) propMap.get("routingRules");
if (rules != null) {
this.routingRules = new HashMap<>();
@@ -202,7 +204,10 @@ public class Slice extends ZkNodeProps implements Iterable<Replica> {
private Replica findLeader() {
for (Replica replica : replicas.values()) {
- if (replica.getStr(LEADER) != null) return replica;
+ if (replica.getStr(LEADER) != null) {
+ assert replica.getType() == Type.TLOG || replica.getType() == Type.NRT: "Pull replica should not become leader!";
+ return replica;
+ }
}
return null;
}
@@ -215,7 +220,7 @@ public class Slice extends ZkNodeProps implements Iterable<Replica> {
}
/**
- * Gets the list of replicas for this slice.
+ * Gets the list of all replicas for this slice.
*/
public Collection<Replica> getReplicas() {
return replicas.values();
@@ -227,6 +232,13 @@ public class Slice extends ZkNodeProps implements Iterable<Replica> {
public List<Replica> getReplicas(Predicate<Replica> pred) {
return replicas.values().stream().filter(pred).collect(Collectors.toList());
}
+
+ /**
+ * Gets the list of replicas that have a type present in s
+ */
+ public List<Replica> getReplicas(EnumSet<Replica.Type> s) {
+ return this.getReplicas(r->s.contains(r.getType()));
+ }
/**
* Get the map of coreNodeName to replicas for this slice.
@@ -238,7 +250,7 @@ public class Slice extends ZkNodeProps implements Iterable<Replica> {
public Map<String,Replica> getReplicasCopy() {
return new LinkedHashMap<>(replicas);
}
-
+
public Replica getLeader() {
return leader;
}
@@ -272,4 +284,5 @@ public class Slice extends ZkNodeProps implements Iterable<Replica> {
public void write(JSONWriter jsonWriter) {
jsonWriter.write(propMap);
}
+
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/2fc41d56/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
index 7cefbee..158b53a 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
@@ -23,6 +23,7 @@ import java.net.URLDecoder;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
@@ -39,6 +40,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
import org.apache.solr.common.Callable;
import org.apache.solr.common.SolrException;
@@ -96,7 +98,9 @@ public class ZkStateReader implements Closeable {
public static final String MAX_SHARDS_PER_NODE = "maxShardsPerNode";
public static final String AUTO_ADD_REPLICAS = "autoAddReplicas";
public static final String MAX_CORES_PER_NODE = "maxCoresPerNode";
- public static final String REALTIME_REPLICAS = "realtimeReplicas";
+ public static final String PULL_REPLICAS = "pullReplicas";
+ public static final String NRT_REPLICAS = "nrtReplicas";
+ public static final String TLOG_REPLICAS = "tlogReplicas";
public static final String ROLES = "/roles.json";
@@ -106,6 +110,8 @@ public class ZkStateReader implements Closeable {
public static final String LEGACY_CLOUD = "legacyCloud";
public static final String URL_SCHEME = "urlScheme";
+
+ public static final String REPLICA_TYPE = "type";
/** A view of the current state of all collections; combines all the different state sources into a single view. */
@@ -780,6 +786,12 @@ public class ZkStateReader implements Closeable {
public List<ZkCoreNodeProps> getReplicaProps(String collection, String shardId, String thisCoreNodeName,
Replica.State mustMatchStateFilter, Replica.State mustNotMatchStateFilter) {
+ //TODO: We don't need all these getReplicaProps method overloading. Also, it's odd that the default is to return replicas of type TLOG and NRT only
+ return getReplicaProps(collection, shardId, thisCoreNodeName, mustMatchStateFilter, null, EnumSet.of(Replica.Type.TLOG, Replica.Type.NRT));
+ }
+
+ public List<ZkCoreNodeProps> getReplicaProps(String collection, String shardId, String thisCoreNodeName,
+ Replica.State mustMatchStateFilter, Replica.State mustNotMatchStateFilter, final EnumSet<Replica.Type> acceptReplicaType) {
assert thisCoreNodeName != null;
ClusterState clusterState = this.clusterState;
if (clusterState == null) {
@@ -798,7 +810,7 @@ public class ZkStateReader implements Closeable {
Map<String,Replica> shardMap = replicas.getReplicasMap();
List<ZkCoreNodeProps> nodes = new ArrayList<>(shardMap.size());
- for (Entry<String,Replica> entry : shardMap.entrySet()) {
+ for (Entry<String,Replica> entry : shardMap.entrySet().stream().filter((e)->acceptReplicaType.contains(e.getValue().getType())).collect(Collectors.toList())) {
ZkCoreNodeProps nodeProps = new ZkCoreNodeProps(entry.getValue());
String coreNodeName = entry.getValue().getName();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/2fc41d56/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java b/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java
index f3e0d7e..3fc6157 100644
--- a/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java
+++ b/solr/solrj/src/java/org/apache/solr/common/params/CoreAdminParams.java
@@ -128,6 +128,11 @@ public abstract class CoreAdminParams
*/
public static final String NEW_COLLECTION = "newCollection";
+ /**
+ * Tells the CoreAdminHandler that the new Core will be a replica of a particular {@link org.apache.solr.common.cloud.Replica.Type}
+ */
+ public static final String REPLICA_TYPE = "replicaType";
+
public enum CoreAdminAction {
STATUS(true),
UNLOAD,
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/2fc41d56/solr/solrj/src/test/org/apache/solr/client/solrj/CollectionAdminRequestRequiredParamsTest.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/CollectionAdminRequestRequiredParamsTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/CollectionAdminRequestRequiredParamsTest.java
index 50f2d18..0f9b147 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/CollectionAdminRequestRequiredParamsTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/CollectionAdminRequestRequiredParamsTest.java
@@ -23,6 +23,8 @@ import java.util.Set;
import org.apache.lucene.util.LuceneTestCase;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.ShardParams;
import org.apache.solr.common.params.SolrParams;
@@ -74,6 +76,13 @@ public class CollectionAdminRequestRequiredParamsTest extends LuceneTestCase {
.setRouteKey("route")
.setCollectionName("collection");
assertContainsParams(request.getParams(), ACTION, COLLECTION, ShardParams._ROUTE_);
+
+ // with type parameter
+ request = new CollectionAdminRequest.AddReplica()
+ .setShardName("shard")
+ .setCollectionName("collection")
+ .setType(Replica.Type.NRT);
+ assertContainsParams(request.getParams(), ACTION, COLLECTION, SHARD, ZkStateReader.REPLICA_TYPE);
}
public void testAddReplicaProp() {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/2fc41d56/solr/test-framework/src/java/org/apache/solr/BaseDistributedSearchTestCase.java
----------------------------------------------------------------------
diff --git a/solr/test-framework/src/java/org/apache/solr/BaseDistributedSearchTestCase.java b/solr/test-framework/src/java/org/apache/solr/BaseDistributedSearchTestCase.java
index 8c6eb60..ed778a6 100644
--- a/solr/test-framework/src/java/org/apache/solr/BaseDistributedSearchTestCase.java
+++ b/solr/test-framework/src/java/org/apache/solr/BaseDistributedSearchTestCase.java
@@ -451,7 +451,7 @@ public abstract class BaseDistributedSearchTestCase extends SolrTestCaseJ4 {
return buildUrl(port, context);
}
- protected void addFields(SolrInputDocument doc, Object... fields) {
+ protected static void addFields(SolrInputDocument doc, Object... fields) {
for (int i = 0; i < fields.length; i += 2) {
doc.addField((String) (fields[i]), fields[i + 1]);
}
@@ -1123,6 +1123,7 @@ public abstract class BaseDistributedSearchTestCase extends SolrTestCaseJ4 {
coreProperties.setProperty("config", "${solrconfig:solrconfig.xml}");
coreProperties.setProperty("schema", "${schema:schema.xml}");
coreProperties.setProperty("coreNodeName", "${coreNodeName:}");
+ coreProperties.setProperty("replicaType", "${replicaType:}");
writeCoreProperties(jettyHome.toPath().resolve("cores").resolve("collection1"), coreProperties, "collection1");
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/2fc41d56/solr/test-framework/src/java/org/apache/solr/SolrTestCaseJ4.java
----------------------------------------------------------------------
diff --git a/solr/test-framework/src/java/org/apache/solr/SolrTestCaseJ4.java b/solr/test-framework/src/java/org/apache/solr/SolrTestCaseJ4.java
index cd49357..a55e2ea 100644
--- a/solr/test-framework/src/java/org/apache/solr/SolrTestCaseJ4.java
+++ b/solr/test-framework/src/java/org/apache/solr/SolrTestCaseJ4.java
@@ -778,8 +778,8 @@ public abstract class SolrTestCaseJ4 extends LuceneTestCase {
* is set.
*/
public static void deleteCore() {
- log.info("###deleteCore" );
if (h != null) {
+ log.info("###deleteCore" );
// If the test case set up Zk, it should still have it as available,
// otherwise the core close will just be unnecessarily delayed.
CoreContainer cc = h.getCoreContainer();
@@ -2123,9 +2123,17 @@ public abstract class SolrTestCaseJ4 extends LuceneTestCase {
SolrDocumentList list1 = (SolrDocumentList) expected;
SolrDocumentList list2 = (SolrDocumentList) actual;
- if(Float.compare(list1.getMaxScore(), list2.getMaxScore()) != 0 || list1.getNumFound() != list2.getNumFound() ||
- list1.getStart() != list2.getStart()) {
+ if (list1.getMaxScore() == null) {
+ if (list2.getMaxScore() != null) {
+ return false;
+ }
+ } else if (list2.getMaxScore() == null) {
return false;
+ } else {
+ if (Float.compare(list1.getMaxScore(), list2.getMaxScore()) != 0 || list1.getNumFound() != list2.getNumFound() ||
+ list1.getStart() != list2.getStart()) {
+ return false;
+ }
}
for(int i=0; i<list1.getNumFound(); i++) {
if(!compareSolrDocument(list1.get(i), list2.get(i))) {
@@ -2425,24 +2433,27 @@ public abstract class SolrTestCaseJ4 extends LuceneTestCase {
}
return result;
}
-
- protected void waitForWarming() throws InterruptedException {
- RefCounted<SolrIndexSearcher> registeredSearcher = h.getCore().getRegisteredSearcher();
- RefCounted<SolrIndexSearcher> newestSearcher = h.getCore().getNewestSearcher(false);
- ;
+
+ protected static void waitForWarming(SolrCore core) throws InterruptedException {
+ RefCounted<SolrIndexSearcher> registeredSearcher = core.getRegisteredSearcher();
+ RefCounted<SolrIndexSearcher> newestSearcher = core.getNewestSearcher(false);
while (registeredSearcher == null || registeredSearcher.get() != newestSearcher.get()) {
if (registeredSearcher != null) {
registeredSearcher.decref();
}
newestSearcher.decref();
Thread.sleep(50);
- registeredSearcher = h.getCore().getRegisteredSearcher();
- newestSearcher = h.getCore().getNewestSearcher(false);
+ registeredSearcher = core.getRegisteredSearcher();
+ newestSearcher = core.getNewestSearcher(false);
}
registeredSearcher.decref();
newestSearcher.decref();
}
+ protected void waitForWarming() throws InterruptedException {
+ waitForWarming(h.getCore());
+ }
+
@BeforeClass
public static void chooseMPForMP() throws Exception {
if (random().nextBoolean()) {