You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by tf...@apache.org on 2017/04/25 23:49:43 UTC
[1/3] lucene-solr:jira/solr-10233: Moved patch to (new) branch.
Updated to master
Repository: lucene-solr
Updated Branches:
refs/heads/jira/solr-10233 [created] 3d149074e
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test/org/apache/solr/cloud/TestPassiveReplica.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestPassiveReplica.java b/solr/core/src/test/org/apache/solr/cloud/TestPassiveReplica.java
new file mode 100644
index 0000000..80cc25e
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/TestPassiveReplica.java
@@ -0,0 +1,549 @@
+/*
+ * 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.TimeUnit;
+import java.util.stream.Collectors;
+
+import org.apache.lucene.util.LuceneTestCase.Slow;
+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.HttpSolrClient;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.response.CollectionAdminResponse;
+import org.apache.solr.client.solrj.response.QueryResponse;
+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.util.TestInjection;
+import org.apache.solr.util.TimeOut;
+import org.apache.zookeeper.KeeperException;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.carrotsearch.randomizedtesting.annotations.Repeat;
+
+@Slow
+public class TestPassiveReplica extends SolrCloudTestCase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ // TODO: Make sure that FORCELEADER can't be used with Passive
+ // TODO: Backup/Snapshot should not work on passive replicas
+ // TODO: ADDSHARD operation
+
+ 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"))
+ .configure();
+ CollectionAdminRequest.ClusterProp clusterPropRequest = CollectionAdminRequest.setClusterProperty(ZkStateReader.LEGACY_CLOUD, "false");
+ CollectionAdminResponse response = clusterPropRequest.process(cluster.getSolrClient());
+ assertEquals(0, response.getStatus());
+ }
+
+ @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());
+ LOG.info("Collection deleted");
+ waitForDeletion(collectionName);
+ }
+ super.tearDown();
+ }
+
+ // Just to compare test time, nocommit
+ @Ignore
+ public void testCreateDelete2() throws Exception {
+ try {
+ CollectionAdminRequest.createCollection(collectionName, "conf", 1, 8, 0, 0).process(cluster.getSolrClient());
+ DocCollection docCollection = getCollectionState(collectionName);
+ assertNotNull(docCollection);
+// assertEquals("Expecting 4 relpicas per shard",
+// 8, docCollection.getReplicas().size());
+// assertEquals("Expecting 6 passive replicas, 3 per shard",
+// 6, docCollection.getReplicas(EnumSet.of(Replica.Type.PASSIVE)).size());
+// assertEquals("Expecting 2 writer replicas, one per shard",
+// 2, docCollection.getReplicas(EnumSet.of(Replica.Type.WRITER)).size());
+// for (Slice s:docCollection.getSlices()) {
+// // read-only replicas can never become leaders
+// assertFalse(s.getLeader().isReadOnly());
+// }
+ } finally {
+ zkClient().printLayoutToStdOut();
+ }
+ }
+
+ @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, 1, 0, 3)
+ .setMaxShardsPerNode(100)
+ .process(cluster.getSolrClient());
+ DocCollection docCollection = getCollectionState(collectionName);
+ assertNotNull(docCollection);
+ assertEquals("Expecting 4 relpicas per shard",
+ 8, docCollection.getReplicas().size());
+ assertEquals("Expecting 6 passive replicas, 3 per shard",
+ 6, docCollection.getReplicas(EnumSet.of(Replica.Type.PASSIVE)).size());
+ assertEquals("Expecting 2 writer replicas, one per shard",
+ 2, docCollection.getReplicas(EnumSet.of(Replica.Type.REALTIME)).size());
+ for (Slice s:docCollection.getSlices()) {
+ // read-only replicas can never become leaders
+ assertFalse(s.getLeader().getType() == Replica.Type.PASSIVE);
+ 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()),
+ 1, shardElectionNodes.size());
+ }
+ assertUlogPresence(docCollection);
+ } finally {
+ zkClient().printLayoutToStdOut();
+ }
+ }
+
+ /**
+ * Asserts that Update logs exist for replicas of type {@link org.apache.solr.common.cloud.Replica.Type#REALTIME}, but not
+ * for replicas of type {@link org.apache.solr.common.cloud.Replica.Type#PASSIVE}
+ */
+ 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);
+ assertEquals("Update log should not exist for replicas of type Passive", r.getType() == Replica.Type.REALTIME,
+ new java.io.File(core.getUlogDir()).exists());
+ } finally {
+ core.close();
+ }
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public void testAddDocs() throws Exception {
+ int numReadOnlyReplicas = 1 + random().nextInt(3);
+ CollectionAdminRequest.createCollection(collectionName, "conf", 1, 1, 0, numReadOnlyReplicas)
+ .setMaxShardsPerNode(100)
+ .process(cluster.getSolrClient());
+ waitForState("Expected collection to be created with 1 shard and " + (numReadOnlyReplicas + 1) + " replicas", collectionName, clusterShape(1, numReadOnlyReplicas + 1));
+ DocCollection docCollection = assertNumberOfReplicas(1, 0, numReadOnlyReplicas, false, true);
+ 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())) {
+ leaderClient.commit(); // TODO: this shouldn't be necessary here
+ 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.PASSIVE))) {
+ //TODO: assert replication < REPLICATION_TIMEOUT_SECS
+ try (HttpSolrClient readOnlyReplicaClient = getHttpSolrClient(r.getCoreUrl())) {
+ while (true) {
+ try {
+ assertEquals("Replica " + r.getName() + " not up to date after 10 seconds",
+ 1, readOnlyReplicaClient.query(new SolrQuery("*:*")).getResults().getNumFound());
+ break;
+ } catch (AssertionError e) {
+ if (t.hasTimedOut()) {
+ throw e;
+ } else {
+ Thread.sleep(100);
+ }
+ }
+ }
+ SolrQuery req = new SolrQuery(
+ "qt", "/admin/plugins",
+ "stats", "true");
+ QueryResponse statsResponse = readOnlyReplicaClient.query(req);
+// assertEquals("Replicas shouldn't process the add document request: " + statsResponse,
+// 0L, ((NamedList<Object>)statsResponse.getResponse()).findRecursive("plugins", "UPDATE", "updateHandler", "stats", "adds"));
+ assertEquals("Replicas shouldn't process the add document request: " + statsResponse,
+ 0L, ((Map<String, Object>)((NamedList<Object>)statsResponse.getResponse()).findRecursive("plugins", "UPDATE", "/update", "stats")).get("UPDATE./update.requests"));
+ }
+ }
+ assertUlogPresence(docCollection);
+ }
+
+ public void testAddRemovePassiveReplica() throws Exception {
+ CollectionAdminRequest.createCollection(collectionName, "conf", 2, 1, 0, 0)
+ .setMaxShardsPerNode(100)
+ .process(cluster.getSolrClient());
+ cluster.getSolrClient().getZkStateReader().registerCore(collectionName); //TODO: Is this needed?
+ waitForState("Expected collection to be created with 2 shards and 1 replica each", collectionName, clusterShape(2, 1));
+ DocCollection docCollection = assertNumberOfReplicas(2, 0, 0, false, true);
+ assertEquals(2, docCollection.getSlices().size());
+
+ CollectionAdminRequest.addReplicaToShard(collectionName, "shard1", Replica.Type.PASSIVE).process(cluster.getSolrClient());
+ docCollection = assertNumberOfReplicas(2, 0, 1, true, false);
+ CollectionAdminRequest.addReplicaToShard(collectionName, "shard2", Replica.Type.PASSIVE).process(cluster.getSolrClient());
+ docCollection = assertNumberOfReplicas(2, 0, 2, true, false);
+
+ waitForState("Expecting collection to have 2 shards and 2 replica each", collectionName, clusterShape(2, 2));
+
+ //Delete passive replica from shard1
+ CollectionAdminRequest.deleteReplica(
+ collectionName,
+ "shard1",
+ docCollection.getSlice("shard1").getReplicas(EnumSet.of(Replica.Type.PASSIVE)).get(0).getName())
+ .process(cluster.getSolrClient());
+ assertNumberOfReplicas(2, 0, 1, true, true);
+ }
+
+ public void testRemoveAllWriterReplicas() throws Exception {
+ doTestNoLeader(true);
+ }
+
+ public void testKillLeader() throws Exception {
+ doTestNoLeader(false);
+ }
+
+ @Ignore("Ignore until I figure out a way to reliably record state transitions")
+ public void testPassiveReplicaStates() throws Exception {
+ // Validate that passive replicas go through the correct states when starting, stopping, reconnecting
+ CollectionAdminRequest.createCollection(collectionName, "conf", 1, 1, 0, 0)
+ .setMaxShardsPerNode(100)
+ .process(cluster.getSolrClient());
+// cluster.getSolrClient().getZkStateReader().registerCore(collectionName); //TODO: Is this needed?
+ waitForState("Replica not added", collectionName, activeReplicaCount(1, 0, 0));
+ addDocs(500);
+ List<Replica.State> statesSeen = new ArrayList<>(3);
+ cluster.getSolrClient().registerCollectionStateWatcher(collectionName, (liveNodes, collectionState) -> {
+ Replica r = collectionState.getSlice("shard1").getReplica("core_node2");
+ LOG.info("CollectionStateWatcher state change: {}", r);
+ if (r == null) {
+ return false;
+ }
+ statesSeen.add(r.getState());
+ LOG.info("CollectionStateWatcher saw state: {}", r.getState());
+ return r.getState() == Replica.State.ACTIVE;
+ });
+ CollectionAdminRequest.addReplicaToShard(collectionName, "shard1", Replica.Type.PASSIVE).process(cluster.getSolrClient());
+ waitForState("Replica not added", collectionName, activeReplicaCount(1, 0, 1));
+ zkClient().printLayoutToStdOut();
+ LOG.info("Saw states: " + Arrays.toString(statesSeen.toArray()));
+ assertEquals("Expecting DOWN->RECOVERING->ACTIVE but saw: " + Arrays.toString(statesSeen.toArray()), 3, statesSeen.size());
+ assertEquals("Expecting DOWN->RECOVERING->ACTIVE but saw: " + Arrays.toString(statesSeen.toArray()), Replica.State.DOWN, statesSeen.get(0));
+ assertEquals("Expecting DOWN->RECOVERING->ACTIVE but saw: " + Arrays.toString(statesSeen.toArray()), Replica.State.RECOVERING, statesSeen.get(0));
+ assertEquals("Expecting DOWN->RECOVERING->ACTIVE but saw: " + Arrays.toString(statesSeen.toArray()), Replica.State.ACTIVE, statesSeen.get(0));
+ }
+
+ public void testPassiveReplicaCantConnectToZooKeeper() {
+
+ }
+
+ public void testRealTimeGet() {
+ // should be redirected to writers
+ }
+
+ /*
+ * validate that replication still happens on a new leader
+ */
+ private void doTestNoLeader(boolean removeReplica) throws Exception {
+ CollectionAdminRequest.createCollection(collectionName, "conf", 1, 1, 0, 1)
+ .setMaxShardsPerNode(100)
+ .process(cluster.getSolrClient());
+ cluster.getSolrClient().getZkStateReader().registerCore(collectionName); //TODO: Is this needed?
+ waitForState("Expected collection to be created with 1 shard and 2 replicas", collectionName, clusterShape(1, 2));
+ DocCollection docCollection = assertNumberOfReplicas(1, 0, 1, false, true);
+
+ // 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.PASSIVE)));
+
+ // Delete leader replica from shard1
+ ignoreException("No registered leader was found"); //These are expected
+ 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, 0, 1, true, true);
+
+ // Check that there is no leader for the shard
+ Replica leader = docCollection.getSlice("shard1").getLeader();
+ assertTrue(leader == null || !leader.isActive(cluster.getSolrClient().getZkStateReader().getClusterState().getLiveNodes()));
+
+ // Passive replica on the other hand should be active
+ Replica passiveReplica = docCollection.getSlice("shard1").getReplicas(EnumSet.of(Replica.Type.PASSIVE)).get(0);
+ assertTrue(passiveReplica.isActive(cluster.getSolrClient().getZkStateReader().getClusterState().getLiveNodes()));
+
+ // add document, this should fail since there is no leader. Passive replica should not accept the update
+ expectThrows(SolrException.class, () ->
+ cluster.getSolrClient().add(collectionName, new SolrInputDocument("id", "2", "foo", "zoo"))
+ );
+
+ // Also fails if I send the update to the passive replica explicitly
+ try (HttpSolrClient passiveReplicaClient = getHttpSolrClient(docCollection.getReplicas(EnumSet.of(Replica.Type.PASSIVE)).get(0).getCoreUrl())) {
+ expectThrows(SolrException.class, () ->
+ cluster.getSolrClient().add(collectionName, new SolrInputDocument("id", "2", "foo", "zoo"))
+ );
+ }
+
+ // Queries should still work
+ waitForNumDocsInAllReplicas(1, docCollection.getReplicas(EnumSet.of(Replica.Type.PASSIVE)));
+ // Add realtime replica back. Since there is no rt now, new rt will have no docs. There will be data loss, since the it will become the leader
+ // and passive replicas will replicate from it. Maybe we want to change this. Replicate from passive replicas is not a good idea, since they
+ // are by definition out of date.
+ if (removeReplica) {
+ CollectionAdminRequest.addReplicaToShard(collectionName, "shard1", Replica.Type.REALTIME).process(cluster.getSolrClient());
+ } else {
+ ChaosMonkey.start(leaderJetty);
+ }
+ waitForState("Expected collection to be 1x2", collectionName, clusterShape(1, 2));
+ unIgnoreException("No registered leader was found"); // Should have a leader from now on
+
+ // Validate that the new writer is the leader now
+ cluster.getSolrClient().getZkStateReader().forceUpdateCollection(collectionName);
+ docCollection = getCollectionState(collectionName);
+ leader = docCollection.getSlice("shard1").getLeader();
+ assertTrue(leader != null && leader.isActive(cluster.getSolrClient().getZkStateReader().getClusterState().getLiveNodes()));
+
+ //nocommit: If jetty is restarted, the replication is not forced, and replica doesn't replicate from leader until new docs are added. Is this the correct behavior? Why should these two cases be different?
+ if (removeReplica) {
+ // Passive replicas will replicate the empty index if a new replica was added and becomes leader
+ waitForNumDocsInAllReplicas(0, docCollection.getReplicas(EnumSet.of(Replica.Type.PASSIVE)));
+ }
+
+ // add docs agin
+ cluster.getSolrClient().add(collectionName, new SolrInputDocument("id", "2", "foo", "zoo"));
+ s = docCollection.getSlices().iterator().next();
+ try (HttpSolrClient leaderClient = getHttpSolrClient(s.getLeader().getCoreUrl())) {
+ leaderClient.commit();
+ assertEquals(1, leaderClient.query(new SolrQuery("*:*")).getResults().getNumFound());
+ }
+ waitForNumDocsInAllReplicas(1, docCollection.getReplicas(EnumSet.of(Replica.Type.PASSIVE)), "id:2");
+ waitForNumDocsInAllReplicas(1, docCollection.getReplicas(EnumSet.of(Replica.Type.PASSIVE)));
+ }
+
+ public void testKillPassiveReplica() throws Exception {
+ CollectionAdminRequest.createCollection(collectionName, "conf", 1, 1, 0, 1)
+ .setMaxShardsPerNode(100)
+ .process(cluster.getSolrClient());
+ cluster.getSolrClient().getZkStateReader().registerCore(collectionName); //TODO: Is this needed?
+ waitForState("Expected collection to be created with 1 shard and 2 replicas", collectionName, clusterShape(1, 2));
+ DocCollection docCollection = assertNumberOfReplicas(1, 0, 1, false, true);
+ assertEquals(1, docCollection.getSlices().size());
+
+ waitForNumDocsInAllActiveReplicas(0);
+ cluster.getSolrClient().add(collectionName, new SolrInputDocument("id", "1", "foo", "bar"));
+ cluster.getSolrClient().commit(collectionName);
+ waitForNumDocsInAllActiveReplicas(1);
+
+ JettySolrRunner passiveReplicaJetty = cluster.getReplicaJetty(docCollection.getSlice("shard1").getReplicas(EnumSet.of(Replica.Type.PASSIVE)).get(0));
+ ChaosMonkey.kill(passiveReplicaJetty);
+ waitForState("Replica not removed", collectionName, activeReplicaCount(1, 0, 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(passiveReplicaJetty);
+ waitForState("Replica not added", collectionName, activeReplicaCount(1, 0, 1));
+ waitForNumDocsInAllActiveReplicas(2);
+ }
+
+ public void testAddDocsToPassive() {
+
+ }
+
+ public void testSearchWhileReplicationHappens() {
+
+ }
+
+ private void waitForNumDocsInAllActiveReplicas(int numDocs) throws IOException, SolrServerException, InterruptedException {
+ DocCollection docCollection = getCollectionState(collectionName);
+ waitForNumDocsInAllReplicas(numDocs, docCollection.getReplicas().stream().filter(r -> r.getState() == Replica.State.ACTIVE).collect(Collectors.toList()));
+ }
+
+ private void waitForNumDocsInAllReplicas(int numDocs, Collection<Replica> replicas) throws IOException, SolrServerException, InterruptedException {
+ waitForNumDocsInAllReplicas(numDocs, replicas, "*:*");
+ }
+
+ private void waitForNumDocsInAllReplicas(int numDocs, Collection<Replica> replicas, String query) throws IOException, SolrServerException, InterruptedException {
+ TimeOut t = new TimeOut(REPLICATION_TIMEOUT_SECS, TimeUnit.SECONDS);
+ for (Replica r:replicas) {
+ try (HttpSolrClient replicaClient = getHttpSolrClient(r.getCoreUrl())) {
+ while (true) {
+ try {
+ assertEquals("Replica " + r.getName() + " not up to date after " + REPLICATION_TIMEOUT_SECS + " 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)) {
+ LOG.info("Collection not yet deleted");
+ 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 numWriter, int numActive, int numPassive, boolean updateCollection, boolean activeOnly) throws KeeperException, InterruptedException {
+ if (updateCollection) {
+ cluster.getSolrClient().getZkStateReader().forceUpdateCollection(collectionName);
+ }
+ DocCollection docCollection = getCollectionState(collectionName);
+ assertNotNull(docCollection);
+ assertEquals("Unexpected number of writer replicas: " + docCollection, numWriter,
+ docCollection.getReplicas(EnumSet.of(Replica.Type.REALTIME)).stream().filter(r->!activeOnly || r.getState() == Replica.State.ACTIVE).count());
+ assertEquals("Unexpected number of passive replicas: " + docCollection, numPassive,
+ docCollection.getReplicas(EnumSet.of(Replica.Type.PASSIVE)).stream().filter(r->!activeOnly || r.getState() == Replica.State.ACTIVE).count());
+ assertEquals("Unexpected number of active replicas: " + docCollection, numActive,
+ docCollection.getReplicas(EnumSet.of(Replica.Type.APPEND)).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 numWriter, int numActive, int numPassive) {
+ return (liveNodes, collectionState) -> {
+ int writersFound = 0, activesFound = 0, passivesFound = 0;
+ if (collectionState == null)
+ return false;
+ for (Slice slice : collectionState) {
+ for (Replica replica : slice) {
+ if (replica.isActive(liveNodes))
+ switch (replica.getType()) {
+ case APPEND:
+ activesFound++;
+ break;
+ case PASSIVE:
+ passivesFound++;
+ break;
+ case REALTIME:
+ writersFound++;
+ break;
+ default:
+ throw new AssertionError("Unexpected replica type");
+ }
+ }
+ }
+ return numWriter == writersFound && numActive == activesFound && numPassive == passivesFound;
+ };
+ }
+
+ private void addDocs(int numDocs) throws SolrServerException, IOException {
+ List<SolrInputDocument> docs = new ArrayList<>(numDocs);
+ for (int i = 0; i < numDocs; i++) {
+ docs.add(new SolrInputDocument("id", String.valueOf(i), "fieldName_s", String.valueOf(i)));
+ }
+ cluster.getSolrClient().add(collectionName, docs);
+ cluster.getSolrClient().commit(collectionName);
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/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..ba92a02 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 useAppendReplicas() {
+ return false;
}
@AfterClass
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/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/3d149074/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..ebb758d 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 useAppendReplicas() {
+ return onlyLeaderIndexes;
}
@After
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java
index ac388d2..0f062f5 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java
@@ -25,6 +25,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -896,7 +897,7 @@ public class CloudSolrClient extends SolrClient {
String url = zkProps.getCoreUrl();
urls.add(url);
if (!directUpdatesToLeadersOnly) {
- for (Replica replica : slice.getReplicas()) {
+ for (Replica replica : slice.getReplicas(EnumSet.of(Replica.Type.APPEND, Replica.Type.REALTIME))) {
if (!replica.getNodeName().equals(leader.getNodeName()) &&
!replica.getName().equals(leader.getName())) {
ZkCoreNodeProps zkProps1 = new ZkCoreNodeProps(replica);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/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 ec43e11..b248f92 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,8 +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;
-
/**
* This class is experimental and subject to change.
*
@@ -321,10 +322,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 numRealtimeReplicas the number of {@link org.apache.solr.common.cloud.Replica.Type#REALTIME} replicas
+ * @param numAppendReplicas the number of {@link org.apache.solr.common.cloud.Replica.Type#APPEND} replicas
+ * @param numPassiveReplicas the number of {@link org.apache.solr.common.cloud.Replica.Type#PASSIVE} replicas
+ */
+ public static Create createCollection(String collection, String config, int numShards, int numRealtimeReplicas, int numAppendReplicas, int numPassiveReplicas) {
+ return new Create(collection, config, numShards, numRealtimeReplicas, numAppendReplicas, numPassiveReplicas);
+ }
+
+ /**
+ * 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
*/
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);
}
/**
@@ -338,7 +352,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);
}
/**
@@ -362,39 +376,36 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
protected String routerField;
protected Integer numShards;
protected Integer maxShardsPerNode;
- protected Integer replicationFactor;
+ protected Integer realtimeReplicas;
+ protected Integer passiveReplicas;
+ protected Integer appendReplicas;
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 numRealtimeReplicas, int numAppendReplicas, int numPassiveReplicas) { // TODO: maybe add other constructors
super(CollectionAction.CREATE, SolrIdentifierValidator.validateCollectionName(collection));
this.configName = config;
this.numShards = numShards;
- this.replicationFactor = numReplicas;
+ this.realtimeReplicas = numRealtimeReplicas;
+ this.passiveReplicas = numPassiveReplicas;
+ this.appendReplicas = numAppendReplicas;
}
- private Create(String collection, int numShards, int numReplicas) {
- super(CollectionAction.CREATE, SolrIdentifierValidator.validateCollectionName(collection));
- this.numShards = numShards;
- this.replicationFactor = numReplicas;
- }
-
- private Create(String collection, String config, String shards, int numReplicas) {
+ private Create(String collection, String config, String shards, int numRealtimeReplicas) {
super(CollectionAction.CREATE, SolrIdentifierValidator.validateCollectionName(collection));
this.configName = config;
- this.replicationFactor = numReplicas;
+ this.realtimeReplicas = numRealtimeReplicas;
this.shards = shards;
this.routerName = ImplicitDocRouter.NAME;
}
@@ -409,8 +420,10 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
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;}
+ public Create setAppendReplicas(Integer appendReplicas) { this.appendReplicas = appendReplicas; return this;}
+
@Deprecated
- public Create setReplicationFactor(Integer repl) { this.replicationFactor = repl; return this; }
+ public Create setReplicationFactor(Integer repl) { this.realtimeReplicas = 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; }
@@ -421,9 +434,17 @@ 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; }
+ /**
+ *
+ * @deprecated Use {@link #getNumRealtimeReplicas()}
+ */
+ @Deprecated
+ public Integer getReplicationFactor() { return getNumRealtimeReplicas(); }
+ public Integer getNumRealtimeReplicas() { return realtimeReplicas; }
public Boolean getAutoAddReplicas() { return autoAddReplicas; }
public Integer getRealtimeReplicas() { return realtimeReplicas; }
+ public Integer getAppendReplicas() {return appendReplicas;}
+
public Integer getStateFormat() { return stateFormat; }
/**
@@ -504,21 +525,25 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
if (routerField != null) {
params.set("router.field", routerField);
}
- if (replicationFactor != null) {
- params.set( "replicationFactor", replicationFactor);
+ if (realtimeReplicas != null) {
+ params.set( "replicationFactor", realtimeReplicas);// Keep both for compatibility?
+ params.set( ZkStateReader.REALTIME_REPLICAS, realtimeReplicas);
}
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 (passiveReplicas != null) {
+ params.set(ZkStateReader.PASSIVE_REPLICAS, passiveReplicas);
+ }
+ if (appendReplicas != null) {
+ params.set(ZkStateReader.APPEND_REPLICAS, appendReplicas);
+ }
if(rule != null) params.set("rule", rule);
if(snitch != null) params.set("snitch", snitch);
return params;
@@ -1615,19 +1640,26 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
}
-
+
/**
* Returns a SolrRequest to add a replica 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.REALTIME);
+ }
+
+ /**
+ * 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.REALTIME);
}
// ADDREPLICA request
@@ -1640,6 +1672,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)}
@@ -1649,11 +1682,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() {
@@ -1727,6 +1761,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() {
@@ -1752,6 +1791,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/3d149074/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..3b409b7 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,7 +34,6 @@ 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;
/**
@@ -60,8 +60,6 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
private final Integer replicationFactor;
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);
@@ -86,11 +84,7 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
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 +127,6 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
switch (propName) {
case MAX_SHARDS_PER_NODE:
case REPLICATION_FACTOR:
- case REALTIME_REPLICAS:
return Integer.parseInt(o.toString());
case AUTO_ADD_REPLICAS:
return Boolean.parseBoolean(o.toString());
@@ -234,10 +227,6 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
return maxShardsPerNode;
}
- public int getRealtimeReplicas() {
- return realtimeReplicas;
- }
-
public String getZNode(){
return znode;
}
@@ -311,6 +300,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
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/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..8f3ed15 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 #REALTIME} support NRT (soft commits) and RTG.
+ * Any {@link #REALTIME} replica can become a leader. A shard leader will forward updates to all active {@link #REALTIME} and
+ * {@link #APPEND} replicas.
+ */
+ REALTIME,
+ /**
+ * Writes to transaction log, but not to index, uses replication. Any {@link #APPEND} replica can become leader (by first
+ * applying all local transaction log elements). If a replica is of type {@link #APPEND} but is also the leader, it will behave
+ * as a {@link #REALTIME}. A shard leader will forward updates to all active {@link #REALTIME} and {@link #APPEND} replicas.
+ */
+ APPEND,
+ /**
+ * Doesn\u2019t index or writes to transaction log. Just replicates from {@link #REALTIME} or {@link #APPEND} replicas. {@link #PASSIVE}
+ * replicas can\u2019t become shard leaders (i.e., if there are only passive replicas in the collection at some point, updates will fail
+ * same as if there is no leaders, queries continue to work), so they don\u2019t even participate in elections.
+ */
+ PASSIVE
+ }
+
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.REALTIME;
+ } 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/3d149074/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..2cd716c 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.APPEND || replica.getType() == Type.REALTIME;
+ 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 state 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/3d149074/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..af7ab26 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;
@@ -92,11 +94,18 @@ public class ZkStateReader implements Closeable {
public static final String REJOIN_AT_HEAD_PROP = "rejoinAtHead";
public static final String SOLR_SECURITY_CONF_PATH = "/security.json";
+ /**
+ *@deprecated Use {@link #REALTIME_REPLICAS}
+ */
+ @Deprecated
public static final String REPLICATION_FACTOR = "replicationFactor";
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";
+ //TODO: Move these constants out of ZkStateReader
+ public static final String PASSIVE_REPLICAS = "passiveReplicas";
public static final String REALTIME_REPLICAS = "realtimeReplicas";
+ public static final String APPEND_REPLICAS = "appendReplicas";
public static final String ROLES = "/roles.json";
@@ -106,6 +115,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 +791,12 @@ public class ZkStateReader implements Closeable {
public List<ZkCoreNodeProps> getReplicaProps(String collection, String shardId, String thisCoreNodeName,
Replica.State mustMatchStateFilter, Replica.State mustNotMatchStateFilter) {
+ //nocommit
+ return getReplicaProps(collection, shardId, thisCoreNodeName, mustMatchStateFilter, null, EnumSet.of(Replica.Type.APPEND, Replica.Type.REALTIME));
+ }
+
+ 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 +815,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/3d149074/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..e64892d 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 this 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/3d149074/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..7a89dec 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.REALTIME);
+ assertContainsParams(request.getParams(), ACTION, COLLECTION, SHARD, ZkStateReader.REPLICA_TYPE);
}
public void testAddReplicaProp() {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/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 54ab06d..c94f24c 100644
--- a/solr/test-framework/src/java/org/apache/solr/SolrTestCaseJ4.java
+++ b/solr/test-framework/src/java/org/apache/solr/SolrTestCaseJ4.java
@@ -776,8 +776,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();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/test-framework/src/java/org/apache/solr/cloud/AbstractDistribZkTestBase.java
----------------------------------------------------------------------
diff --git a/solr/test-framework/src/java/org/apache/solr/cloud/AbstractDistribZkTestBase.java b/solr/test-framework/src/java/org/apache/solr/cloud/AbstractDistribZkTestBase.java
index 7141eed..e5d2619 100644
--- a/solr/test-framework/src/java/org/apache/solr/cloud/AbstractDistribZkTestBase.java
+++ b/solr/test-framework/src/java/org/apache/solr/cloud/AbstractDistribZkTestBase.java
@@ -258,6 +258,7 @@ public abstract class AbstractDistribZkTestBase extends BaseDistributedSearchTes
int maxIterations = 100;
Replica.State coreState = null;
while(maxIterations-->0) {
+ System.out.println("ClusterState" + reader.getClusterState());
Slice slice = reader.getClusterState().getSlice(collection, shard);
if(slice!=null) {
Replica replica = slice.getReplicasMap().get(coreNodeName);
@@ -266,6 +267,8 @@ public abstract class AbstractDistribZkTestBase extends BaseDistributedSearchTes
if(coreState == expectedState) {
return;
}
+ } else {
+ System.out.println(slice.getReplicasMap());
}
}
Thread.sleep(50);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java
----------------------------------------------------------------------
diff --git a/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java b/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java
index 48f7670..e9e0f3d 100644
--- a/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java
+++ b/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java
@@ -272,8 +272,8 @@ public abstract class AbstractFullDistribZkTestBase extends AbstractDistribZkTes
shardToJetty, shardToLeaderJetty);
}
- protected int getRealtimeReplicas() {
- return -1;
+ protected boolean useAppendReplicas() {
+ return false;
}
protected CloudSolrClient createCloudClient(String defaultCollection) {
@@ -387,7 +387,7 @@ public abstract class AbstractFullDistribZkTestBase extends AbstractDistribZkTes
CollectionParams.CollectionAction.CREATE.toLower(), "name",
DEFAULT_COLLECTION, "numShards", String.valueOf(sliceCount),
DocCollection.STATE_FORMAT, getStateFormat(),
- ZkStateReader.REALTIME_REPLICAS, getRealtimeReplicas())));
+ ZkStateReader.REALTIME_REPLICAS, useAppendReplicas())));
zkClient.close();
}
@@ -1570,11 +1570,24 @@ public abstract class AbstractFullDistribZkTestBase extends AbstractDistribZkTes
String shardNames = (String) collectionProps.get(SHARDS_PROP);
numShards = StrUtils.splitSmart(shardNames,',').size();
}
- Integer replicationFactor = (Integer) collectionProps.get(ZkStateReader.REPLICATION_FACTOR);
- if(replicationFactor==null){
- replicationFactor = (Integer) OverseerCollectionMessageHandler.COLL_PROPS.get(ZkStateReader.REPLICATION_FACTOR);
+ Integer numRealtimeReplicas = (Integer) collectionProps.get(ZkStateReader.REALTIME_REPLICAS);
+ if (numRealtimeReplicas == null) {
+ numRealtimeReplicas = (Integer) collectionProps.get(ZkStateReader.REPLICATION_FACTOR);
+ }
+ if(numRealtimeReplicas == null){
+ numRealtimeReplicas = (Integer) OverseerCollectionMessageHandler.COLL_PROPS.get(ZkStateReader.REPLICATION_FACTOR);
+ }
+ if (numRealtimeReplicas == null) {
+ numRealtimeReplicas = Integer.valueOf(0);
+ }
+ Integer numAppendReplicas = (Integer) collectionProps.get(ZkStateReader.APPEND_REPLICAS);
+ if (numAppendReplicas == null) {
+ numAppendReplicas = Integer.valueOf(0);
+ }
+ Integer numPassiveReplicas = (Integer) collectionProps.get(ZkStateReader.PASSIVE_REPLICAS);
+ if (numPassiveReplicas == null) {
+ numPassiveReplicas = Integer.valueOf(0);
}
-
if (confSetName != null) {
params.set("collection.configName", confSetName);
}
@@ -1582,7 +1595,7 @@ public abstract class AbstractFullDistribZkTestBase extends AbstractDistribZkTes
int clientIndex = random().nextInt(2);
List<Integer> list = new ArrayList<>();
list.add(numShards);
- list.add(replicationFactor);
+ list.add(numRealtimeReplicas + numAppendReplicas + numPassiveReplicas);
if (collectionInfos != null) {
collectionInfos.put(collectionName, list);
}
@@ -1610,26 +1623,30 @@ public abstract class AbstractFullDistribZkTestBase extends AbstractDistribZkTes
protected CollectionAdminResponse createCollection(Map<String,List<Integer>> collectionInfos,
String collectionName, int numShards, int replicationFactor, int maxShardsPerNode, SolrClient client, String createNodeSetStr) throws SolrServerException, IOException {
+ int numRealtimeReplicas = useAppendReplicas()?0:replicationFactor;
+ int numAppendReplicas = useAppendReplicas()?replicationFactor:0;
return createCollection(collectionInfos, collectionName,
Utils.makeMap(
NUM_SLICES, numShards,
- ZkStateReader.REPLICATION_FACTOR, replicationFactor,
+ ZkStateReader.REALTIME_REPLICAS, numRealtimeReplicas,
+ ZkStateReader.APPEND_REPLICAS, numAppendReplicas,
CREATE_NODE_SET, createNodeSetStr,
- ZkStateReader.MAX_SHARDS_PER_NODE, maxShardsPerNode,
- ZkStateReader.REALTIME_REPLICAS, getRealtimeReplicas()),
+ ZkStateReader.MAX_SHARDS_PER_NODE, maxShardsPerNode),
client);
}
protected CollectionAdminResponse createCollection(Map<String, List<Integer>> collectionInfos,
String collectionName, int numShards, int replicationFactor, int maxShardsPerNode, SolrClient client, String createNodeSetStr, String configName) throws SolrServerException, IOException {
+ int numRealtimeReplicas = useAppendReplicas()?0:replicationFactor;
+ int numAppendReplicas = useAppendReplicas()?replicationFactor:0;
return createCollection(collectionInfos, collectionName,
Utils.makeMap(
NUM_SLICES, numShards,
- ZkStateReader.REPLICATION_FACTOR, replicationFactor,
+ ZkStateReader.REALTIME_REPLICAS, numRealtimeReplicas,
+ ZkStateReader.APPEND_REPLICAS, numAppendReplicas,
CREATE_NODE_SET, createNodeSetStr,
- ZkStateReader.MAX_SHARDS_PER_NODE, maxShardsPerNode,
- ZkStateReader.REALTIME_REPLICAS, getRealtimeReplicas()),
+ ZkStateReader.MAX_SHARDS_PER_NODE, maxShardsPerNode),
client, configName);
}
@@ -1808,11 +1825,12 @@ public abstract class AbstractFullDistribZkTestBase extends AbstractDistribZkTes
int numShards ) throws Exception {
int maxShardsPerNode = ((((numShards+1) * replicationFactor) / getCommonCloudSolrClient()
.getZkStateReader().getClusterState().getLiveNodes().size())) + 1;
-
+ int numRealtimeReplicas = useAppendReplicas()?0:replicationFactor;
+ int numAppendReplicas = useAppendReplicas()?replicationFactor:0;
Map<String, Object> props = makeMap(
- ZkStateReader.REPLICATION_FACTOR, replicationFactor,
ZkStateReader.MAX_SHARDS_PER_NODE, maxShardsPerNode,
- ZkStateReader.REALTIME_REPLICAS, getRealtimeReplicas(),
+ ZkStateReader.REALTIME_REPLICAS, numRealtimeReplicas,
+ ZkStateReader.APPEND_REPLICAS, numAppendReplicas,
NUM_SLICES, numShards);
Map<String,List<Integer>> collectionInfos = new HashMap<>();
createCollection(collectionInfos, collName, props, client);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/test-framework/src/java/org/apache/solr/cloud/ChaosMonkey.java
----------------------------------------------------------------------
diff --git a/solr/test-framework/src/java/org/apache/solr/cloud/ChaosMonkey.java b/solr/test-framework/src/java/org/apache/solr/cloud/ChaosMonkey.java
index 5cae356..2ad42d1 100644
--- a/solr/test-framework/src/java/org/apache/solr/cloud/ChaosMonkey.java
+++ b/solr/test-framework/src/java/org/apache/solr/cloud/ChaosMonkey.java
@@ -187,7 +187,7 @@ public class ChaosMonkey {
private static void stopJettySolrRunner(JettySolrRunner jetty) throws Exception {
assert(jetty != null);
- monkeyLog("stop shard! " + jetty.getLocalPort());
+ monkeyLog("stop jetty! " + jetty.getLocalPort());
SolrDispatchFilter sdf = jetty.getSolrDispatchFilter();
if (sdf != null) {
try {
@@ -231,7 +231,7 @@ public class ChaosMonkey {
IpTables.blockPort(jetty.getLocalPort());
- monkeyLog("kill shard! " + jetty.getLocalPort());
+ monkeyLog("kill jetty! " + jetty.getLocalPort());
jetty.stop();
[2/3] lucene-solr:jira/solr-10233: Moved patch to (new) branch.
Updated to master
Posted by tf...@apache.org.
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/update/UpdateLog.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/update/UpdateLog.java b/solr/core/src/java/org/apache/solr/update/UpdateLog.java
index c50add4..ddacb19 100644
--- a/solr/core/src/java/org/apache/solr/update/UpdateLog.java
+++ b/solr/core/src/java/org/apache/solr/update/UpdateLog.java
@@ -1706,7 +1706,7 @@ public static final int VERSION_IDX = 1;
public void doReplay(TransactionLog translog) {
try {
- loglog.warn("Starting log replay " + translog + " active=" + activeLog + " starting pos=" + recoveryInfo.positionOfStart);
+ loglog.warn("Starting log replay " + translog + " active=" + activeLog + " starting pos=" + recoveryInfo.positionOfStart + " inSortedOrder=" + inSortedOrder);
long lastStatusTime = System.nanoTime();
if (inSortedOrder) {
tlogReader = translog.getSortedReader(recoveryInfo.positionOfStart);
@@ -1786,7 +1786,7 @@ public static final int VERSION_IDX = 1;
recoveryInfo.adds++;
AddUpdateCommand cmd = convertTlogEntryToAddUpdateCommand(req, entry, oper, version);
cmd.setFlags(UpdateCommand.REPLAY | UpdateCommand.IGNORE_AUTOCOMMIT);
- log.debug("{} {}", oper == ADD ? "add" : "update", cmd);
+ if (debug) log.debug("{} {}", oper == ADD ? "add" : "update", cmd);
proc.processAdd(cmd);
break;
}
@@ -1854,6 +1854,7 @@ public static final int VERSION_IDX = 1;
// something wrong with the request?
}
assert TestInjection.injectUpdateLogReplayRandomPause();
+
}
CommitUpdateCommand cmd = new CommitUpdateCommand(req, false);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java b/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
index cb1b2fb..e9f63d5 100644
--- a/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
+++ b/solr/core/src/java/org/apache/solr/update/processor/DistributedUpdateProcessor.java
@@ -25,6 +25,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -279,7 +280,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
// this is set to true in the constructor if the next processors in the chain
// are custom and may modify the SolrInputDocument racing with its serialization for replication
private final boolean cloneRequiredOnLeader;
- private final boolean onlyLeaderIndexes;
+ private final Replica.Type replicaType;
public DistributedUpdateProcessor(SolrQueryRequest req, SolrQueryResponse rsp, UpdateRequestProcessor next) {
this(req, rsp, new AtomicUpdateDocumentMerger(req), next);
@@ -324,12 +325,10 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
if (cloudDesc != null) {
collection = cloudDesc.getCollectionName();
- ClusterState cstate = zkController.getClusterState();
- DocCollection coll = cstate.getCollection(collection);
- onlyLeaderIndexes = coll.getRealtimeReplicas() == 1;
+ replicaType = cloudDesc.getReplicaType();
} else {
collection = null;
- onlyLeaderIndexes = false;
+ replicaType = Replica.Type.REALTIME;
}
boolean shouldClone = false;
@@ -666,7 +665,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
// used for deleteByQuery to get the list of nodes this leader should forward to
- private List<Node> setupRequest() {
+ private List<Node> setupRequestForDBQ() {
List<Node> nodes = null;
String shardId = cloudDesc.getShardId();
@@ -680,7 +679,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
forwardToLeader = false;
List<ZkCoreNodeProps> replicaProps = zkController.getZkStateReader()
- .getReplicaProps(collection, shardId, leaderReplica.getName(), null, Replica.State.DOWN);
+ .getReplicaProps(collection, shardId, leaderReplica.getName(), null, Replica.State.DOWN, EnumSet.of(Replica.Type.REALTIME, Replica.Type.APPEND));
if (replicaProps != null) {
nodes = new ArrayList<>(replicaProps.size());
for (ZkCoreNodeProps props : replicaProps) {
@@ -1190,7 +1189,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
checkDeleteByQueries = true;
}
}
- if (onlyLeaderIndexes && (cmd.getFlags() & UpdateCommand.REPLAY) == 0) {
+ if (replicaType == Replica.Type.APPEND && (cmd.getFlags() & UpdateCommand.REPLAY) == 0) {
cmd.setFlags(cmd.getFlags() | UpdateCommand.IGNORE_INDEXWRITER);
}
}
@@ -1576,7 +1575,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
if (zkEnabled && DistribPhase.TOLEADER == phase) {
// This core should be a leader
isLeader = true;
- replicas = setupRequest();
+ replicas = setupRequestForDBQ();
} else if (DistribPhase.FROMLEADER == phase) {
isLeader = false;
}
@@ -1610,8 +1609,9 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
String myShardId = req.getCore().getCoreDescriptor().getCloudDescriptor().getShardId();
Replica leaderReplica = zkController.getZkStateReader().getLeaderRetry(
collection, myShardId);
+ // DBQ forwarded to Realtime and Append
List<ZkCoreNodeProps> replicaProps = zkController.getZkStateReader()
- .getReplicaProps(collection, myShardId, leaderReplica.getName(), null, Replica.State.DOWN);
+ .getReplicaProps(collection, myShardId, leaderReplica.getName(), null, Replica.State.DOWN, EnumSet.of(Replica.Type.REALTIME, Replica.Type.APPEND));
if (replicaProps != null) {
final List<Node> myReplicas = new ArrayList<>(replicaProps.size());
for (ZkCoreNodeProps replicaProp : replicaProps) {
@@ -1699,10 +1699,10 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
return;
}
- if (onlyLeaderIndexes && (cmd.getFlags() & UpdateCommand.REPLAY) == 0) {
+ if (replicaType == Replica.Type.APPEND && (cmd.getFlags() & UpdateCommand.REPLAY) == 0) {
+ // Append replica not leader, don't write the DBQ to IW
cmd.setFlags(cmd.getFlags() | UpdateCommand.IGNORE_INDEXWRITER);
}
-
doLocalDelete(cmd);
}
}
@@ -1857,7 +1857,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
}
}
- if (onlyLeaderIndexes && (cmd.getFlags() & UpdateCommand.REPLAY) == 0) {
+ if (replicaType == Replica.Type.APPEND && (cmd.getFlags() & UpdateCommand.REPLAY) == 0) {
cmd.setFlags(cmd.getFlags() | UpdateCommand.IGNORE_INDEXWRITER);
}
}
@@ -1884,14 +1884,14 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
zkCheck();
nodes = getCollectionUrls(req, req.getCore().getCoreDescriptor()
- .getCloudDescriptor().getCollectionName());
+ .getCloudDescriptor().getCollectionName(), EnumSet.of(Replica.Type.APPEND,Replica.Type.REALTIME));
if (isLeader && nodes.size() == 1) {
singleLeader = true;
}
}
if (!zkEnabled || req.getParams().getBool(COMMIT_END_POINT, false) || singleLeader) {
- if (onlyLeaderIndexes) {
+ if (replicaType == Replica.Type.APPEND) { // REALTIME will always commit
try {
Replica leaderReplica = zkController.getZkStateReader().getLeaderRetry(
collection, cloudDesc.getShardId());
@@ -1904,7 +1904,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
doLocalCommit(cmd);
} else {
assert TestInjection.waitForInSyncWithLeader(req.getCore(),
- zkController, collection, cloudDesc.getShardId());
+ zkController, collection, cloudDesc.getShardId()): "Core " + req.getCore() + " not in sync with leader";
}
} catch (InterruptedException e) {
throw new SolrException(ErrorCode.SERVICE_UNAVAILABLE, "Exception finding leader for shard " + cloudDesc.getShardId(), e);
@@ -1958,7 +1958,7 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
- private List<Node> getCollectionUrls(SolrQueryRequest req, String collection) {
+ private List<Node> getCollectionUrls(SolrQueryRequest req, String collection, EnumSet<Replica.Type> types) {
ClusterState clusterState = req.getCore()
.getCoreContainer().getZkController().getClusterState();
Map<String,Slice> slices = clusterState.getSlicesMap(collection);
@@ -1973,6 +1973,10 @@ public class DistributedUpdateProcessor extends UpdateRequestProcessor {
Map<String,Replica> shardMap = replicas.getReplicasMap();
for (Entry<String,Replica> entry : shardMap.entrySet()) {
+ if (!types.contains(entry.getValue().getType())) {
+ log.info("getCollectionUrls: Skipping replica " + entry.getValue().getName());//nocommit: too verbose
+ continue;
+ }
ZkCoreNodeProps nodeProps = new ZkCoreNodeProps(entry.getValue());
if (clusterState.liveNodesContain(nodeProps.getNodeName())) {
urls.add(new StdNode(nodeProps, collection, replicas.getName()));
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test-files/log4j.properties
----------------------------------------------------------------------
diff --git a/solr/core/src/test-files/log4j.properties b/solr/core/src/test-files/log4j.properties
index 2697203..c464a9f 100644
--- a/solr/core/src/test-files/log4j.properties
+++ b/solr/core/src/test-files/log4j.properties
@@ -1,5 +1,5 @@
# Logging level
-log4j.rootLogger=INFO, CONSOLE
+log4j.rootLogger=DEBUG, CONSOLE
log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
log4j.appender.CONSOLE.Target=System.err
@@ -10,6 +10,12 @@ log4j.logger.org.apache.hadoop=WARN
log4j.logger.org.apache.directory=WARN
log4j.logger.org.apache.solr.hadoop=INFO
+log4j.logger.org.apache.solr.cloud.OverseerTaskProcessor=INFO
+log4j.logger.org.apache.solr.cloud.OverseerTaskQueue=INFO
+log4j.logger.org.apache.solr.cloud.OverseerTaskQueue=INFO
+log4j.logger.org.apache.solr.common.cloud.SolrZkClient=INFO
+log4j.logger.org.apache.solr.util.stats.InstrumentedPoolingHttpClientConnectionManager=INFO
+log4j.logger.com.codehale.metrics=INFO
#log4j.logger.org.apache.solr.update.processor.LogUpdateProcessorFactory=DEBUG
#log4j.logger.org.apache.solr.update.processor.DistributedUpdateProcessor=DEBUG
#log4j.logger.org.apache.solr.update.PeerSync=DEBUG
@@ -31,6 +37,6 @@ log4j.logger.org.apache.solr.hadoop=INFO
#log4j.logger.org.apache.http.impl.conn.PoolingHttpClientConnectionManager=DEBUG
#log4j.logger.org.apache.http.impl.conn.BasicClientConnectionManager=DEBUG
-#log4j.logger.org.apache.http=DEBUG
+log4j.logger.org.apache.http=INFO
#log4j.logger.org.apache.solr.client.solrj.impl.SolrHttpRequestRetryHandler=DEBUG
-#log4j.logger.org.eclipse.jetty=DEBUG
+log4j.logger.org.eclipse.jetty=INFO
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test-files/solr/configsets/cloud-minimal/conf/solrconfig.xml
----------------------------------------------------------------------
diff --git a/solr/core/src/test-files/solr/configsets/cloud-minimal/conf/solrconfig.xml b/solr/core/src/test-files/solr/configsets/cloud-minimal/conf/solrconfig.xml
index 059e58f..8da7d28 100644
--- a/solr/core/src/test-files/solr/configsets/cloud-minimal/conf/solrconfig.xml
+++ b/solr/core/src/test-files/solr/configsets/cloud-minimal/conf/solrconfig.xml
@@ -33,7 +33,7 @@
<commitWithin>
<softCommit>${solr.commitwithin.softcommit:true}</softCommit>
</commitWithin>
- <updateLog></updateLog>
+ <updateLog class="${solr.ulog:solr.UpdateLog}"></updateLog>
</updateHandler>
<requestHandler name="/select" class="solr.SearchHandler">
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZk2Test.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZk2Test.java b/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZk2Test.java
index 5eb4b3b..51f9fe9 100644
--- a/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZk2Test.java
+++ b/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZk2Test.java
@@ -62,8 +62,8 @@ public class BasicDistributedZk2Test extends AbstractFullDistribZkTestBase {
}
@Override
- protected int getRealtimeReplicas() {
- return onlyLeaderIndexes? 1 : -1;
+ protected boolean useAppendReplicas() {
+ return onlyLeaderIndexes;
}
@Test
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZkTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZkTest.java b/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZkTest.java
index 1c23c9c..8d0839f 100644
--- a/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZkTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/BasicDistributedZkTest.java
@@ -119,8 +119,8 @@ public class BasicDistributedZkTest extends AbstractFullDistribZkTestBase {
}
@Override
- protected int getRealtimeReplicas() {
- return onlyLeaderIndexes? 1 : -1;
+ protected boolean useAppendReplicas() {
+ return onlyLeaderIndexes;
}
@Override
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test/org/apache/solr/cloud/ChaosMonkeyNothingIsSafeTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/ChaosMonkeyNothingIsSafeTest.java b/solr/core/src/test/org/apache/solr/cloud/ChaosMonkeyNothingIsSafeTest.java
index ffc5262..a389005 100644
--- a/solr/core/src/test/org/apache/solr/cloud/ChaosMonkeyNothingIsSafeTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/ChaosMonkeyNothingIsSafeTest.java
@@ -112,8 +112,8 @@ public class ChaosMonkeyNothingIsSafeTest extends AbstractFullDistribZkTestBase
}
@Override
- protected int getRealtimeReplicas() {
- return onlyLeaderIndexes? 1 : -1;
+ protected boolean useAppendReplicas() {
+ return onlyLeaderIndexes;
}
@Test
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test/org/apache/solr/cloud/CollectionsAPIDistributedZkTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/CollectionsAPIDistributedZkTest.java b/solr/core/src/test/org/apache/solr/cloud/CollectionsAPIDistributedZkTest.java
index ed9ed41..e75a854 100644
--- a/solr/core/src/test/org/apache/solr/cloud/CollectionsAPIDistributedZkTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/CollectionsAPIDistributedZkTest.java
@@ -286,7 +286,7 @@ public class CollectionsAPIDistributedZkTest extends SolrCloudTestCase {
// first we make a core with the core name the collections api
// will try and use - this will cause our mock fail
Create createCmd = new Create();
- createCmd.setCoreName("halfcollection_shard1_replica1");
+ createCmd.setCoreName(Assign.buildCoreName("halfcollection", "shard1", Replica.Type.REALTIME, 1));
createCmd.setCollection("halfcollectionblocker");
String dataDir = createTempDir().toFile().getAbsolutePath();
createCmd.setDataDir(dataDir);
@@ -298,7 +298,7 @@ public class CollectionsAPIDistributedZkTest extends SolrCloudTestCase {
}
createCmd = new Create();
- createCmd.setCoreName("halfcollection_shard1_replica1");
+ createCmd.setCoreName(Assign.buildCoreName("halfcollection", "shard1", Replica.Type.REALTIME, 1));
createCmd.setCollection("halfcollectionblocker2");
dataDir = createTempDir().toFile().getAbsolutePath();
createCmd.setDataDir(dataDir);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test/org/apache/solr/cloud/CollectionsAPISolrJTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/CollectionsAPISolrJTest.java b/solr/core/src/test/org/apache/solr/cloud/CollectionsAPISolrJTest.java
index 3e0d840..e2a80b6 100644
--- a/solr/core/src/test/org/apache/solr/cloud/CollectionsAPISolrJTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/CollectionsAPISolrJTest.java
@@ -62,7 +62,7 @@ public class CollectionsAPISolrJTest extends SolrCloudTestCase {
Map<String, NamedList<Integer>> coresStatus = response.getCollectionCoresStatus();
assertEquals(4, coresStatus.size());
for (int i=0; i<4; i++) {
- NamedList<Integer> status = coresStatus.get(collectionName + "_shard" + (i/2+1) + "_replica" + (i%2+1));
+ NamedList<Integer> status = coresStatus.get(Assign.buildCoreName(collectionName, "shard" + (i/2+1), Replica.Type.REALTIME, (i%2+1)));
assertEquals(0, (int)status.get("status"));
assertTrue(status.get("QTime") > 0);
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test/org/apache/solr/cloud/ForceLeaderTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/ForceLeaderTest.java b/solr/core/src/test/org/apache/solr/cloud/ForceLeaderTest.java
index 8904ea8..db9ecb4 100644
--- a/solr/core/src/test/org/apache/solr/cloud/ForceLeaderTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/ForceLeaderTest.java
@@ -58,8 +58,8 @@ public class ForceLeaderTest extends HttpPartitionTest {
private final boolean onlyLeaderIndexes = random().nextBoolean();
@Override
- protected int getRealtimeReplicas() {
- return onlyLeaderIndexes? 1 : -1;
+ protected boolean useAppendReplicas() {
+ return onlyLeaderIndexes;
}
@Test
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test/org/apache/solr/cloud/HttpPartitionTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/HttpPartitionTest.java b/solr/core/src/test/org/apache/solr/cloud/HttpPartitionTest.java
index 01002cf..d0b0c5e 100644
--- a/solr/core/src/test/org/apache/solr/cloud/HttpPartitionTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/HttpPartitionTest.java
@@ -85,8 +85,8 @@ public class HttpPartitionTest extends AbstractFullDistribZkTestBase {
}
@Override
- protected int getRealtimeReplicas() {
- return onlyLeaderIndexes? 1 : -1;
+ protected boolean useAppendReplicas() {
+ return onlyLeaderIndexes;
}
/**
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test/org/apache/solr/cloud/LeaderInitiatedRecoveryOnCommitTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/LeaderInitiatedRecoveryOnCommitTest.java b/solr/core/src/test/org/apache/solr/cloud/LeaderInitiatedRecoveryOnCommitTest.java
index 457b9d9..fd1b403 100644
--- a/solr/core/src/test/org/apache/solr/cloud/LeaderInitiatedRecoveryOnCommitTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/LeaderInitiatedRecoveryOnCommitTest.java
@@ -46,8 +46,8 @@ public class LeaderInitiatedRecoveryOnCommitTest extends BasicDistributedZkTest
}
@Override
- protected int getRealtimeReplicas() {
- return onlyLeaderIndexes? 1 : -1;
+ protected boolean useAppendReplicas() {
+ return onlyLeaderIndexes;
}
@Override
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test/org/apache/solr/cloud/OnlyLeaderIndexesTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/OnlyLeaderIndexesTest.java b/solr/core/src/test/org/apache/solr/cloud/OnlyLeaderIndexesTest.java
index a4e8d6f..6297408 100644
--- a/solr/core/src/test/org/apache/solr/cloud/OnlyLeaderIndexesTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/OnlyLeaderIndexesTest.java
@@ -19,6 +19,7 @@ package org.apache.solr.cloud;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.EnumSet;
import java.util.List;
import java.util.concurrent.Semaphore;
@@ -30,9 +31,11 @@ import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
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.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;
@@ -44,6 +47,7 @@ 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.zookeeper.KeeperException;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -59,10 +63,13 @@ public class OnlyLeaderIndexesTest extends SolrCloudTestCase {
.addConfig("config", TEST_PATH().resolve("configsets")
.resolve("cloud-minimal-inplace-updates").resolve("conf"))
.configure();
+
+ CollectionAdminRequest.ClusterProp clusterPropRequest = CollectionAdminRequest.setClusterProperty(ZkStateReader.LEGACY_CLOUD, "false");
+ CollectionAdminResponse response = clusterPropRequest.process(cluster.getSolrClient());
+ assertEquals(0, response.getStatus());
CollectionAdminRequest
- .createCollection(COLLECTION, "config", 1, 3)
- .setRealtimeReplicas(1)
+ .createCollection(COLLECTION, "config", 1, 0, 3, 0)
.setMaxShardsPerNode(1)
.process(cluster.getSolrClient());
AbstractDistribZkTestBase.waitForRecoveriesToFinish(COLLECTION, cluster.getSolrClient().getZkStateReader(),
@@ -71,6 +78,7 @@ public class OnlyLeaderIndexesTest extends SolrCloudTestCase {
@Test
public void test() throws Exception {
+ assertNumberOfReplicas(0, 3, 0, false, true);
basicTest();
recoveryTest();
dbiTest();
@@ -252,6 +260,7 @@ public class OnlyLeaderIndexesTest extends SolrCloudTestCase {
.add(sdoc("id", "4"))
.process(cloudClient, COLLECTION);
ChaosMonkey.start(oldLeaderJetty);
+ waitForState("Replica not removed", "collection1", activeReplicaCount(0, 3, 0));
AbstractDistribZkTestBase.waitForRecoveriesToFinish(COLLECTION, cluster.getSolrClient().getZkStateReader(),
false, true, 60);
checkRTG(1,4, cluster.getJettySolrRunners());
@@ -347,7 +356,7 @@ public class OnlyLeaderIndexesTest extends SolrCloudTestCase {
}
private void checkShardConsistency(int expected, int numTry) throws Exception{
-
+ String replicaNotInSync = null;
for (int i = 0; i < numTry; i++) {
boolean inSync = true;
for (JettySolrRunner solrRunner: cluster.getJettySolrRunners()) {
@@ -357,15 +366,16 @@ public class OnlyLeaderIndexesTest extends SolrCloudTestCase {
long results = client.query(COLLECTION, query).getResults().getNumFound();
if (expected != results) {
inSync = false;
- Thread.sleep(500);
+ replicaNotInSync = solrRunner.getNodeName();
break;
}
}
}
if (inSync) return;
+ Thread.sleep(500);
}
- fail("Some replicas are not in sync with leader");
+ fail("Some replicas are not in sync with leader: " + replicaNotInSync);
}
private void waitForReplicasCatchUp(int numTry) throws IOException, InterruptedException {
@@ -431,5 +441,48 @@ public class OnlyLeaderIndexesTest extends SolrCloudTestCase {
}
return rs;
}
+
+ // TODO: This is copy/paste from TestPassiveReplica, refactor
+ private DocCollection assertNumberOfReplicas(int numWriter, int numActive, int numPassive, boolean updateCollection, boolean activeOnly) throws KeeperException, InterruptedException {
+ if (updateCollection) {
+ cluster.getSolrClient().getZkStateReader().forceUpdateCollection("collection1");
+ }
+ DocCollection docCollection = getCollectionState("collection1");
+ assertNotNull(docCollection);
+ assertEquals("Unexpected number of writer replicas: " + docCollection, numWriter,
+ docCollection.getReplicas(EnumSet.of(Replica.Type.REALTIME)).stream().filter(r->!activeOnly || r.getState() == Replica.State.ACTIVE).count());
+ assertEquals("Unexpected number of passive replicas: " + docCollection, numPassive,
+ docCollection.getReplicas(EnumSet.of(Replica.Type.PASSIVE)).stream().filter(r->!activeOnly || r.getState() == Replica.State.ACTIVE).count());
+ assertEquals("Unexpected number of active replicas: " + docCollection, numActive,
+ docCollection.getReplicas(EnumSet.of(Replica.Type.APPEND)).stream().filter(r->!activeOnly || r.getState() == Replica.State.ACTIVE).count());
+ return docCollection;
+ }
+
+ private CollectionStatePredicate activeReplicaCount(int numWriter, int numActive, int numPassive) {
+ return (liveNodes, collectionState) -> {
+ int writersFound = 0, activesFound = 0, passivesFound = 0;
+ if (collectionState == null)
+ return false;
+ for (Slice slice : collectionState) {
+ for (Replica replica : slice) {
+ if (replica.isActive(liveNodes))
+ switch (replica.getType()) {
+ case APPEND:
+ activesFound++;
+ break;
+ case PASSIVE:
+ passivesFound++;
+ break;
+ case REALTIME:
+ writersFound++;
+ break;
+ default:
+ throw new AssertionError("Unexpected replica type");
+ }
+ }
+ }
+ return numWriter == writersFound && numActive == activesFound && numPassive == passivesFound;
+ };
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionConfigSetProcessorTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionConfigSetProcessorTest.java b/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionConfigSetProcessorTest.java
index 48ac91f..340adbb 100644
--- a/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionConfigSetProcessorTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionConfigSetProcessorTest.java
@@ -27,6 +27,7 @@ import org.apache.solr.client.solrj.SolrResponse;
import org.apache.solr.cloud.Overseer.LeaderStatus;
import org.apache.solr.cloud.OverseerTaskQueue.QueueEvent;
import org.apache.solr.common.cloud.ClusterState;
+import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
@@ -346,7 +347,7 @@ public class OverseerCollectionConfigSetProcessorTest extends SolrTestCaseJ4 {
assertEquals(numberOfSlices * numberOfReplica, coreNames.size());
for (int i = 1; i <= numberOfSlices; i++) {
for (int j = 1; j <= numberOfReplica; j++) {
- String coreName = COLLECTION_NAME + "_shard" + i + "_replica" + j;
+ String coreName = Assign.buildCoreName(COLLECTION_NAME, "shard" + i, Replica.Type.REALTIME, j);
assertTrue("Shard " + coreName + " was not created",
coreNames.contains(coreName));
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test/org/apache/solr/cloud/OverseerTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/OverseerTest.java b/solr/core/src/test/org/apache/solr/cloud/OverseerTest.java
index b0721a2..f5c5db0 100644
--- a/solr/core/src/test/org/apache/solr/cloud/OverseerTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/OverseerTest.java
@@ -1210,7 +1210,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
ZkStateReader.STATE_PROP, Replica.State.RECOVERING.toString());
q.offer(Utils.toJSON(m));
-
+
m = new ZkNodeProps(Overseer.QUEUE_OPERATION, OverseerAction.STATE.toLower(),
ZkStateReader.BASE_URL_PROP, "http://127.0.0.1/solr",
@@ -1221,7 +1221,7 @@ public class OverseerTest extends SolrTestCaseJ4 {
ZkStateReader.STATE_PROP, Replica.State.ACTIVE.toString());
q.offer(Utils.toJSON(m));
-
+
Stat stat = new Stat();
byte[] data = zkClient.getData("/clusterstate.json", null, stat, true);
// Simulate an external modification
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test/org/apache/solr/cloud/RecoveryAfterSoftCommitTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/RecoveryAfterSoftCommitTest.java b/solr/core/src/test/org/apache/solr/cloud/RecoveryAfterSoftCommitTest.java
index a8e14bf..1da9aca 100644
--- a/solr/core/src/test/org/apache/solr/cloud/RecoveryAfterSoftCommitTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/RecoveryAfterSoftCommitTest.java
@@ -40,8 +40,8 @@ public class RecoveryAfterSoftCommitTest extends AbstractFullDistribZkTestBase {
}
@Override
- protected int getRealtimeReplicas() {
- return onlyLeaderIndexes? 1: -1;
+ protected boolean useAppendReplicas() {
+ return onlyLeaderIndexes;
}
@BeforeClass
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test/org/apache/solr/cloud/ShardSplitTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/ShardSplitTest.java b/solr/core/src/test/org/apache/solr/cloud/ShardSplitTest.java
index bf9b5e0..e00ea0d 100644
--- a/solr/core/src/test/org/apache/solr/cloud/ShardSplitTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/ShardSplitTest.java
@@ -79,6 +79,11 @@ public class ShardSplitTest extends BasicDistributedZkTest {
public ShardSplitTest() {
schemaString = "schema15.xml"; // we need a string id
}
+
+ @Override
+ protected boolean useAppendReplicas() {
+ return false;
+ }
@Override
public void distribSetUp() throws Exception {
@@ -86,12 +91,6 @@ public class ShardSplitTest extends BasicDistributedZkTest {
useFactory(null);
}
- //TODO for now, onlyLeaderIndexes do not work with ShardSplitTest
- @Override
- protected int getRealtimeReplicas() {
- return -1;
- }
-
@Test
public void test() throws Exception {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test/org/apache/solr/cloud/TestAppendReplica.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestAppendReplica.java b/solr/core/src/test/org/apache/solr/cloud/TestAppendReplica.java
new file mode 100644
index 0000000..fe353d2
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/TestAppendReplica.java
@@ -0,0 +1,808 @@
+/*
+ * 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.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+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.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.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.BeforeClass;
+import org.junit.Ignore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.carrotsearch.randomizedtesting.annotations.Repeat;
+
+@Slow
+public class TestAppendReplica 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();
+ CollectionAdminRequest.ClusterProp clusterPropRequest = CollectionAdminRequest.setClusterProperty(ZkStateReader.LEGACY_CLOUD, "false");
+ CollectionAdminResponse response = clusterPropRequest.process(cluster.getSolrClient());
+ assertEquals(0, response.getStatus());
+ }
+
+ @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();
+ }
+
+ // Just to compare test time, nocommit
+ @Ignore
+ public void testCreateDelete2() throws Exception {
+ try {
+ CollectionAdminRequest.createCollection(collectionName, "conf", 1, 8, 0, 0).process(cluster.getSolrClient());
+ DocCollection docCollection = getCollectionState(collectionName);
+ assertNotNull(docCollection);
+// assertEquals("Expecting 4 relpicas per shard",
+// 8, docCollection.getReplicas().size());
+// assertEquals("Expecting 6 passive replicas, 3 per shard",
+// 6, docCollection.getReplicas(EnumSet.of(Replica.Type.PASSIVE)).size());
+// assertEquals("Expecting 2 writer replicas, one per shard",
+// 2, docCollection.getReplicas(EnumSet.of(Replica.Type.WRITER)).size());
+// for (Slice s:docCollection.getSlices()) {
+// // read-only replicas can never become leaders
+// assertFalse(s.getLeader().isReadOnly());
+// }
+ } finally {
+ zkClient().printLayoutToStdOut();
+ }
+ }
+
+ /**
+ * Asserts that Update logs exist for replicas of type {@link org.apache.solr.common.cloud.Replica.Type#REALTIME}, but not
+ * for replicas of type {@link org.apache.solr.common.cloud.Replica.Type#PASSIVE}
+ */
+ 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());
+ 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 append replicas, 4 per shard",
+ 8, docCollection.getReplicas(EnumSet.of(Replica.Type.APPEND)).size());
+ assertEquals("Expecting no realtime replicas",
+ 0, docCollection.getReplicas(EnumSet.of(Replica.Type.REALTIME)).size());
+ assertEquals("Expecting no passive replicas",
+ 0, docCollection.getReplicas(EnumSet.of(Replica.Type.PASSIVE)).size());
+ for (Slice s:docCollection.getSlices()) {
+ assertTrue(s.getLeader().getType() == Replica.Type.APPEND);
+ 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);
+ } finally {
+ zkClient().printLayoutToStdOut();
+ }
+ }
+
+ public void testAddDocs() throws Exception {
+ int numAppendReplicas = 1 + random().nextInt(3);
+ DocCollection docCollection = createAndWaitForCollection(1, 0, numAppendReplicas, 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.APPEND))) {
+ //TODO: assert replication < REPLICATION_TIMEOUT_SECS
+ try (HttpSolrClient appendReplicaClient = getHttpSolrClient(r.getCoreUrl())) {
+ while (true) {
+ try {
+ assertEquals("Replica " + r.getName() + " not up to date after 10 seconds",
+ 1, appendReplicaClient.query(new SolrQuery("*:*")).getResults().getNumFound());
+ // Append replicas process all updates
+ SolrQuery req = new SolrQuery(
+ "qt", "/admin/plugins",
+ "stats", "true");
+ QueryResponse statsResponse = appendReplicaClient.query(req);
+// TODO: uncomment when SOLR-10569 is fixed
+// assertEquals("Append replicas should recive all updates. Replica: " + r + ", response: " + statsResponse,
+// 1L, ((NamedList<Object>)statsResponse.getResponse()).findRecursive("plugins", "UPDATE", "updateHandler", "stats", "cumulative_adds"));
+ break;
+ } catch (AssertionError e) {
+ if (t.hasTimedOut()) {
+ throw e;
+ } else {
+ Thread.sleep(100);
+ }
+ }
+ }
+ }
+ }
+ assertUlogPresence(docCollection);
+ }
+
+ public void testAddRemoveAppendReplica() throws Exception {
+ DocCollection docCollection = createAndWaitForCollection(2, 0, 1, 0);
+ assertEquals(2, docCollection.getSlices().size());
+
+ CollectionAdminRequest.addReplicaToShard(collectionName, "shard1", Replica.Type.APPEND).process(cluster.getSolrClient());
+ docCollection = assertNumberOfReplicas(0, 3, 0, true, false);
+ CollectionAdminRequest.addReplicaToShard(collectionName, "shard2", Replica.Type.APPEND).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 passive replica from shard1
+ CollectionAdminRequest.deleteReplica(
+ collectionName,
+ "shard1",
+ docCollection.getSlice("shard1").getReplicas(EnumSet.of(Replica.Type.APPEND)).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 testPassiveReplicaStates() {
+ // Validate that passive replicas go through the correct states when starting, stopping, reconnecting
+ }
+
+ public void testPassiveReplicaCantConnectToZooKeeper() {
+
+ }
+
+ public void testRealTimeGet() {
+ // should be redirected to writers or error
+ }
+
+ /*
+ * 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.APPEND)), 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.APPEND)), REPLICATION_TIMEOUT_SECS);
+ // Start back the node
+ if (removeReplica) {
+ CollectionAdminRequest.addReplicaToShard(collectionName, "shard1", Replica.Type.APPEND).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.APPEND)), REPLICATION_TIMEOUT_SECS);
+ }
+
+ public void testKillAppendReplica() 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 passiveReplicaJetty = cluster.getReplicaJetty(docCollection.getSlice("shard1").getReplicas(EnumSet.of(Replica.Type.APPEND)).get(0));
+ ChaosMonkey.kill(passiveReplicaJetty);
+ 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(passiveReplicaJetty);
+ waitForState("Replica not added", collectionName, activeReplicaCount(0, 2, 0));
+ waitForNumDocsInAllActiveReplicas(2);
+ }
+
+ public void testSearchWhileReplicationHappens() {
+
+ }
+
+ public void testReplication() {
+ // Validate incremental replication
+ }
+
+ 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);
+ }
+
+ 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);
+ // Replica recovery
+ new UpdateRequest()
+ .add(sdoc("id", "5"))
+ .process(cloudClient, collectionName);
+ JettySolrRunner solrRunner = getSolrRunner(false).get(0);
+ if (useKill) {
+ ChaosMonkey.kill(solrRunner);
+ } else {
+ ChaosMonkey.stop(solrRunner);
+ }
+ 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
+ waitForNumDocsInAllActiveReplicas(4);
+
+ // If I add the doc immediately, the leader fails to communicate with the follower with broken pipe. Related to SOLR-9555 I believe
+ //nocommit
+ Thread.sleep(10000);
+
+ // More Replica recovery testing
+ new UpdateRequest()
+ .add(sdoc("id", "7"))
+ .process(cloudClient, collectionName);
+ checkRTG(3,7, cluster.getJettySolrRunners());
+ DirectUpdateHandler2.commitOnClose = false;
+ ChaosMonkey.stop(solrRunner);
+ DirectUpdateHandler2.commitOnClose = true;
+ ChaosMonkey.start(solrRunner);
+ waitForState("Replica didn't recover", collectionName, activeReplicaCount(0,2,0));
+ checkRTG(3,7, cluster.getJettySolrRunners());
+ waitForNumDocsInAllActiveReplicas(5, 0);
+
+ // 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());
+ waitForNumDocsInAllActiveReplicas(5, 0);
+ 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();
+ }
+ }
+
+ 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 numRealtimeReplicas, int numAppendReplicas, int numPassiveReplicas) throws SolrServerException, IOException, KeeperException, InterruptedException {
+ CollectionAdminRequest.createCollection(collectionName, "conf", numShards, numRealtimeReplicas, numAppendReplicas, numPassiveReplicas)
+ .setMaxShardsPerNode(100)
+ .process(cluster.getSolrClient());
+ int numReplicasPerShard = numRealtimeReplicas + numAppendReplicas + numPassiveReplicas;
+ cluster.getSolrClient().getZkStateReader().registerCore(collectionName); //TODO: Is this needed?
+ waitForState("Expected collection to be created with " + numShards + " shards and " + numReplicasPerShard + " replicas",
+ collectionName, clusterShape(numShards, numReplicasPerShard));
+ return assertNumberOfReplicas(numRealtimeReplicas*numShards, numAppendReplicas*numShards, numPassiveReplicas*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 " + REPLICATION_TIMEOUT_SECS + " 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 numWriter, int numActive, int numPassive, boolean updateCollection, boolean activeOnly) throws KeeperException, InterruptedException {
+ if (updateCollection) {
+ cluster.getSolrClient().getZkStateReader().forceUpdateCollection(collectionName);
+ }
+ DocCollection docCollection = getCollectionState(collectionName);
+ assertNotNull(docCollection);
+ assertEquals("Unexpected number of writer replicas: " + docCollection, numWriter,
+ docCollection.getReplicas(EnumSet.of(Replica.Type.REALTIME)).stream().filter(r->!activeOnly || r.getState() == Replica.State.ACTIVE).count());
+ assertEquals("Unexpected number of passive replicas: " + docCollection, numPassive,
+ docCollection.getReplicas(EnumSet.of(Replica.Type.PASSIVE)).stream().filter(r->!activeOnly || r.getState() == Replica.State.ACTIVE).count());
+ assertEquals("Unexpected number of active replicas: " + docCollection, numActive,
+ docCollection.getReplicas(EnumSet.of(Replica.Type.APPEND)).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 numWriter, int numActive, int numPassive) {
+ return (liveNodes, collectionState) -> {
+ int writersFound = 0, activesFound = 0, passivesFound = 0;
+ if (collectionState == null)
+ return false;
+ for (Slice slice : collectionState) {
+ for (Replica replica : slice) {
+ if (replica.isActive(liveNodes))
+ switch (replica.getType()) {
+ case APPEND:
+ activesFound++;
+ break;
+ case PASSIVE:
+ passivesFound++;
+ break;
+ case REALTIME:
+ writersFound++;
+ break;
+ default:
+ throw new AssertionError("Unexpected replica type");
+ }
+ }
+ }
+ return numWriter == writersFound && numActive == activesFound && numPassive == passivesFound;
+ };
+ }
+
+ 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/3d149074/solr/core/src/test/org/apache/solr/cloud/TestCloudRecovery.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestCloudRecovery.java b/solr/core/src/test/org/apache/solr/cloud/TestCloudRecovery.java
index b592861..965c169 100644
--- a/solr/core/src/test/org/apache/solr/cloud/TestCloudRecovery.java
+++ b/solr/core/src/test/org/apache/solr/cloud/TestCloudRecovery.java
@@ -66,8 +66,7 @@ public class TestCloudRecovery extends SolrCloudTestCase {
onlyLeaderIndexes = random().nextBoolean();
CollectionAdminRequest
- .createCollection(COLLECTION, "config", 2, 2)
- .setRealtimeReplicas(onlyLeaderIndexes? 1: -1)
+ .createCollection(COLLECTION, "config", 2, onlyLeaderIndexes?0:2,onlyLeaderIndexes?2:0,0)
.setMaxShardsPerNode(2)
.process(cluster.getSolrClient());
AbstractDistribZkTestBase.waitForRecoveriesToFinish(COLLECTION, cluster.getSolrClient().getZkStateReader(),
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/test/org/apache/solr/cloud/TestCollectionAPI.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestCollectionAPI.java b/solr/core/src/test/org/apache/solr/cloud/TestCollectionAPI.java
index 8fbfee3..dd55f23 100644
--- a/solr/core/src/test/org/apache/solr/cloud/TestCollectionAPI.java
+++ b/solr/core/src/test/org/apache/solr/cloud/TestCollectionAPI.java
@@ -61,7 +61,6 @@ public class TestCollectionAPI extends ReplicaPropertiesBase {
public void test() throws Exception {
try (CloudSolrClient client = createCloudClient(null)) {
CollectionAdminRequest.Create req = CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf1",2,2);
- req.setRealtimeReplicas(1);
req.setMaxShardsPerNode(2);
client.request(req);
createCollection(null, COLLECTION_NAME1, 1, 1, 1, client, null, "conf1");
@@ -173,7 +172,7 @@ public class TestCollectionAPI extends ReplicaPropertiesBase {
Map<String, Object> collection = (Map<String, Object>) collections.get(COLLECTION_NAME);
assertNotNull(collection);
assertEquals("conf1", collection.get("configName"));
- assertEquals("1", collection.get("realtimeReplicas"));
+// assertEquals("1", collection.get("realtimeReplicas"));
}
}
[3/3] lucene-solr:jira/solr-10233: Moved patch to (new) branch.
Updated to master
Posted by tf...@apache.org.
Moved patch to (new) branch. Updated to master
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/3d149074
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/3d149074
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/3d149074
Branch: refs/heads/jira/solr-10233
Commit: 3d149074e143ec685a3d079e9acf33bd9e0e6b40
Parents: dd171ff
Author: Tomas Fernandez Lobbe <tf...@apache.org>
Authored: Tue Apr 25 16:37:08 2017 -0700
Committer: Tomas Fernandez Lobbe <tf...@apache.org>
Committed: Tue Apr 25 16:37:08 2017 -0700
----------------------------------------------------------------------
.../org/apache/solr/cloud/AddReplicaCmd.java | 9 +-
.../src/java/org/apache/solr/cloud/Assign.java | 11 +-
.../org/apache/solr/cloud/CloudDescriptor.java | 17 +-
.../apache/solr/cloud/CreateCollectionCmd.java | 38 +-
.../org/apache/solr/cloud/ElectionContext.java | 13 +-
.../java/org/apache/solr/cloud/MigrateCmd.java | 8 +-
.../org/apache/solr/cloud/MoveReplicaCmd.java | 4 +-
.../cloud/OverseerCollectionMessageHandler.java | 89 +-
.../org/apache/solr/cloud/RecoveryStrategy.java | 182 ++++-
.../apache/solr/cloud/ReplicateFromLeader.java | 37 +-
.../org/apache/solr/cloud/SplitShardCmd.java | 2 +-
.../org/apache/solr/cloud/ZkController.java | 50 +-
.../solr/cloud/overseer/ReplicaMutator.java | 7 +-
.../solr/cloud/overseer/SliceMutator.java | 26 +-
.../solr/cloud/overseer/ZkStateWriter.java | 1 +
.../apache/solr/cloud/rule/ReplicaAssigner.java | 6 +-
.../org/apache/solr/core/CoreContainer.java | 37 +-
.../org/apache/solr/handler/IndexFetcher.java | 17 +-
.../solr/handler/admin/CollectionsHandler.java | 15 +-
.../solr/handler/admin/CoreAdminHandler.java | 3 +
.../solr/handler/admin/PrepRecoveryOp.java | 3 +-
.../handler/component/RealTimeGetComponent.java | 5 +
.../solr/update/DefaultSolrCoreState.java | 10 +-
.../solr/update/DirectUpdateHandler2.java | 20 +-
.../org/apache/solr/update/UpdateCommand.java | 1 +
.../org/apache/solr/update/UpdateHandler.java | 6 +-
.../java/org/apache/solr/update/UpdateLog.java | 5 +-
.../processor/DistributedUpdateProcessor.java | 38 +-
solr/core/src/test-files/log4j.properties | 12 +-
.../cloud-minimal/conf/solrconfig.xml | 2 +-
.../solr/cloud/BasicDistributedZk2Test.java | 4 +-
.../solr/cloud/BasicDistributedZkTest.java | 4 +-
.../cloud/ChaosMonkeyNothingIsSafeTest.java | 4 +-
.../cloud/CollectionsAPIDistributedZkTest.java | 4 +-
.../solr/cloud/CollectionsAPISolrJTest.java | 2 +-
.../org/apache/solr/cloud/ForceLeaderTest.java | 4 +-
.../apache/solr/cloud/HttpPartitionTest.java | 4 +-
.../LeaderInitiatedRecoveryOnCommitTest.java | 4 +-
.../solr/cloud/OnlyLeaderIndexesTest.java | 63 +-
...verseerCollectionConfigSetProcessorTest.java | 3 +-
.../org/apache/solr/cloud/OverseerTest.java | 4 +-
.../solr/cloud/RecoveryAfterSoftCommitTest.java | 4 +-
.../org/apache/solr/cloud/ShardSplitTest.java | 11 +-
.../apache/solr/cloud/TestAppendReplica.java | 808 +++++++++++++++++++
.../apache/solr/cloud/TestCloudRecovery.java | 3 +-
.../apache/solr/cloud/TestCollectionAPI.java | 3 +-
.../apache/solr/cloud/TestPassiveReplica.java | 549 +++++++++++++
.../cloud/hdfs/HdfsBasicDistributedZkTest.java | 4 +-
.../reporters/solr/SolrCloudReportersTest.java | 2 +-
.../solr/update/TestInPlaceUpdatesDistrib.java | 4 +-
.../solr/client/solrj/impl/CloudSolrClient.java | 3 +-
.../solrj/request/CollectionAdminRequest.java | 100 ++-
.../apache/solr/common/cloud/DocCollection.java | 23 +-
.../org/apache/solr/common/cloud/Replica.java | 32 +
.../org/apache/solr/common/cloud/Slice.java | 21 +-
.../apache/solr/common/cloud/ZkStateReader.java | 19 +-
.../solr/common/params/CoreAdminParams.java | 5 +
...ollectionAdminRequestRequiredParamsTest.java | 9 +
.../java/org/apache/solr/SolrTestCaseJ4.java | 2 +-
.../solr/cloud/AbstractDistribZkTestBase.java | 3 +
.../cloud/AbstractFullDistribZkTestBase.java | 52 +-
.../java/org/apache/solr/cloud/ChaosMonkey.java | 4 +-
62 files changed, 2143 insertions(+), 292 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java b/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
index 6bb3350..4420a92 100644
--- a/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
@@ -72,6 +72,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
String node = message.getStr(CoreAdminParams.NODE);
String shard = message.getStr(SHARD_ID_PROP);
String coreName = message.getStr(CoreAdminParams.NAME);
+ Replica.Type replicaType = Replica.Type.valueOf(message.getStr(ZkStateReader.REPLICA_TYPE, Replica.Type.REALTIME.name()));
boolean parallel = message.getBool("parallel", false);
if (StringUtils.isBlank(coreName)) {
coreName = message.getStr(CoreAdminParams.PROPERTY_PREFIX + CoreAdminParams.NAME);
@@ -93,7 +94,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
// Kind of unnecessary, but it does put the logic of whether to override maxShardsPerNode in one place.
if (!skipCreateReplicaInClusterState) {
node = getNodesForNewReplicas(clusterState, collection, shard, 1, node,
- ocmh.overseer.getZkController().getCoreContainer()).get(0).nodeName;
+ ocmh.overseer.getZkController().getCoreContainer()).get(0).nodeName;// TODO: use replica type in this logic too
}
log.info("Node Identified {} for creating new replica", node);
@@ -101,7 +102,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Node: " + node + " is not live");
}
if (coreName == null) {
- coreName = Assign.buildCoreName(coll, shard);
+ coreName = Assign.buildCoreName(coll, shard, replicaType);
} else if (!skipCreateReplicaInClusterState) {
//Validate that the core name is unique in that collection
for (Slice slice : coll.getSlices()) {
@@ -126,7 +127,8 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
ZkStateReader.CORE_NAME_PROP, coreName,
ZkStateReader.STATE_PROP, Replica.State.DOWN.toString(),
ZkStateReader.BASE_URL_PROP, zkStateReader.getBaseUrlForNodeName(node),
- ZkStateReader.NODE_NAME_PROP, node);
+ ZkStateReader.NODE_NAME_PROP, node,
+ ZkStateReader.REPLICA_TYPE, replicaType.name());
Overseer.getStateUpdateQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(props));
}
params.set(CoreAdminParams.CORE_NODE_NAME,
@@ -142,6 +144,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
params.set(CoreAdminParams.NAME, coreName);
params.set(COLL_CONF, configName);
params.set(CoreAdminParams.COLLECTION, collection);
+ params.set(CoreAdminParams.REPLICA_TYPE, replicaType.name());
if (shard != null) {
params.set(CoreAdminParams.SHARD, shard);
} else if (routeKey != null) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/cloud/Assign.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/Assign.java b/solr/core/src/java/org/apache/solr/cloud/Assign.java
index ba03ccd..ca784e5 100644
--- a/solr/core/src/java/org/apache/solr/cloud/Assign.java
+++ b/solr/core/src/java/org/apache/solr/cloud/Assign.java
@@ -107,12 +107,16 @@ public class Assign {
returnShardId = shardIdNames.get(0);
return returnShardId;
}
+
+ public static String buildCoreName(String collectionName, String shard, Replica.Type type, int replicaNum) {
+ return collectionName + "_" + shard + "_replica_" + type.name().substring(0,1).toLowerCase() + replicaNum;
+ }
- static String buildCoreName(DocCollection collection, String shard) {
+ public static String buildCoreName(DocCollection collection, String shard, Replica.Type type) {
Slice slice = collection.getSlice(shard);
int replicaNum = slice.getReplicas().size();
for (; ; ) {
- String replicaName = collection.getName() + "_" + shard + "_replica" + replicaNum;
+ String replicaName = buildCoreName(collection.getName(), shard, type, replicaNum);
boolean exists = false;
for (Replica replica : slice.getReplicas()) {
if (replicaName.equals(replica.getStr(CORE_NAME_PROP))) {
@@ -121,9 +125,8 @@ public class Assign {
}
}
if (exists) replicaNum++;
- else break;
+ else return replicaName;
}
- return collection.getName() + "_" + shard + "_replica" + replicaNum;
}
static class ReplicaCount {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/cloud/CloudDescriptor.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/CloudDescriptor.java b/solr/core/src/java/org/apache/solr/cloud/CloudDescriptor.java
index 719b1d1..ff29afc 100644
--- a/solr/core/src/java/org/apache/solr/cloud/CloudDescriptor.java
+++ b/solr/core/src/java/org/apache/solr/cloud/CloudDescriptor.java
@@ -44,6 +44,13 @@ public class CloudDescriptor {
volatile Replica.State lastPublished = Replica.State.ACTIVE;
public static final String NUM_SHARDS = "numShards";
+
+ public static final String REPLICA_TYPE = "replicaType";
+
+ /**
+ * The type of replica this core hosts
+ */
+ private final Replica.Type replicaType;
public CloudDescriptor(String coreName, Properties props, CoreDescriptor cd) {
this.cd = cd;
@@ -57,7 +64,7 @@ public class CloudDescriptor {
if (Strings.isNullOrEmpty(nodeName))
this.nodeName = null;
this.numShards = PropertiesUtil.toInteger(props.getProperty(CloudDescriptor.NUM_SHARDS), null);
-
+ this.replicaType = Replica.Type.valueOf(props.getProperty(CloudDescriptor.REPLICA_TYPE, Replica.Type.REALTIME.name()));
for (String propName : props.stringPropertyNames()) {
if (propName.startsWith(ZkController.COLLECTION_PARAM_PREFIX)) {
collectionParams.put(propName.substring(ZkController.COLLECTION_PARAM_PREFIX.length()), props.getProperty(propName));
@@ -65,6 +72,10 @@ public class CloudDescriptor {
}
}
+ public boolean requiresTransactionLog() {
+ return this.replicaType != Replica.Type.PASSIVE;
+ }
+
public Replica.State getLastPublished() {
return lastPublished;
}
@@ -155,4 +166,8 @@ public class CloudDescriptor {
collectionParams.put(ent.getKey(), ent.getValue());
}
}
+
+ public Replica.Type getReplicaType() {
+ return replicaType;
+ }
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java b/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
index a1bb70e..41b8420 100644
--- a/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
@@ -60,8 +60,7 @@ import static org.apache.solr.cloud.OverseerCollectionMessageHandler.COLL_CONF;
import static org.apache.solr.cloud.OverseerCollectionMessageHandler.CREATE_NODE_SET;
import static org.apache.solr.cloud.OverseerCollectionMessageHandler.NUM_SLICES;
import static org.apache.solr.cloud.OverseerCollectionMessageHandler.RANDOM;
-import static org.apache.solr.common.cloud.ZkStateReader.MAX_SHARDS_PER_NODE;
-import static org.apache.solr.common.cloud.ZkStateReader.REPLICATION_FACTOR;
+import static org.apache.solr.common.cloud.ZkStateReader.*;
import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDREPLICA;
import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
import static org.apache.solr.common.params.CommonParams.NAME;
@@ -96,7 +95,9 @@ public class CreateCollectionCmd implements Cmd {
// look at the replication factor and see if it matches reality
// if it does not, find best nodes to create more cores
- int repFactor = message.getInt(REPLICATION_FACTOR, 1);
+ int numRealtimeReplicas = message.getInt(REALTIME_REPLICAS, message.getInt(REPLICATION_FACTOR, 1));
+ int numPassiveReplicas = message.getInt(PASSIVE_REPLICAS, 0);
+ int numAppendReplicas = message.getInt(APPEND_REPLICAS, 0);
ShardHandler shardHandler = ocmh.shardHandlerFactory.getShardHandler();
final String async = message.getStr(ASYNC);
@@ -116,8 +117,8 @@ public class CreateCollectionCmd implements Cmd {
int maxShardsPerNode = message.getInt(MAX_SHARDS_PER_NODE, 1);
- if (repFactor <= 0) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, REPLICATION_FACTOR + " must be greater than 0");
+ if (numRealtimeReplicas + numAppendReplicas <= 0) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, REALTIME_REPLICAS + " + " + APPEND_REPLICAS + " must be greater than 0");
}
if (numSlices <= 0) {
@@ -135,11 +136,11 @@ public class CreateCollectionCmd implements Cmd {
positionVsNodes = new HashMap<>();
} else {
- if (repFactor > nodeList.size()) {
+ if (numRealtimeReplicas > nodeList.size()) {
log.warn("Specified "
- + REPLICATION_FACTOR
+ + REALTIME_REPLICAS
+ " of "
- + repFactor
+ + numRealtimeReplicas
+ " on collection "
+ collectionName
+ " is higher than or equal to the number of Solr instances currently live or live and part of your " + CREATE_NODE_SET + "("
@@ -148,19 +149,21 @@ public class CreateCollectionCmd implements Cmd {
}
int maxShardsAllowedToCreate = maxShardsPerNode * nodeList.size();
- int requestedShardsToCreate = numSlices * repFactor;
+ int requestedShardsToCreate = numSlices * (numRealtimeReplicas + numPassiveReplicas + numAppendReplicas);
if (maxShardsAllowedToCreate < requestedShardsToCreate) {
throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Cannot create collection " + collectionName + ". Value of "
+ MAX_SHARDS_PER_NODE + " is " + maxShardsPerNode
+ ", and the number of nodes currently live or live and part of your "+CREATE_NODE_SET+" is " + nodeList.size()
+ ". This allows a maximum of " + maxShardsAllowedToCreate
+ " to be created. Value of " + NUM_SLICES + " is " + numSlices
- + " and value of " + REPLICATION_FACTOR + " is " + repFactor
+ + ", value of " + REALTIME_REPLICAS + " is " + numRealtimeReplicas
+ + ", value of " + APPEND_REPLICAS + " is " + numAppendReplicas
+ + " and value of " + PASSIVE_REPLICAS + " is " + numPassiveReplicas
+ ". This requires " + requestedShardsToCreate
+ " shards to be created (higher than the allowed number)");
}
- positionVsNodes = ocmh.identifyNodes(clusterState, nodeList, message, shardNames, repFactor);
+ positionVsNodes = ocmh.identifyNodes(clusterState, nodeList, message, shardNames, numRealtimeReplicas, numAppendReplicas, numPassiveReplicas);
}
ZkStateReader zkStateReader = ocmh.zkStateReader;
@@ -200,13 +203,15 @@ public class CreateCollectionCmd implements Cmd {
Map<String, String> requestMap = new HashMap<>();
- log.debug(formatString("Creating SolrCores for new collection {0}, shardNames {1} , replicationFactor : {2}",
- collectionName, shardNames, repFactor));
+ log.debug(formatString("Creating SolrCores for new collection {0}, shardNames {1} , realtimeReplicas : {2}, appendReplicas: {3}, passiveReplicas: {4}",
+ collectionName, shardNames, numRealtimeReplicas, numAppendReplicas, numPassiveReplicas));
Map<String,ShardRequest> coresToCreate = new LinkedHashMap<>();
for (Map.Entry<ReplicaAssigner.Position, String> e : positionVsNodes.entrySet()) {
ReplicaAssigner.Position position = e.getKey();
String nodeName = e.getValue();
- String coreName = collectionName + "_" + position.shard + "_replica" + (position.index + 1);
+ // TODO: Adding the suffix is great for debugging, but may be an issue if at some point we want to support a way to change replica type
+// String coreName = collectionName + "_" + position.shard + "_replica" + position.suffix + (position.index + 1);
+ String coreName = Assign.buildCoreName(collectionName, position.shard, position.type, position.index + 1);
log.debug(formatString("Creating core {0} as part of shard {1} of collection {2} on {3}"
, coreName, position.shard, collectionName, nodeName));
@@ -221,7 +226,8 @@ public class CreateCollectionCmd implements Cmd {
ZkStateReader.SHARD_ID_PROP, position.shard,
ZkStateReader.CORE_NAME_PROP, coreName,
ZkStateReader.STATE_PROP, Replica.State.DOWN.toString(),
- ZkStateReader.BASE_URL_PROP, baseUrl);
+ ZkStateReader.BASE_URL_PROP, baseUrl,
+ ZkStateReader.REPLICA_TYPE, position.type.name());
Overseer.getStateUpdateQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(props));
}
@@ -235,6 +241,8 @@ public class CreateCollectionCmd implements Cmd {
params.set(CoreAdminParams.SHARD, position.shard);
params.set(ZkStateReader.NUM_SHARDS_PROP, numSlices);
params.set(CoreAdminParams.NEW_COLLECTION, "true");
+ // This is used to tell the CoreAdminHandler that the new core doesn't need a tlog in case of passive replicas
+ params.set(CoreAdminParams.REPLICA_TYPE, position.type.name());
if (async != null) {
String coreAdminAsyncId = async + Math.abs(System.nanoTime());
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java b/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java
index bdbeca9..21549bc 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java
@@ -20,6 +20,7 @@ import java.io.Closeable;
import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
+import java.util.EnumSet;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
@@ -326,6 +327,8 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
return;
}
+ Replica.Type replicaType;
+
try (SolrCore core = cc.getCore(coreName)) {
if (core == null) {
@@ -338,6 +341,8 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
}
}
+ replicaType = core.getCoreDescriptor().getCloudDescriptor().getReplicaType();
+
// should I be leader?
if (weAreReplacement && !shouldIBeLeader(leaderProps, core, weAreReplacement)) {
rejoinLeaderElection(core);
@@ -423,9 +428,7 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
try {
// we must check LIR before registering as leader
checkLIR(coreName, allReplicasInLine);
-
- boolean onlyLeaderIndexes = zkController.getClusterState().getCollection(collection).getRealtimeReplicas() == 1;
- if (onlyLeaderIndexes) {
+ if (replicaType == Replica.Type.APPEND) {
// stop replicate from old leader
zkController.stopReplicationFromLeader(coreName);
if (weAreReplacement) {
@@ -621,7 +624,7 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
}
// on startup and after connection timeout, wait for all known shards
- if (found >= slices.getReplicasMap().size()) {
+ if (found >= slices.getReplicas(EnumSet.of(Replica.Type.APPEND, Replica.Type.REALTIME)).size()) {
log.info("Enough replicas found to continue.");
return true;
} else {
@@ -629,7 +632,7 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
log.info("Waiting until we see more replicas up for shard {}: total={}"
+ " found={}"
+ " timeoutin={}ms",
- shardId, slices.getReplicasMap().size(), found,
+ shardId, slices.getReplicas(EnumSet.of(Replica.Type.APPEND, Replica.Type.REALTIME)).size(), found,
TimeUnit.MILLISECONDS.convert(timeoutAt - System.nanoTime(), TimeUnit.NANOSECONDS));
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/cloud/MigrateCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/MigrateCmd.java b/solr/core/src/java/org/apache/solr/cloud/MigrateCmd.java
index 7b1ad2c..a1a41bb 100644
--- a/solr/core/src/java/org/apache/solr/cloud/MigrateCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/MigrateCmd.java
@@ -51,7 +51,7 @@ import static org.apache.solr.cloud.OverseerCollectionMessageHandler.COLL_PROP_P
import static org.apache.solr.cloud.OverseerCollectionMessageHandler.CREATE_NODE_SET;
import static org.apache.solr.cloud.OverseerCollectionMessageHandler.NUM_SLICES;
import static org.apache.solr.common.cloud.ZkStateReader.COLLECTION_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.REPLICATION_FACTOR;
+import static org.apache.solr.common.cloud.ZkStateReader.REALTIME_REPLICAS;
import static org.apache.solr.common.cloud.ZkStateReader.SHARD_ID_PROP;
import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDREPLICA;
import static org.apache.solr.common.params.CollectionParams.CollectionAction.CREATE;
@@ -208,7 +208,7 @@ public class MigrateCmd implements OverseerCollectionMessageHandler.Cmd {
Map<String, Object> props = makeMap(
Overseer.QUEUE_OPERATION, CREATE.toLower(),
NAME, tempSourceCollectionName,
- REPLICATION_FACTOR, 1,
+ REALTIME_REPLICAS, 1,
NUM_SLICES, 1,
COLL_CONF, configName,
CREATE_NODE_SET, sourceLeader.getNodeName());
@@ -224,7 +224,7 @@ public class MigrateCmd implements OverseerCollectionMessageHandler.Cmd {
Slice tempSourceSlice = clusterState.getCollection(tempSourceCollectionName).getSlices().iterator().next();
Replica tempSourceLeader = zkStateReader.getLeaderRetry(tempSourceCollectionName, tempSourceSlice.getName(), 120000);
- String tempCollectionReplica1 = tempSourceCollectionName + "_" + tempSourceSlice.getName() + "_replica1";
+ String tempCollectionReplica1 = Assign.buildCoreName(tempSourceCollectionName, tempSourceSlice.getName(), Replica.Type.REALTIME, 1);
String coreNodeName = ocmh.waitForCoreNodeName(tempSourceCollectionName,
sourceLeader.getNodeName(), tempCollectionReplica1);
// wait for the replicas to be seen as active on temp source leader
@@ -257,7 +257,7 @@ public class MigrateCmd implements OverseerCollectionMessageHandler.Cmd {
log.info("Creating a replica of temporary collection: {} on the target leader node: {}",
tempSourceCollectionName, targetLeader.getNodeName());
- String tempCollectionReplica2 = tempSourceCollectionName + "_" + tempSourceSlice.getName() + "_replica2";
+ String tempCollectionReplica2 = Assign.buildCoreName(tempSourceCollectionName, tempSourceSlice.getName(), Replica.Type.REALTIME, 2);
props = new HashMap<>();
props.put(Overseer.QUEUE_OPERATION, ADDREPLICA.toLower());
props.put(COLLECTION_PROP, tempSourceCollectionName);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/cloud/MoveReplicaCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/MoveReplicaCmd.java b/solr/core/src/java/org/apache/solr/cloud/MoveReplicaCmd.java
index 545989e..fed1398 100644
--- a/solr/core/src/java/org/apache/solr/cloud/MoveReplicaCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/MoveReplicaCmd.java
@@ -111,7 +111,7 @@ public class MoveReplicaCmd implements Cmd{
private void moveHdfsReplica(ClusterState clusterState, NamedList results, String dataDir, String targetNode, String async,
DocCollection coll, Replica replica, Slice slice) throws Exception {
- String newCoreName = Assign.buildCoreName(coll, slice.getName());
+ String newCoreName = Assign.buildCoreName(coll, slice.getName(), replica.getType());
ZkNodeProps removeReplicasProps = new ZkNodeProps(
COLLECTION_PROP, coll.getName(),
@@ -155,7 +155,7 @@ public class MoveReplicaCmd implements Cmd{
private void moveNormalReplica(ClusterState clusterState, NamedList results, String targetNode, String async,
DocCollection coll, Replica replica, Slice slice) throws Exception {
- String newCoreName = Assign.buildCoreName(coll, slice.getName());
+ String newCoreName = Assign.buildCoreName(coll, slice.getName(), replica.getType());
ZkNodeProps addReplicasProps = new ZkNodeProps(
COLLECTION_PROP, coll.getName(),
SHARD_ID_PROP, slice.getName(),
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java b/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
index 2c08305..4cae785 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
@@ -16,6 +16,51 @@
*/
package org.apache.solr.cloud;
+import static org.apache.solr.common.cloud.DocCollection.SNITCH;
+import static org.apache.solr.common.cloud.ZkStateReader.BASE_URL_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.COLLECTION_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.CORE_NAME_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.CORE_NODE_NAME_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.ELECTION_NODE_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.PROPERTY_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.PROPERTY_VALUE_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.REJOIN_AT_HEAD_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.REPLICA_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.SHARD_ID_PROP;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDREPLICA;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDREPLICAPROP;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDROLE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.BACKUP;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.BALANCESHARDUNIQUE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.CREATE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.CREATEALIAS;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.CREATESHARD;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.CREATESNAPSHOT;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETEALIAS;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETENODE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETEREPLICA;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETEREPLICAPROP;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETESHARD;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.DELETESNAPSHOT;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.MIGRATE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.MIGRATESTATEFORMAT;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.MOCK_COLL_TASK;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.MOCK_REPLICA_TASK;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.MOCK_SHARD_TASK;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.MOVEREPLICA;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.MODIFYCOLLECTION;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.OVERSEERSTATUS;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.REBALANCELEADERS;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.RELOAD;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.REMOVEROLE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.REPLACENODE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.RESTORE;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.SPLITSHARD;
+import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
+import static org.apache.solr.common.params.CommonParams.NAME;
+import static org.apache.solr.common.util.Utils.makeMap;
+
import java.io.Closeable;
import java.io.IOException;
import java.lang.invoke.MethodHandles;
@@ -33,7 +78,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
-import com.google.common.collect.ImmutableMap;
import org.apache.commons.lang.StringUtils;
import org.apache.solr.client.solrj.SolrResponse;
import org.apache.solr.client.solrj.SolrServerException;
@@ -79,21 +123,7 @@ import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.solr.common.cloud.DocCollection.SNITCH;
-import static org.apache.solr.common.cloud.ZkStateReader.BASE_URL_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.COLLECTION_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.CORE_NAME_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.CORE_NODE_NAME_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.ELECTION_NODE_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.PROPERTY_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.PROPERTY_VALUE_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.REJOIN_AT_HEAD_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.REPLICA_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.SHARD_ID_PROP;
-import static org.apache.solr.common.params.CollectionParams.CollectionAction.*;
-import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
-import static org.apache.solr.common.params.CommonParams.NAME;
-import static org.apache.solr.common.util.Utils.makeMap;
+import com.google.common.collect.ImmutableMap;
/**
* A {@link OverseerMessageHandler} that handles Collections API related
@@ -126,12 +156,12 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler
static final String SKIP_CREATE_REPLICA_IN_CLUSTER_STATE = "skipCreateReplicaInClusterState";
+ //nocommit: review
public static final Map<String, Object> COLL_PROPS = Collections.unmodifiableMap(makeMap(
ROUTER, DocRouter.DEFAULT_NAME,
ZkStateReader.REPLICATION_FACTOR, "1",
ZkStateReader.MAX_SHARDS_PER_NODE, "1",
ZkStateReader.AUTO_ADD_REPLICAS, "false",
- ZkStateReader.REALTIME_REPLICAS, "-1",
DocCollection.RULE, null,
SNITCH, null));
@@ -700,18 +730,33 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler
List<String> nodeList,
ZkNodeProps message,
List<String> shardNames,
- int repFactor) throws IOException {
+ int numRealtimeReplicas,
+ int numAppendReplicas,
+ int numPassiveReplicas) throws IOException {
List<Map> rulesMap = (List) message.get("rule");
if (rulesMap == null) {
int i = 0;
Map<Position, String> result = new HashMap<>();
for (String aShard : shardNames) {
- for (int j = 0; j < repFactor; j++){
- result.put(new Position(aShard, j), nodeList.get(i % nodeList.size()));
+ for (int j = 0; j < numRealtimeReplicas; j++){
+ result.put(new Position(aShard, j, Replica.Type.REALTIME), nodeList.get(i % nodeList.size()));
+ i++;
+ }
+ for (int j = 0; j < numAppendReplicas; j++){
+ result.put(new Position(aShard, j, Replica.Type.APPEND), nodeList.get(i % nodeList.size()));
+ i++;
+ }
+ for (int j = 0; j < numPassiveReplicas; j++){
+ result.put(new Position(aShard, j, Replica.Type.PASSIVE), nodeList.get(i % nodeList.size()));
i++;
}
}
return result;
+ } else {
+ if (numAppendReplicas + numPassiveReplicas != 0) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST,
+ Replica.Type.APPEND + " or " + Replica.Type.PASSIVE + " replica types not supported with placement rules");
+ }
}
List<Rule> rules = new ArrayList<>();
@@ -719,7 +764,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler
Map<String, Integer> sharVsReplicaCount = new HashMap<>();
- for (String shard : shardNames) sharVsReplicaCount.put(shard, repFactor);
+ for (String shard : shardNames) sharVsReplicaCount.put(shard, numRealtimeReplicas);
ReplicaAssigner replicaAssigner = new ReplicaAssigner(rules,
sharVsReplicaCount,
(List<Map>) message.get(SNITCH),
@@ -750,6 +795,8 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler
if (result.size() == coreNames.size()) {
return result;
+ } else {
+ log.debug("Expecting {} cores but found {}", coreNames.size(), result.size());
}
if (timeout.hasTimedOut()) {
throw new SolrException(ErrorCode.SERVER_ERROR, "Timed out waiting to see all replicas: " + coreNames + " in cluster state.");
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java b/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
index 496d082..5a952ea 100644
--- a/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
+++ b/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
@@ -74,7 +74,7 @@ import org.slf4j.LoggerFactory;
* between versions in terms of API or back compat behaviour.
* @lucene.experimental
*/
-public class RecoveryStrategy extends Thread implements Closeable {
+public class RecoveryStrategy implements Runnable, Closeable {
public static class Builder implements NamedListInitializedPlugin {
private NamedList args;
@@ -118,19 +118,17 @@ public class RecoveryStrategy extends Thread implements Closeable {
private boolean recoveringAfterStartup;
private CoreContainer cc;
private volatile HttpUriRequest prevSendPreRecoveryHttpUriRequest;
- private boolean onlyLeaderIndexes;
+ private final Replica.Type replicaType;
protected RecoveryStrategy(CoreContainer cc, CoreDescriptor cd, RecoveryListener recoveryListener) {
this.cc = cc;
this.coreName = cd.getName();
this.recoveryListener = recoveryListener;
- setName("RecoveryThread-"+this.coreName);
zkController = cc.getZkController();
zkStateReader = zkController.getZkStateReader();
baseUrl = zkController.getBaseUrl();
coreZkNodeName = cd.getCloudDescriptor().getCoreNodeName();
- String collection = cd.getCloudDescriptor().getCollectionName();
- onlyLeaderIndexes = zkStateReader.getClusterState().getCollection(collection).getRealtimeReplicas() == 1;
+ replicaType = cd.getCloudDescriptor().getReplicaType();
}
final public int getWaitForUpdatesWithStaleStatePauseMilliSeconds() {
@@ -263,7 +261,8 @@ public class RecoveryStrategy extends Thread implements Closeable {
UpdateRequest ureq = new UpdateRequest();
ureq.setParams(new ModifiableSolrParams());
ureq.getParams().set(DistributedUpdateProcessor.COMMIT_END_POINT, true);
- ureq.getParams().set(UpdateParams.OPEN_SEARCHER, onlyLeaderIndexes);
+// ureq.getParams().set(UpdateParams.OPEN_SEARCHER, onlyLeaderIndexes);// nocommit: Why do we need to open searcher if "onlyLeaderIndexes"?
+ ureq.getParams().set(UpdateParams.OPEN_SEARCHER, true);
ureq.setAction(AbstractUpdateRequest.ACTION.COMMIT, false, true).process(
client);
}
@@ -297,9 +296,162 @@ public class RecoveryStrategy extends Thread implements Closeable {
MDCLoggingContext.clear();
}
}
+
+ final public void doRecovery(SolrCore core) throws KeeperException, InterruptedException {
+ if (core.getCoreDescriptor().getCloudDescriptor().requiresTransactionLog()) {
+ doSyncOrReplicateRecovery(core);
+ } else {
+ doReplicateOnlyRecovery(core);
+ }
+ }
+
+ final private void doReplicateOnlyRecovery(SolrCore core) throws InterruptedException {
+ boolean successfulRecovery = false;
+
+// if (core.getUpdateHandler().getUpdateLog() != null) {
+// SolrException.log(LOG, "'replicate-only' recovery strategy should only be used if no update logs are present, but this core has one: "
+// + core.getUpdateHandler().getUpdateLog());
+// return;
+// }
+ while (!successfulRecovery && !Thread.currentThread().isInterrupted() && !isClosed()) { // don't use interruption or it will close channels though
+ try {
+ CloudDescriptor cloudDesc = core.getCoreDescriptor().getCloudDescriptor();
+ ZkNodeProps leaderprops = zkStateReader.getLeaderRetry(
+ cloudDesc.getCollectionName(), cloudDesc.getShardId());
+ final String leaderBaseUrl = leaderprops.getStr(ZkStateReader.BASE_URL_PROP);
+ final String leaderCoreName = leaderprops.getStr(ZkStateReader.CORE_NAME_PROP);
+
+ String leaderUrl = ZkCoreNodeProps.getCoreUrl(leaderBaseUrl, leaderCoreName);
+
+ String ourUrl = ZkCoreNodeProps.getCoreUrl(baseUrl, coreName);
+
+ boolean isLeader = leaderUrl.equals(ourUrl); //TODO: We can probably delete most of this code if we say this strategy can only be used for passive replicas
+ if (isLeader && !cloudDesc.isLeader()) {
+ throw new SolrException(ErrorCode.SERVER_ERROR, "Cloud state still says we are leader.");
+ }
+ if (cloudDesc.isLeader()) {
+ // we are now the leader - no one else must have been suitable
+ LOG.warn("We have not yet recovered - but we are now the leader!");
+ LOG.info("Finished recovery process.");
+ zkController.publish(core.getCoreDescriptor(), Replica.State.ACTIVE);
+ return;
+ }
+
+
+ LOG.info("Publishing state of core [{}] as recovering, leader is [{}] and I am [{}]", core.getName(), leaderUrl,
+ ourUrl);
+ zkController.publish(core.getCoreDescriptor(), Replica.State.RECOVERING);
+
+ if (isClosed()) {
+ LOG.info("Recovery for core {} has been closed", core.getName());
+ break;
+ }
+
+ if (isClosed()) {
+ LOG.info("Recovery for core {} has been closed", core.getName());
+ break;
+ }
+ LOG.info("Starting Replication Recovery.");
+
+ try {
+ LOG.info("Stopping background replicate from leader process");
+ zkController.stopReplicationFromLeader(coreName);
+ replicate(zkController.getNodeName(), core, leaderprops);
+
+ if (isClosed()) {
+ LOG.info("Recovery for core {} has been closed", core.getName());
+ break;
+ }
+
+ LOG.info("Replication Recovery was successful.");
+ successfulRecovery = true;
+ } catch (Exception e) {
+ SolrException.log(LOG, "Error while trying to recover", e);
+ }
+
+ } catch (Exception e) {
+ SolrException.log(LOG, "Error while trying to recover. core=" + coreName, e);
+ } finally {
+ if (successfulRecovery) {
+ LOG.info("Restaring background replicate from leader process");
+ zkController.startReplicationFromLeader(coreName, false);
+ LOG.info("Registering as Active after recovery.");
+ try {
+ zkController.publish(core.getCoreDescriptor(), Replica.State.ACTIVE);
+ } catch (Exception e) {
+ LOG.error("Could not publish as ACTIVE after succesful recovery", e);
+ successfulRecovery = false;
+ }
+
+ if (successfulRecovery) {
+ close = true;
+ recoveryListener.recovered();
+ }
+ }
+ }
+
+ if (!successfulRecovery) {
+ // lets pause for a moment and we need to try again...
+ // TODO: we don't want to retry for some problems?
+ // Or do a fall off retry...
+ try {
+
+ if (isClosed()) {
+ LOG.info("Recovery for core {} has been closed", core.getName());
+ break;
+ }
+
+ LOG.error("Recovery failed - trying again... (" + retries + ")");
+
+ retries++;
+ if (retries >= maxRetries) {
+ SolrException.log(LOG, "Recovery failed - max retries exceeded (" + retries + ").");
+ try {
+ recoveryFailed(core, zkController, baseUrl, coreZkNodeName, core.getCoreDescriptor());
+ } catch (Exception e) {
+ SolrException.log(LOG, "Could not publish that recovery failed", e);
+ }
+ break;
+ }
+ } catch (Exception e) {
+ SolrException.log(LOG, "An error has occurred during recovery", e);
+ }
+
+ try {
+ // Wait an exponential interval between retries, start at 5 seconds and work up to a minute.
+ // If we're at attempt >= 4, there's no point computing pow(2, retries) because the result
+ // will always be the minimum of the two (12). Since we sleep at 5 seconds sub-intervals in
+ // order to check if we were closed, 12 is chosen as the maximum loopCount (5s * 12 = 1m).
+ double loopCount = retries < 4 ? Math.min(Math.pow(2, retries), 12) : 12;
+ LOG.info("Wait [{}] seconds before trying to recover again (attempt={})", loopCount, retries);
+ for (int i = 0; i < loopCount; i++) {
+ if (isClosed()) {
+ LOG.info("Recovery for core {} has been closed", core.getName());
+ break; // check if someone closed us
+ }
+ Thread.sleep(startingRecoveryDelayMilliSeconds);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.warn("Recovery was interrupted.", e);
+ close = true;
+ }
+ }
+
+ }
+
+ // if replay was skipped (possibly to due pulling a full index from the leader),
+ // then we still need to update version bucket seeds after recovery
+ if (successfulRecovery) {
+ LOG.info("Updating version bucket highest from index after successful recovery.");
+ core.seedVersionBuckets();
+ }
+
+ LOG.info("Finished recovery process, successful=[{}]", Boolean.toString(successfulRecovery));
+}
// TODO: perhaps make this grab a new core each time through the loop to handle core reloads?
- final public void doRecovery(SolrCore core) throws KeeperException, InterruptedException {
+ final public void doSyncOrReplicateRecovery(SolrCore core) throws KeeperException, InterruptedException {
boolean replayed = false;
boolean successfulRecovery = false;
@@ -311,9 +463,9 @@ public class RecoveryStrategy extends Thread implements Closeable {
core.getCoreDescriptor());
return;
}
-
- // we temporary ignore peersync for realtimeReplicas mode
- boolean firstTime = !onlyLeaderIndexes;
+
+ // we temporary ignore peersync for Append replicas
+ boolean firstTime = replicaType != Replica.Type.APPEND;
List<Long> recentVersions;
try (UpdateLog.RecentUpdates recentUpdates = ulog.getRecentUpdates()) {
@@ -365,12 +517,12 @@ public class RecoveryStrategy extends Thread implements Closeable {
}
}
- if (onlyLeaderIndexes) {
+ if (replicaType == Replica.Type.APPEND) {
zkController.stopReplicationFromLeader(coreName);
}
Future<RecoveryInfo> replayFuture = null;
- while (!successfulRecovery && !isInterrupted() && !isClosed()) { // don't use interruption or it will close channels though
+ while (!successfulRecovery && !Thread.currentThread().isInterrupted() && !isClosed()) { // don't use interruption or it will close channels though
try {
CloudDescriptor cloudDesc = core.getCoreDescriptor().getCloudDescriptor();
ZkNodeProps leaderprops = zkStateReader.getLeaderRetry(
@@ -522,8 +674,8 @@ public class RecoveryStrategy extends Thread implements Closeable {
if (successfulRecovery) {
LOG.info("Registering as Active after recovery.");
try {
- if (onlyLeaderIndexes) {
- zkController.startReplicationFromLeader(coreName);
+ if (replicaType == Replica.Type.APPEND) {
+ zkController.startReplicationFromLeader(coreName, true);
}
zkController.publish(core.getCoreDescriptor(), Replica.State.ACTIVE);
} catch (Exception e) {
@@ -605,7 +757,7 @@ public class RecoveryStrategy extends Thread implements Closeable {
if (testing_beforeReplayBufferingUpdates != null) {
testing_beforeReplayBufferingUpdates.run();
}
- if (onlyLeaderIndexes) {
+ if (replicaType == Replica.Type.APPEND) {
// roll over all updates during buffering to new tlog, make RTG available
SolrQueryRequest req = new LocalSolrQueryRequest(core,
new ModifiableSolrParams());
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/cloud/ReplicateFromLeader.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/ReplicateFromLeader.java b/solr/core/src/java/org/apache/solr/cloud/ReplicateFromLeader.java
index 817b371..b0bca44 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ReplicateFromLeader.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ReplicateFromLeader.java
@@ -49,7 +49,12 @@ public class ReplicateFromLeader {
this.coreName = coreName;
}
- public void startReplication() throws InterruptedException {
+ /**
+ * Start a replication handler thread that will periodically pull indices from the shard leader
+ * @param switchTransactionLog if true, ReplicationHandler will rotate the transaction log once
+ * the replication is done
+ */
+ public void startReplication(boolean switchTransactionLog) throws InterruptedException {
try (SolrCore core = cc.getCore(coreName)) {
if (core == null) {
if (cc.isShutDown()) {
@@ -78,20 +83,22 @@ public class ReplicateFromLeader {
}
replicationProcess = new ReplicationHandler();
- replicationProcess.setPollListener((solrCore, pollSuccess) -> {
- if (pollSuccess) {
- String commitVersion = getCommitVersion(core);
- if (commitVersion == null) return;
- if (Long.parseLong(commitVersion) == lastVersion) return;
- UpdateLog updateLog = solrCore.getUpdateHandler().getUpdateLog();
- SolrQueryRequest req = new LocalSolrQueryRequest(core,
- new ModifiableSolrParams());
- CommitUpdateCommand cuc = new CommitUpdateCommand(req, false);
- cuc.setVersion(Long.parseLong(commitVersion));
- updateLog.copyOverOldUpdates(cuc);
- lastVersion = Long.parseLong(commitVersion);
- }
- });
+ if (switchTransactionLog) {
+ replicationProcess.setPollListener((solrCore, pollSuccess) -> {
+ if (pollSuccess) {
+ String commitVersion = getCommitVersion(core);
+ if (commitVersion == null) return;
+ if (Long.parseLong(commitVersion) == lastVersion) return;
+ UpdateLog updateLog = solrCore.getUpdateHandler().getUpdateLog();
+ SolrQueryRequest req = new LocalSolrQueryRequest(core,
+ new ModifiableSolrParams());
+ CommitUpdateCommand cuc = new CommitUpdateCommand(req, false);
+ cuc.setVersion(Long.parseLong(commitVersion));
+ updateLog.copyOverOldUpdates(cuc);
+ lastVersion = Long.parseLong(commitVersion);
+ }
+ });
+ }
replicationProcess.init(replicationConfig);
replicationProcess.inform(core);
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java b/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
index 5a099e1..837af79 100644
--- a/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
@@ -385,7 +385,7 @@ public class SplitShardCmd implements Cmd {
Map<ReplicaAssigner.Position, String> nodeMap = ocmh.identifyNodes(clusterState,
new ArrayList<>(clusterState.getLiveNodes()),
new ZkNodeProps(collection.getProperties()),
- subSlices, repFactor - 1);
+ subSlices, repFactor - 1, 0, 0);
List<Map<String, Object>> replicas = new ArrayList<>((repFactor - 1) * 2);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/cloud/ZkController.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/ZkController.java b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
index b337bd0..55dd9a0 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkController.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
@@ -62,6 +62,7 @@ import org.apache.solr.common.cloud.DefaultZkCredentialsProvider;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.OnReconnect;
import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.Replica.Type;
import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkACLProvider;
@@ -883,12 +884,17 @@ public class ZkController {
try {
// If we're a preferred leader, insert ourselves at the head of the queue
boolean joinAtHead = false;
- Replica replica = zkStateReader.getClusterState().getReplica(desc.getCloudDescriptor().getCollectionName(),
- coreZkNodeName);
+ Replica replica = zkStateReader.getClusterState().getReplica(collection, coreZkNodeName);
if (replica != null) {
joinAtHead = replica.getBool(SliceMutator.PREFERRED_LEADER_PROP, false);
}
- joinElection(desc, afterExpiration, joinAtHead);
+ //TODO WHy would replica be null?
+ if (replica == null || replica.getType() != Type.PASSIVE) {
+ joinElection(desc, afterExpiration, joinAtHead);
+ } else if (replica.getType() == Type.PASSIVE) {
+ log.debug("Replica {} skipping election because replica is passive", coreZkNodeName);
+ startReplicationFromLeader(coreName, false);
+ }
} catch (InterruptedException e) {
// Restore the interrupted status
Thread.currentThread().interrupt();
@@ -905,6 +911,8 @@ public class ZkController {
String ourUrl = ZkCoreNodeProps.getCoreUrl(baseUrl, coreName);
log.debug("We are " + ourUrl + " and leader is " + leaderUrl);
boolean isLeader = leaderUrl.equals(ourUrl);
+ Replica.Type replicaType = zkStateReader.getClusterState().getCollection(collection).getReplica(coreZkNodeName).getType();
+ assert !(isLeader && replicaType == Type.PASSIVE): "Passive replica became leader!";
try (SolrCore core = cc.getCore(desc.getName())) {
@@ -915,8 +923,7 @@ public class ZkController {
// leader election perhaps?
UpdateLog ulog = core.getUpdateHandler().getUpdateLog();
- boolean onlyLeaderIndexes = zkStateReader.getClusterState().getCollection(collection).getRealtimeReplicas() == 1;
- boolean isReplicaInOnlyLeaderIndexes = onlyLeaderIndexes && !isLeader;
+ boolean isReplicaInOnlyLeaderIndexes = replicaType == Replica.Type.APPEND && !isLeader;
if (isReplicaInOnlyLeaderIndexes) {
String commitVersion = ReplicateFromLeader.getCommitVersion(core);
if (commitVersion != null) {
@@ -944,11 +951,12 @@ public class ZkController {
= checkRecovery(recoverReloadedCores, isLeader, skipRecovery, collection, coreZkNodeName, core, cc, afterExpiration);
if (!didRecovery) {
if (isReplicaInOnlyLeaderIndexes) {
- startReplicationFromLeader(coreName);
+ startReplicationFromLeader(coreName, true);
}
publish(desc, Replica.State.ACTIVE);
}
+
core.getCoreDescriptor().getCloudDescriptor().setHasRegistered(true);
}
@@ -960,14 +968,18 @@ public class ZkController {
}
}
- public void startReplicationFromLeader(String coreName) throws InterruptedException {
+ public void startReplicationFromLeader(String coreName, boolean switchTransactionLog) throws InterruptedException {
+ log.info(coreName + " starting replication from leader");
ReplicateFromLeader replicateFromLeader = new ReplicateFromLeader(cc, coreName);
if (replicateFromLeaders.putIfAbsent(coreName, replicateFromLeader) == null) {
- replicateFromLeader.startReplication();
+ replicateFromLeader.startReplication(switchTransactionLog);
+ } else {
+ log.warn("A replicate from leader instance already exists for core {}", coreName);
}
}
public void stopReplicationFromLeader(String coreName) {
+ log.info(coreName + " stopping replication from leader");
ReplicateFromLeader replicateFromLeader = replicateFromLeaders.remove(coreName);
if (replicateFromLeader != null) {
replicateFromLeader.stopReplication();
@@ -1191,6 +1203,7 @@ public class ZkController {
if (state != Replica.State.DOWN) {
final Replica.State lirState = getLeaderInitiatedRecoveryState(collection, shardId, coreNodeName);
if (lirState != null) {
+ assert cd.getCloudDescriptor().getReplicaType() != Replica.Type.PASSIVE;
if (state == Replica.State.ACTIVE) {
// trying to become active, so leader-initiated state must be recovering
if (lirState == Replica.State.RECOVERING) {
@@ -1272,12 +1285,16 @@ public class ZkController {
assert false : "No collection was specified [" + collection + "]";
return;
}
+ Replica replica = zkStateReader.getClusterState().getReplica(collection, coreNodeName);
+
+ if (replica == null || replica.getType() != Type.PASSIVE) {
+ ElectionContext context = electionContexts.remove(new ContextKey(collection, coreNodeName));
- ElectionContext context = electionContexts.remove(new ContextKey(collection, coreNodeName));
-
- if (context != null) {
- context.cancelElection();
+ if (context != null) {
+ context.cancelElection();
+ }
}
+// //TODO: Do we need to stop replication for type==append?
CloudDescriptor cloudDescriptor = cd.getCloudDescriptor();
zkStateReader.unregisterCore(cloudDescriptor.getCollectionName());
@@ -1360,6 +1377,7 @@ public class ZkController {
final String shardId = zkStateReader.getClusterState().getShardId(cd.getCollectionName(), getNodeName(), cd.getName());
if (shardId != null) {
cd.getCloudDescriptor().setShardId(shardId);
+ log.debug("Shard ID is {} for core {} ", shardId, cd.getName());
return;
}
try {
@@ -2407,11 +2425,9 @@ public class ZkController {
for (Slice slice : slices) {
Collection<Replica> replicas = slice.getReplicas();
- for (Replica replica : replicas) {
- if (replica.getName().equals(
- dcore.getCloudDescriptor().getCoreNodeName())) {
- return true;
- }
+ Replica r = slice.getReplica(dcore.getCloudDescriptor().getCoreNodeName());
+ if (r != null) {
+ return true;
}
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/cloud/overseer/ReplicaMutator.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/overseer/ReplicaMutator.java b/solr/core/src/java/org/apache/solr/cloud/overseer/ReplicaMutator.java
index f03eeeb..c467405 100644
--- a/solr/core/src/java/org/apache/solr/cloud/overseer/ReplicaMutator.java
+++ b/solr/core/src/java/org/apache/solr/cloud/overseer/ReplicaMutator.java
@@ -271,11 +271,12 @@ public class ReplicaMutator {
replicaProps.putAll(message.getProperties());
if (slice != null) {
- Replica oldReplica = slice.getReplicasMap().get(coreNodeName);
+ Replica oldReplica = slice.getReplica(coreNodeName);
if (oldReplica != null) {
if (oldReplica.containsKey(ZkStateReader.LEADER_PROP)) {
replicaProps.put(ZkStateReader.LEADER_PROP, oldReplica.get(ZkStateReader.LEADER_PROP));
}
+ replicaProps.put(ZkStateReader.REPLICA_TYPE, oldReplica.getType().toString());
// Move custom props over.
for (Map.Entry<String, Object> ent : oldReplica.getProperties().entrySet()) {
if (ent.getKey().startsWith(COLL_PROP_PREFIX)) {
@@ -311,6 +312,8 @@ public class ReplicaMutator {
Replica replica = new Replica(coreNodeName, replicaProps);
+
+ log.debug("Will update state for replica: " + replica);
Map<String, Object> sliceProps = null;
Map<String, Replica> replicas;
@@ -328,11 +331,11 @@ public class ReplicaMutator {
sliceProps.put(ZkStateReader.STATE_PROP, shardState);
sliceProps.put(Slice.PARENT, shardParent);
}
-
replicas.put(replica.getName(), replica);
slice = new Slice(sliceName, replicas, sliceProps);
DocCollection newCollection = CollectionMutator.updateSlice(collectionName, collection, slice);
+ log.debug("Collection is now: " + newCollection);
return new ZkWriteCommand(collectionName, newCollection);
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java b/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
index ec2ce2e..b1e9691 100644
--- a/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
+++ b/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
@@ -16,14 +16,16 @@
*/
package org.apache.solr.cloud.overseer;
-import java.lang.invoke.MethodHandles;
+import static org.apache.solr.cloud.OverseerCollectionMessageHandler.COLL_PROP_PREFIX;
+import static org.apache.solr.cloud.overseer.CollectionMutator.checkCollectionKeyExistence;
+import static org.apache.solr.common.util.Utils.makeMap;
+import java.lang.invoke.MethodHandles;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
-import com.google.common.collect.ImmutableSet;
import org.apache.solr.cloud.Assign;
import org.apache.solr.cloud.Overseer;
import org.apache.solr.common.cloud.ClusterState;
@@ -37,9 +39,7 @@ import org.apache.solr.common.cloud.ZkStateReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.solr.cloud.OverseerCollectionMessageHandler.COLL_PROP_PREFIX;
-import static org.apache.solr.cloud.overseer.CollectionMutator.checkCollectionKeyExistence;
-import static org.apache.solr.common.util.Utils.makeMap;
+import com.google.common.collect.ImmutableSet;
public class SliceMutator {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -66,14 +66,16 @@ public class SliceMutator {
log.error("Invalid Collection/Slice {}/{} ", coll, slice);
return ZkStateWriter.NO_OP;
}
-
String coreNodeName = Assign.assignNode(collection);
+// Replica replica = new Replica(coreNodeName,
+ // coreNodeName overlaps?
Replica replica = new Replica(coreNodeName,
makeMap(
ZkStateReader.CORE_NAME_PROP, message.getStr(ZkStateReader.CORE_NAME_PROP),
ZkStateReader.BASE_URL_PROP, message.getStr(ZkStateReader.BASE_URL_PROP),
ZkStateReader.STATE_PROP, message.getStr(ZkStateReader.STATE_PROP),
- ZkStateReader.NODE_NAME_PROP, message.getStr(ZkStateReader.NODE_NAME_PROP)));
+ ZkStateReader.NODE_NAME_PROP, message.getStr(ZkStateReader.NODE_NAME_PROP),
+ ZkStateReader.REPLICA_TYPE, message.get(ZkStateReader.REPLICA_TYPE)));
return new ZkWriteCommand(coll, updateReplica(collection, sl, replica.getName(), replica));
}
@@ -248,13 +250,15 @@ public class SliceMutator {
}
public static DocCollection updateReplica(DocCollection collection, final Slice slice, String coreNodeName, final Replica replica) {
- Map<String, Replica> copy = slice.getReplicasCopy();
+ Map<String, Replica> replicasCopy = slice.getReplicasCopy();
if (replica == null) {
- copy.remove(coreNodeName);
+ replicasCopy.remove(coreNodeName);
} else {
- copy.put(replica.getName(), replica);
+ replicasCopy.put(replica.getName(), replica);
}
- Slice newSlice = new Slice(slice.getName(), copy, slice.getProperties());
+ Slice newSlice = new Slice(slice.getName(), replicasCopy, slice.getProperties());
+ log.info("Old Slice: " + slice);
+ log.info("New Slice: " + newSlice);
return CollectionMutator.updateSlice(collection.getName(), collection, newSlice);
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java b/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java
index 23fb56c..a906b86 100644
--- a/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java
+++ b/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java
@@ -268,6 +268,7 @@ public class ZkStateWriter {
}
}
+ log.debug("New Cluster State is: " + clusterState);
return clusterState;
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/cloud/rule/ReplicaAssigner.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/rule/ReplicaAssigner.java b/solr/core/src/java/org/apache/solr/cloud/rule/ReplicaAssigner.java
index 3eab8b4..ebe29f1 100644
--- a/solr/core/src/java/org/apache/solr/cloud/rule/ReplicaAssigner.java
+++ b/solr/core/src/java/org/apache/solr/cloud/rule/ReplicaAssigner.java
@@ -62,10 +62,12 @@ public class ReplicaAssigner {
public static class Position implements Comparable<Position> {
public final String shard;
public final int index;
+ public final Replica.Type type;
- public Position(String shard, int replicaIdx) {
+ public Position(String shard, int replicaIdx, Replica.Type type) {
this.shard = shard;
this.index = replicaIdx;
+ this.type = type;
}
@Override
@@ -188,7 +190,7 @@ public class ReplicaAssigner {
List<Position> positions = new ArrayList<>();
for (int pos : p) {
for (int j = 0; j < shardVsReplicaCount.get(shardNames.get(pos)); j++) {
- positions.add(new Position(shardNames.get(pos), j));
+ positions.add(new Position(shardNames.get(pos), j, Replica.Type.REALTIME));
}
}
Collections.sort(positions);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/core/CoreContainer.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/core/CoreContainer.java b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
index 28c1eaf..a4f15f7 100644
--- a/solr/core/src/java/org/apache/solr/core/CoreContainer.java
+++ b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
@@ -16,6 +16,17 @@
*/
package org.apache.solr.core;
+import static java.util.Objects.requireNonNull;
+import static org.apache.solr.common.params.CommonParams.AUTHC_PATH;
+import static org.apache.solr.common.params.CommonParams.AUTHZ_PATH;
+import static org.apache.solr.common.params.CommonParams.COLLECTIONS_HANDLER_PATH;
+import static org.apache.solr.common.params.CommonParams.CONFIGSETS_HANDLER_PATH;
+import static org.apache.solr.common.params.CommonParams.CORES_HANDLER_PATH;
+import static org.apache.solr.common.params.CommonParams.INFO_HANDLER_PATH;
+import static org.apache.solr.common.params.CommonParams.METRICS_PATH;
+import static org.apache.solr.common.params.CommonParams.ZK_PATH;
+import static org.apache.solr.security.AuthenticationPlugin.AUTHENTICATION_PLUGIN_PROP;
+
import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.nio.file.Path;
@@ -56,6 +67,7 @@ import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.Replica.State;
+import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.IOUtils;
import org.apache.solr.common.util.Utils;
@@ -95,17 +107,7 @@ import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static java.util.Objects.requireNonNull;
-import static org.apache.solr.common.params.CommonParams.AUTHC_PATH;
-import static org.apache.solr.common.params.CommonParams.AUTHZ_PATH;
-import static org.apache.solr.common.params.CommonParams.COLLECTIONS_HANDLER_PATH;
-import static org.apache.solr.common.params.CommonParams.CONFIGSETS_HANDLER_PATH;
-import static org.apache.solr.common.params.CommonParams.CORES_HANDLER_PATH;
-import static org.apache.solr.common.params.CommonParams.INFO_HANDLER_PATH;
-import static org.apache.solr.common.params.CommonParams.METRICS_PATH;
-import static org.apache.solr.common.params.CommonParams.ZK_PATH;
import static org.apache.solr.core.CorePropertiesLocator.PROPERTIES_FILENAME;
-import static org.apache.solr.security.AuthenticationPlugin.AUTHENTICATION_PLUGIN_PROP;
/**
*
@@ -1184,10 +1186,15 @@ public class CoreContainer {
SolrCore newCore = core.reload(coreConfig);
registerCore(cd, newCore, false, false);
if (getZkController() != null) {
- boolean onlyLeaderIndexes = getZkController().getClusterState().getCollection(cd.getCollectionName()).getRealtimeReplicas() == 1;
- if (onlyLeaderIndexes && !cd.getCloudDescriptor().isLeader()) {
+ DocCollection docCollection = getZkController().getClusterState().getCollection(cd.getCollectionName());
+ Replica replica = docCollection.getReplica(cd.getCloudDescriptor().getCoreNodeName());
+ assert replica != null;
+ if (replica.getType() == Replica.Type.APPEND) { //TODO: needed here?
getZkController().stopReplicationFromLeader(core.getName());
- getZkController().startReplicationFromLeader(newCore.getName());
+ if (!cd.getCloudDescriptor().isLeader()) {
+ getZkController().startReplicationFromLeader(newCore.getName(), true);
+ }
+
}
}
} catch (SolrCoreState.CoreIsClosedException e) {
@@ -1273,6 +1280,10 @@ public class CoreContainer {
if (zkSys.getZkController() != null) {
// cancel recovery in cloud mode
core.getSolrCoreState().cancelRecovery();
+ if (core.getCoreDescriptor().getCloudDescriptor().getReplicaType() == Replica.Type.PASSIVE) { // TODO: Also for Replica.Type.ACTIVE?
+ // Stop replication if this is part of a passive replica before closing the code
+ zkSys.getZkController().stopReplicationFromLeader(name);
+ }
}
core.unloadOnClose(cd, deleteIndexDir, deleteDataDir, deleteInstanceDir);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java b/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
index 96e505a..84b36b5 100644
--- a/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
+++ b/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
@@ -352,17 +352,28 @@ public class IndexFetcher {
// when we are a bit more confident we may want to try a partial replication
// if the error is connection related or something, but we have to be careful
forceReplication = true;
+ LOG.info("Last replication failed, so I'll force replication");
}
try {
if (fetchFromLeader) {
+ assert !solrCore.isClosed(): "Replication should be stopped before closing the core";
Replica replica = getLeaderReplica();
CloudDescriptor cd = solrCore.getCoreDescriptor().getCloudDescriptor();
if (cd.getCoreNodeName().equals(replica.getName())) {
return IndexFetchResult.EXPECTING_NON_LEADER;
}
- masterUrl = replica.getCoreUrl();
- LOG.info("Updated masterUrl to " + masterUrl);
+ if (replica.getState() != Replica.State.ACTIVE) {
+ LOG.info("Replica {} is leader but it's state is {}, skipping replication", replica.getName(), replica.getState());
+ return IndexFetchResult.EXPECTING_NON_LEADER;//nocommit: not the correct error
+ }
+ if (!replica.getCoreUrl().equals(masterUrl)) {
+ masterUrl = replica.getCoreUrl();
+ LOG.info("Updated masterUrl to {}", masterUrl);
+ // TODO: Do we need to set forceReplication = true?
+ } else {
+ LOG.debug("masterUrl didn't change");
+ }
}
//get the current 'replicateable' index version in the master
NamedList response;
@@ -410,6 +421,7 @@ public class IndexFetcher {
if (forceReplication && commit.getGeneration() != 0) {
// since we won't get the files for an empty index,
// we just clear ours and commit
+ LOG.info("New index in Master. Deleting mine...");
RefCounted<IndexWriter> iw = solrCore.getUpdateHandler().getSolrCoreState().getIndexWriter(solrCore);
try {
iw.get().deleteAll();
@@ -422,6 +434,7 @@ public class IndexFetcher {
//there is nothing to be replicated
successfulInstall = true;
+ LOG.debug("Nothing to replicate, master's version is 0");
return IndexFetchResult.MASTER_VERSION_ZERO;
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java b/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
index d5c4927..e657b62 100644
--- a/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
+++ b/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
@@ -113,15 +113,7 @@ import static org.apache.solr.common.cloud.DocCollection.DOC_ROUTER;
import static org.apache.solr.common.cloud.DocCollection.RULE;
import static org.apache.solr.common.cloud.DocCollection.SNITCH;
import static org.apache.solr.common.cloud.DocCollection.STATE_FORMAT;
-import static org.apache.solr.common.cloud.ZkStateReader.AUTO_ADD_REPLICAS;
-import static org.apache.solr.common.cloud.ZkStateReader.COLLECTION_PROP;
-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.PROPERTY_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.PROPERTY_VALUE_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.REPLICATION_FACTOR;
-import static org.apache.solr.common.cloud.ZkStateReader.REPLICA_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.SHARD_ID_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.*;
import static org.apache.solr.common.params.CollectionAdminParams.COUNT_PROP;
import static org.apache.solr.common.params.CollectionParams.CollectionAction.*;
import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
@@ -407,6 +399,8 @@ public class CollectionsHandler extends RequestHandlerBase implements Permission
AUTO_ADD_REPLICAS,
RULE,
SNITCH,
+ PASSIVE_REPLICAS,
+ APPEND_REPLICAS,
REALTIME_REPLICAS);
if (props.get(STATE_FORMAT) == null) {
@@ -634,7 +628,8 @@ public class CollectionsHandler extends RequestHandlerBase implements Permission
_ROUTE_,
CoreAdminParams.NAME,
INSTANCE_DIR,
- DATA_DIR);
+ DATA_DIR,
+ REPLICA_TYPE);
return copyPropertiesWithPrefix(req.getParams(), props, COLL_PROP_PREFIX);
}),
OVERSEERSTATUS_OP(OVERSEERSTATUS, (req, rsp, h) -> (Map) new LinkedHashMap<>()),
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java b/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java
index 6746332..31ad4f7 100644
--- a/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java
+++ b/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java
@@ -171,7 +171,9 @@ public class CoreAdminHandler extends RequestHandlerBase implements PermissionNa
final CallInfo callInfo = new CallInfo(this, req, rsp, op);
if (taskId == null) {
+ log.info("Starting Operation: " + req);
callInfo.call();
+ log.info("Done with Operation: " + req);
} else {
try {
MDC.put("CoreAdminHandler.asyncId", taskId);
@@ -227,6 +229,7 @@ public class CoreAdminHandler extends RequestHandlerBase implements PermissionNa
.put(CoreAdminParams.ROLES, CoreDescriptor.CORE_ROLES)
.put(CoreAdminParams.CORE_NODE_NAME, CoreDescriptor.CORE_NODE_NAME)
.put(ZkStateReader.NUM_SHARDS_PROP, CloudDescriptor.NUM_SHARDS)
+ .put(CoreAdminParams.REPLICA_TYPE, CloudDescriptor.REPLICA_TYPE)
.build();
protected static Map<String, String> buildCoreParams(SolrParams params) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/handler/admin/PrepRecoveryOp.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/admin/PrepRecoveryOp.java b/solr/core/src/java/org/apache/solr/handler/admin/PrepRecoveryOp.java
index 0c2c903..8e5408b 100644
--- a/solr/core/src/java/org/apache/solr/handler/admin/PrepRecoveryOp.java
+++ b/solr/core/src/java/org/apache/solr/handler/admin/PrepRecoveryOp.java
@@ -18,6 +18,7 @@
package org.apache.solr.handler.admin;
import java.lang.invoke.MethodHandles;
+import java.util.Objects;
import org.apache.lucene.search.MatchAllDocsQuery;
import org.apache.solr.cloud.CloudDescriptor;
@@ -175,7 +176,7 @@ class PrepRecoveryOp implements CoreAdminHandler.CoreAdminOp {
"I was asked to wait on state " + waitForState + " for "
+ shardId + " in " + collection + " on " + nodeName
+ " but I still do not see the requested state. I see state: "
- + state.toString() + " live:" + live + " leader from ZK: " + leaderInfo
+ + Objects.toString(state) + " live:" + live + " leader from ZK: " + leaderInfo
);
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java b/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
index c0ceddb..c4ee76b 100644
--- a/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
+++ b/solr/core/src/java/org/apache/solr/handler/component/RealTimeGetComponent.java
@@ -106,6 +106,11 @@ public class RealTimeGetComponent extends SearchComponent
SolrQueryResponse rsp = rb.rsp;
SolrParams params = req.getParams();
+ if (req.getCore().getCoreDescriptor().getCloudDescriptor() != null
+ && !req.getCore().getCoreDescriptor().getCloudDescriptor().requiresTransactionLog()) {
+ return;
+ }
+
if (!params.getBool(COMPONENT_NAME, true)) {
return;
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java b/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java
index bc2afa8..f19b2df 100644
--- a/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java
+++ b/solr/core/src/java/org/apache/solr/update/DefaultSolrCoreState.java
@@ -16,6 +16,7 @@
*/
package org.apache.solr.update;
+import java.io.Closeable;
import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.concurrent.ExecutionException;
@@ -32,8 +33,8 @@ import org.apache.lucene.index.MergePolicy;
import org.apache.lucene.search.Sort;
import org.apache.solr.cloud.ActionThrottle;
import org.apache.solr.cloud.RecoveryStrategy;
-import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.common.SolrException;
+import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.core.CoreContainer;
import org.apache.solr.core.CoreDescriptor;
import org.apache.solr.core.DirectoryFactory;
@@ -65,7 +66,8 @@ public final class DefaultSolrCoreState extends SolrCoreState implements Recover
private DirectoryFactory directoryFactory;
private final RecoveryStrategy.Builder recoveryStrategyBuilder;
- private volatile RecoveryStrategy recoveryStrat;
+ private volatile RecoveryStrategy recoveryStrat; //nocommit: Make interface
+// private volatile Thread recoveryStrat;
private volatile boolean lastReplicationSuccess = true;
@@ -365,9 +367,11 @@ public final class DefaultSolrCoreState extends SolrCoreState implements Recover
public void cancelRecovery() {
if (recoveryStrat != null) {
try {
- recoveryStrat.close();
+ ((Closeable)recoveryStrat).close();
} catch (NullPointerException e) {
// okay
+ } catch (IOException e) {
+ // okay
}
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/update/DirectUpdateHandler2.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/update/DirectUpdateHandler2.java b/solr/core/src/java/org/apache/solr/update/DirectUpdateHandler2.java
index dd179f2..ec70451 100644
--- a/solr/core/src/java/org/apache/solr/update/DirectUpdateHandler2.java
+++ b/solr/core/src/java/org/apache/solr/update/DirectUpdateHandler2.java
@@ -45,7 +45,7 @@ import org.apache.lucene.search.TermQuery;
import org.apache.lucene.util.BytesRefHash;
import org.apache.solr.cloud.ZkController;
import org.apache.solr.common.SolrException;
-import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.core.SolrConfig.UpdateHandlerInfo;
import org.apache.solr.core.SolrCore;
@@ -122,12 +122,9 @@ public class DirectUpdateHandler2 extends UpdateHandler implements SolrCoreState
indexWriterCloseWaitsForMerges = updateHandlerInfo.indexWriterCloseWaitsForMerges;
ZkController zkController = core.getCoreContainer().getZkController();
- if (zkController != null) {
- DocCollection dc = zkController.getClusterState().getCollection(core.getCoreDescriptor().getCollectionName());
- if (dc.getRealtimeReplicas() == 1) {
- commitWithinSoftCommit = false;
- commitTracker.setOpenSearcher(true);
- }
+ if (zkController != null && core.getCoreDescriptor().getCloudDescriptor().getReplicaType() == Replica.Type.APPEND) {
+ commitWithinSoftCommit = false;
+ commitTracker.setOpenSearcher(true);
}
}
@@ -248,7 +245,7 @@ public class DirectUpdateHandler2 extends UpdateHandler implements SolrCoreState
cmd.overwrite = false;
}
try {
- if ( (cmd.getFlags() & UpdateCommand.IGNORE_INDEXWRITER) != 0) {
+ if ((cmd.getFlags() & UpdateCommand.IGNORE_INDEXWRITER) != 0) {
if (ulog != null) ulog.add(cmd);
return 1;
}
@@ -424,7 +421,7 @@ public class DirectUpdateHandler2 extends UpdateHandler implements SolrCoreState
deleteByIdCommands.increment();
deleteByIdCommandsCumulative.mark();
- if ( (cmd.getFlags() & UpdateCommand.IGNORE_INDEXWRITER) != 0 ) {
+ if ((cmd.getFlags() & UpdateCommand.IGNORE_INDEXWRITER) != 0 ) {
if (ulog != null) ulog.delete(cmd);
return;
}
@@ -488,7 +485,7 @@ public class DirectUpdateHandler2 extends UpdateHandler implements SolrCoreState
deleteByQueryCommandsCumulative.mark();
boolean madeIt=false;
try {
- if ( (cmd.getFlags() & UpdateCommand.IGNORE_INDEXWRITER) != 0) {
+ if ((cmd.getFlags() & UpdateCommand.IGNORE_INDEXWRITER) != 0) {
if (ulog != null) ulog.deleteByQuery(cmd);
madeIt = true;
return;
@@ -547,7 +544,6 @@ public class DirectUpdateHandler2 extends UpdateHandler implements SolrCoreState
}
}
-
@Override
public int mergeIndexes(MergeIndexesCommand cmd) throws IOException {
mergeIndexesCommands.mark();
@@ -920,7 +916,7 @@ public class DirectUpdateHandler2 extends UpdateHandler implements SolrCoreState
* Calls either {@link IndexWriter#updateDocValues} or {@link IndexWriter#updateDocument} as
* needed based on {@link AddUpdateCommand#isInPlaceUpdate}.
* <p>
- * If the this is an UPDATE_INPLACE cmd, then all fields inclued in
+ * If the this is an UPDATE_INPLACE cmd, then all fields included in
* {@link AddUpdateCommand#getLuceneDocument} must either be the uniqueKey field, or be DocValue
* only fields.
* </p>
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/update/UpdateCommand.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/update/UpdateCommand.java b/solr/core/src/java/org/apache/solr/update/UpdateCommand.java
index b124271..6c0fc50 100644
--- a/solr/core/src/java/org/apache/solr/update/UpdateCommand.java
+++ b/solr/core/src/java/org/apache/solr/update/UpdateCommand.java
@@ -34,6 +34,7 @@ public abstract class UpdateCommand implements Cloneable {
public static int PEER_SYNC = 0x00000004; // update command is a missing update being provided by a peer.
public static int IGNORE_AUTOCOMMIT = 0x00000008; // this update should not count toward triggering of autocommits.
public static int CLEAR_CACHES = 0x00000010; // clear caches associated with the update log. used when applying reordered DBQ updates when doing an add.
+ // TODO: rename to something like "APPEND_REPLICAS_IGNORE_IW", or maybe just "FROM_LEADER"?
public static int IGNORE_INDEXWRITER = 0x00000020;
public UpdateCommand(SolrQueryRequest req) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3d149074/solr/core/src/java/org/apache/solr/update/UpdateHandler.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/update/UpdateHandler.java b/solr/core/src/java/org/apache/solr/update/UpdateHandler.java
index 49d2664..5c6e33b 100644
--- a/solr/core/src/java/org/apache/solr/update/UpdateHandler.java
+++ b/solr/core/src/java/org/apache/solr/update/UpdateHandler.java
@@ -115,7 +115,7 @@ public abstract class UpdateHandler implements SolrInfoBean {
public UpdateHandler(SolrCore core) {
this(core, null);
}
-
+
public UpdateHandler(SolrCore core, UpdateLog updateLog) {
this.core=core;
idField = core.getLatestSchema().getUniqueKeyField();
@@ -124,7 +124,9 @@ public abstract class UpdateHandler implements SolrInfoBean {
PluginInfo ulogPluginInfo = core.getSolrConfig().getPluginInfo(UpdateLog.class.getName());
- if (updateLog == null && ulogPluginInfo != null && ulogPluginInfo.isEnabled()) {
+ // If this is a replica of type passive, don't create the update log
+ boolean skipUpdateLog = core.getCoreDescriptor().getCloudDescriptor() != null && !core.getCoreDescriptor().getCloudDescriptor().requiresTransactionLog();
+ if (updateLog == null && ulogPluginInfo != null && !skipUpdateLog) {
String dataDir = (String)ulogPluginInfo.initArgs.get("dir");
String ulogDir = core.getCoreDescriptor().getUlogDir();