You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by va...@apache.org on 2017/11/06 16:13:24 UTC
[1/4] lucene-solr:branch_7x: SOLR-11003: Support bi-directional
syncing of cdcr clusters. We still only support actively into one cluster
cluster,
but have the ability to switch indexing clusters and cdcr will replicate
correctly
Repository: lucene-solr
Updated Branches:
refs/heads/branch_7x 1af62a6ed -> a30c92a79
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrReplicationHandlerTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrReplicationHandlerTest.java b/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrReplicationHandlerTest.java
new file mode 100644
index 0000000..65826c4
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrReplicationHandlerTest.java
@@ -0,0 +1,331 @@
+/*
+ * 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.cdcr;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.lucene.util.LuceneTestCase.Nightly;
+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.impl.CloudSolrClient;
+import org.apache.solr.cloud.ChaosMonkey;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.util.DefaultSolrThreadFactory;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class is testing the cdcr extension to the {@link org.apache.solr.handler.ReplicationHandler} and
+ * {@link org.apache.solr.handler.IndexFetcher}.
+ */
+@Nightly
+public class CdcrReplicationHandlerTest extends BaseCdcrDistributedZkTest {
+
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ @Override
+ public void distribSetUp() throws Exception {
+ schemaString = "schema15.xml"; // we need a string id
+ createTargetCollection = false; // we do not need the target cluster
+ shardCount = 1; // we need only one shard
+ // we need a persistent directory, otherwise the UpdateHandler will erase existing tlog files after restarting a node
+ System.setProperty("solr.directoryFactory", "solr.StandardDirectoryFactory");
+ super.distribSetUp();
+ }
+
+ /**
+ * Test the scenario where the slave is killed from the start. The replication
+ * strategy should fetch all the missing tlog files from the leader.
+ */
+ @Test
+ @ShardsFixed(num = 2)
+ public void testFullReplication() throws Exception {
+ List<CloudJettyRunner> slaves = this.getShardToSlaveJetty(SOURCE_COLLECTION, SHARD1);
+ ChaosMonkey.stop(slaves.get(0).jetty);
+
+ for (int i = 0; i < 10; i++) {
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int j = i * 10; j < (i * 10) + 10; j++) {
+ docs.add(getDoc(id, Integer.toString(j)));
+ }
+ index(SOURCE_COLLECTION, docs);
+ }
+
+ assertNumDocs(100, SOURCE_COLLECTION);
+
+ // Restart the slave node to trigger Replication strategy
+ this.restartServer(slaves.get(0));
+
+ this.assertUpdateLogsEquals(SOURCE_COLLECTION, 10);
+ }
+
+ /**
+ * Test the scenario where the slave is killed before receiving all the documents. The replication
+ * strategy should fetch all the missing tlog files from the leader.
+ */
+ @Test
+ @ShardsFixed(num = 2)
+ public void testPartialReplication() throws Exception {
+ for (int i = 0; i < 5; i++) {
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int j = i * 20; j < (i * 20) + 20; j++) {
+ docs.add(getDoc(id, Integer.toString(j)));
+ }
+ index(SOURCE_COLLECTION, docs);
+ }
+
+ List<CloudJettyRunner> slaves = this.getShardToSlaveJetty(SOURCE_COLLECTION, SHARD1);
+ ChaosMonkey.stop(slaves.get(0).jetty);
+
+ for (int i = 5; i < 10; i++) {
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int j = i * 20; j < (i * 20) + 20; j++) {
+ docs.add(getDoc(id, Integer.toString(j)));
+ }
+ index(SOURCE_COLLECTION, docs);
+ }
+
+ assertNumDocs(200, SOURCE_COLLECTION);
+
+ // Restart the slave node to trigger Replication strategy
+ this.restartServer(slaves.get(0));
+
+ // at this stage, the slave should have replicated the 5 missing tlog files
+ this.assertUpdateLogsEquals(SOURCE_COLLECTION, 10);
+ }
+
+ /**
+ * Test the scenario where the slave is killed before receiving a commit. This creates a truncated tlog
+ * file on the slave node. The replication strategy should detect this truncated file, and fetch the
+ * non-truncated file from the leader.
+ */
+ @Test
+ @ShardsFixed(num = 2)
+ public void testPartialReplicationWithTruncatedTlog() throws Exception {
+ CloudSolrClient client = createCloudClient(SOURCE_COLLECTION);
+ List<CloudJettyRunner> slaves = this.getShardToSlaveJetty(SOURCE_COLLECTION, SHARD1);
+
+ try {
+ for (int i = 0; i < 10; i++) {
+ for (int j = i * 20; j < (i * 20) + 20; j++) {
+ client.add(getDoc(id, Integer.toString(j)));
+
+ // Stop the slave in the middle of a batch to create a truncated tlog on the slave
+ if (j == 45) {
+ ChaosMonkey.stop(slaves.get(0).jetty);
+ }
+
+ }
+ commit(SOURCE_COLLECTION);
+ }
+ } finally {
+ client.close();
+ }
+
+ assertNumDocs(200, SOURCE_COLLECTION);
+
+ // Restart the slave node to trigger Replication recovery
+ this.restartServer(slaves.get(0));
+
+ // at this stage, the slave should have replicated the 5 missing tlog files
+ this.assertUpdateLogsEquals(SOURCE_COLLECTION, 10);
+ }
+
+ /**
+ * Test the scenario where the slave first recovered with a PeerSync strategy, then with a Replication strategy.
+ * The PeerSync strategy will generate a single tlog file for all the missing updates on the slave node.
+ * If a Replication strategy occurs at a later stage, it should remove this tlog file generated by PeerSync
+ * and fetch the corresponding tlog files from the leader.
+ */
+ @Test
+ @ShardsFixed(num = 2)
+ public void testPartialReplicationAfterPeerSync() throws Exception {
+ for (int i = 0; i < 5; i++) {
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int j = i * 10; j < (i * 10) + 10; j++) {
+ docs.add(getDoc(id, Integer.toString(j)));
+ }
+ index(SOURCE_COLLECTION, docs);
+ }
+
+ List<CloudJettyRunner> slaves = this.getShardToSlaveJetty(SOURCE_COLLECTION, SHARD1);
+ ChaosMonkey.stop(slaves.get(0).jetty);
+
+ for (int i = 5; i < 10; i++) {
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int j = i * 10; j < (i * 10) + 10; j++) {
+ docs.add(getDoc(id, Integer.toString(j)));
+ }
+ index(SOURCE_COLLECTION, docs);
+ }
+
+ assertNumDocs(100, SOURCE_COLLECTION);
+
+ // Restart the slave node to trigger PeerSync recovery
+ // (the update windows between leader and slave is small enough)
+ this.restartServer(slaves.get(0));
+
+ ChaosMonkey.stop(slaves.get(0).jetty);
+
+ for (int i = 10; i < 15; i++) {
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int j = i * 20; j < (i * 20) + 20; j++) {
+ docs.add(getDoc(id, Integer.toString(j)));
+ }
+ index(SOURCE_COLLECTION, docs);
+ }
+
+ // restart the slave node to trigger Replication recovery
+ this.restartServer(slaves.get(0));
+
+ // at this stage, the slave should have replicated the 5 missing tlog files
+ this.assertUpdateLogsEquals(SOURCE_COLLECTION, 15);
+ }
+
+ /**
+ * Test the scenario where the slave is killed while the leader is still receiving updates.
+ * The slave should buffer updates while in recovery, then replay them at the end of the recovery.
+ * If updates were properly buffered and replayed, then the slave should have the same number of documents
+ * than the leader. This checks if cdcr tlog replication interferes with buffered updates - SOLR-8263.
+ */
+ @Test
+ @ShardsFixed(num = 2)
+ public void testReplicationWithBufferedUpdates() throws Exception {
+ List<CloudJettyRunner> slaves = this.getShardToSlaveJetty(SOURCE_COLLECTION, SHARD1);
+
+ AtomicInteger numDocs = new AtomicInteger(0);
+ ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(new DefaultSolrThreadFactory("cdcr-test-update-scheduler"));
+ executor.scheduleWithFixedDelay(new UpdateThread(numDocs), 10, 10, TimeUnit.MILLISECONDS);
+
+ // Restart the slave node to trigger Replication strategy
+ this.restartServer(slaves.get(0));
+
+ // shutdown the update thread and wait for its completion
+ executor.shutdown();
+ executor.awaitTermination(500, TimeUnit.MILLISECONDS);
+
+ // check that we have the expected number of documents in the cluster
+ assertNumDocs(numDocs.get(), SOURCE_COLLECTION);
+
+ // check that we have the expected number of documents on the slave
+ assertNumDocs(numDocs.get(), slaves.get(0));
+ }
+
+ private void assertNumDocs(int expectedNumDocs, CloudJettyRunner jetty)
+ throws InterruptedException, IOException, SolrServerException {
+ SolrClient client = createNewSolrServer(jetty.url);
+ try {
+ int cnt = 30; // timeout after 15 seconds
+ AssertionError lastAssertionError = null;
+ while (cnt > 0) {
+ try {
+ assertEquals(expectedNumDocs, client.query(new SolrQuery("*:*")).getResults().getNumFound());
+ return;
+ }
+ catch (AssertionError e) {
+ lastAssertionError = e;
+ cnt--;
+ Thread.sleep(500);
+ }
+ }
+ throw new AssertionError("Timeout while trying to assert number of documents @ " + jetty.url, lastAssertionError);
+ } finally {
+ client.close();
+ }
+ }
+
+ private class UpdateThread implements Runnable {
+
+ private AtomicInteger numDocs;
+
+ private UpdateThread(AtomicInteger numDocs) {
+ this.numDocs = numDocs;
+ }
+
+ @Override
+ public void run() {
+ try {
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int j = numDocs.get(); j < (numDocs.get() + 10); j++) {
+ docs.add(getDoc(id, Integer.toString(j)));
+ }
+ index(SOURCE_COLLECTION, docs);
+ numDocs.getAndAdd(10);
+ log.info("Sent batch of {} updates - numDocs:{}", docs.size(), numDocs);
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ }
+
+ private List<CloudJettyRunner> getShardToSlaveJetty(String collection, String shard) {
+ List<CloudJettyRunner> jetties = new ArrayList<>(shardToJetty.get(collection).get(shard));
+ CloudJettyRunner leader = shardToLeaderJetty.get(collection).get(shard);
+ jetties.remove(leader);
+ return jetties;
+ }
+
+ /**
+ * Asserts that the update logs are in sync between the leader and slave. The leader and the slaves
+ * must have identical tlog files.
+ */
+ protected void assertUpdateLogsEquals(String collection, int numberOfTLogs) throws Exception {
+ CollectionInfo info = collectInfo(collection);
+ Map<String, List<CollectionInfo.CoreInfo>> shardToCoresMap = info.getShardToCoresMap();
+
+ for (String shard : shardToCoresMap.keySet()) {
+ Map<Long, Long> leaderFilesMeta = this.getFilesMeta(info.getLeader(shard).ulogDir);
+ Map<Long, Long> slaveFilesMeta = this.getFilesMeta(info.getReplicas(shard).get(0).ulogDir);
+
+ assertEquals("Incorrect number of tlog files on the leader", numberOfTLogs, leaderFilesMeta.size());
+ assertEquals("Incorrect number of tlog files on the slave", numberOfTLogs, slaveFilesMeta.size());
+
+ for (Long leaderFileVersion : leaderFilesMeta.keySet()) {
+ assertTrue("Slave is missing a tlog for version " + leaderFileVersion, slaveFilesMeta.containsKey(leaderFileVersion));
+ assertEquals("Slave's tlog file size differs for version " + leaderFileVersion, leaderFilesMeta.get(leaderFileVersion), slaveFilesMeta.get(leaderFileVersion));
+ }
+ }
+ }
+
+ private Map<Long, Long> getFilesMeta(String dir) {
+ File file = new File(dir);
+ if (!file.isDirectory()) {
+ assertTrue("Path to tlog " + dir + " does not exists or it's not a directory.", false);
+ }
+
+ Map<Long, Long> filesMeta = new HashMap<>();
+ for (File tlogFile : file.listFiles()) {
+ filesMeta.put(Math.abs(Long.parseLong(tlogFile.getName().substring(tlogFile.getName().lastIndexOf('.') + 1))), tlogFile.length());
+ }
+ return filesMeta;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrRequestHandlerTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrRequestHandlerTest.java b/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrRequestHandlerTest.java
new file mode 100644
index 0000000..237cc58
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrRequestHandlerTest.java
@@ -0,0 +1,153 @@
+/*
+ * 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.cdcr;
+
+import org.apache.lucene.util.LuceneTestCase.Nightly;
+import org.apache.solr.common.util.NamedList;
+import org.apache.solr.handler.CdcrParams;
+import org.junit.Test;
+
+@Nightly
+public class CdcrRequestHandlerTest extends BaseCdcrDistributedZkTest {
+
+ @Override
+ public void distribSetUp() throws Exception {
+ schemaString = "schema15.xml"; // we need a string id
+ createTargetCollection = false; // we do not need the target cluster
+ super.distribSetUp();
+ }
+
+ // check that the life-cycle state is properly synchronised across nodes
+ @Test
+ @ShardsFixed(num = 2)
+ public void testLifeCycleActions() throws Exception {
+ // check initial status
+ this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STOPPED, CdcrParams.BufferState.ENABLED);
+
+ // send start action to first shard
+ NamedList rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
+ NamedList status = (NamedList) rsp.get(CdcrParams.CdcrAction.STATUS.toLower());
+ assertEquals(CdcrParams.ProcessState.STARTED.toLower(), status.get(CdcrParams.ProcessState.getParam()));
+
+ // check status
+ this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STARTED, CdcrParams.BufferState.ENABLED);
+
+ // Restart the leader of shard 1
+ this.restartServer(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1));
+
+ // check status - the node that died should have picked up the original state
+ this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STARTED, CdcrParams.BufferState.ENABLED);
+
+ // send stop action to second shard
+ rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.STOP);
+ status = (NamedList) rsp.get(CdcrParams.CdcrAction.STATUS.toLower());
+ assertEquals(CdcrParams.ProcessState.STOPPED.toLower(), status.get(CdcrParams.ProcessState.getParam()));
+
+ // check status
+ this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STOPPED, CdcrParams.BufferState.ENABLED);
+ }
+
+ // check the checkpoint API
+ @Test
+ @ShardsFixed(num = 2)
+ public void testCheckpointActions() throws Exception {
+ // initial request on an empty index, must return -1
+ NamedList rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.COLLECTIONCHECKPOINT);
+ assertEquals(-1l, rsp.get(CdcrParams.CHECKPOINT));
+
+ index(SOURCE_COLLECTION, getDoc(id, "a")); // shard 2
+
+ // only one document indexed in shard 2, the checkpoint must be still -1
+ rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.COLLECTIONCHECKPOINT);
+ assertEquals(-1l, rsp.get(CdcrParams.CHECKPOINT));
+
+ index(SOURCE_COLLECTION, getDoc(id, "b")); // shard 1
+
+ // a second document indexed in shard 1, the checkpoint must come from shard 2
+ rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.COLLECTIONCHECKPOINT);
+ long checkpoint1 = (Long) rsp.get(CdcrParams.CHECKPOINT);
+ long expected = (Long) invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.SHARDCHECKPOINT).get(CdcrParams.CHECKPOINT);
+ assertEquals(expected, checkpoint1);
+
+ index(SOURCE_COLLECTION, getDoc(id, "c")); // shard 1
+
+ // a third document indexed in shard 1, the checkpoint must still come from shard 2
+ rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.COLLECTIONCHECKPOINT);
+ assertEquals(checkpoint1, rsp.get(CdcrParams.CHECKPOINT));
+
+ index(SOURCE_COLLECTION, getDoc(id, "d")); // shard 2
+
+ // a fourth document indexed in shard 2, the checkpoint must come from shard 1
+ rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.COLLECTIONCHECKPOINT);
+ long checkpoint2 = (Long) rsp.get(CdcrParams.CHECKPOINT);
+ expected = (Long) invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.SHARDCHECKPOINT).get(CdcrParams.CHECKPOINT);
+ assertEquals(expected, checkpoint2);
+
+ // send a delete by query
+ deleteByQuery(SOURCE_COLLECTION, "*:*");
+
+ // all the checkpoints must come from the DBQ
+ rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.COLLECTIONCHECKPOINT);
+ long checkpoint3 = (Long) rsp.get(CdcrParams.CHECKPOINT);
+ assertTrue(checkpoint3 > 0); // ensure that checkpoints from deletes are in absolute form
+ checkpoint3 = (Long) invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.SHARDCHECKPOINT).get(CdcrParams.CHECKPOINT);
+ assertTrue(checkpoint3 > 0); // ensure that checkpoints from deletes are in absolute form
+ checkpoint3 = (Long) invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.SHARDCHECKPOINT).get(CdcrParams.CHECKPOINT);
+ assertTrue(checkpoint3 > 0); // ensure that checkpoints from deletes are in absolute form
+
+ // replication never started, lastProcessedVersion should be -1 for both shards
+ rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.LASTPROCESSEDVERSION);
+ long lastVersion = (Long) rsp.get(CdcrParams.LAST_PROCESSED_VERSION);
+ assertEquals(-1l, lastVersion);
+
+ rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.LASTPROCESSEDVERSION);
+ lastVersion = (Long) rsp.get(CdcrParams.LAST_PROCESSED_VERSION);
+ assertEquals(-1l, lastVersion);
+ }
+
+ // check that the buffer state is properly synchronised across nodes
+ @Test
+ @ShardsFixed(num = 2)
+ public void testBufferActions() throws Exception {
+ // check initial status
+ this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STOPPED, CdcrParams.BufferState.ENABLED);
+
+ // send disable buffer action to first shard
+ NamedList rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.DISABLEBUFFER);
+ NamedList status = (NamedList) rsp.get(CdcrParams.CdcrAction.STATUS.toLower());
+ assertEquals(CdcrParams.BufferState.DISABLED.toLower(), status.get(CdcrParams.BufferState.getParam()));
+
+ // check status
+ this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STOPPED, CdcrParams.BufferState.DISABLED);
+
+ // Restart the leader of shard 1
+ this.restartServer(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1));
+
+ // check status
+ this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STOPPED, CdcrParams.BufferState.DISABLED);
+
+ // send enable buffer action to second shard
+ rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.ENABLEBUFFER);
+ status = (NamedList) rsp.get(CdcrParams.CdcrAction.STATUS.toLower());
+ assertEquals(CdcrParams.BufferState.ENABLED.toLower(), status.get(CdcrParams.BufferState.getParam()));
+
+ // check status
+ this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STOPPED, CdcrParams.BufferState.ENABLED);
+ }
+
+}
+
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrTestsUtil.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrTestsUtil.java b/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrTestsUtil.java
new file mode 100644
index 0000000..99aa471
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrTestsUtil.java
@@ -0,0 +1,111 @@
+/*
+ * 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.cdcr;
+
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.util.LinkedHashMap;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.SolrQuery;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.impl.CloudSolrClient;
+import org.apache.solr.client.solrj.response.QueryResponse;
+import org.apache.solr.common.params.CommonParams;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.NamedList;
+import org.apache.solr.handler.CdcrParams;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CdcrTestsUtil extends SolrTestCaseJ4{
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ protected static void cdcrStart(CloudSolrClient client) throws SolrServerException, IOException {
+ QueryResponse response = invokeCdcrAction(client, CdcrParams.CdcrAction.START);
+ assertEquals("started", ((NamedList) response.getResponse().get("status")).get("process"));
+ }
+
+ protected static void cdcrStop(CloudSolrClient client) throws SolrServerException, IOException {
+ QueryResponse response = invokeCdcrAction(client, CdcrParams.CdcrAction.STOP);
+ assertEquals("stopped", ((NamedList) response.getResponse().get("status")).get("process"));
+ }
+
+ protected static void cdcrEnableBuffer(CloudSolrClient client) throws IOException, SolrServerException {
+ QueryResponse response = invokeCdcrAction(client, CdcrParams.CdcrAction.ENABLEBUFFER);
+ assertEquals("enabled", ((NamedList) response.getResponse().get("status")).get("buffer"));
+ }
+
+ protected static void cdcrDisableBuffer(CloudSolrClient client) throws IOException, SolrServerException {
+ QueryResponse response = invokeCdcrAction(client, CdcrParams.CdcrAction.DISABLEBUFFER);
+ assertEquals("disabled", ((NamedList) response.getResponse().get("status")).get("buffer"));
+ }
+
+ protected static QueryResponse invokeCdcrAction(CloudSolrClient client, CdcrParams.CdcrAction action) throws IOException, SolrServerException {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+ params.set(CommonParams.QT, "/cdcr");
+ params.set(CommonParams.ACTION, action.toLower());
+ return client.query(params);
+ }
+
+ protected static QueryResponse getCdcrQueue(CloudSolrClient client) throws SolrServerException, IOException {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+ params.set(CommonParams.QT, "/cdcr");
+ params.set(CommonParams.ACTION, CdcrParams.QUEUES);
+ return client.query(params);
+ }
+
+ protected static Object getFingerPrintMaxVersion(CloudSolrClient client, String shardNames, int numDocs) throws SolrServerException, IOException, InterruptedException {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+ params.set(CommonParams.QT, "/get");
+ params.set("fingerprint", true);
+ params.set("shards", shardNames);
+ params.set("getVersions", numDocs);
+
+ QueryResponse response = null;
+ long start = System.nanoTime();
+ while (System.nanoTime() - start <= TimeUnit.NANOSECONDS.convert(20, TimeUnit.SECONDS)) {
+ response = client.query(params);
+ if (response.getResponse() != null && response.getResponse().get("fingerprint") != null) {
+ return (long)((LinkedHashMap)response.getResponse().get("fingerprint")).get("maxVersionEncountered");
+ }
+ Thread.sleep(200);
+ }
+ log.error("maxVersionEncountered not found for client : " + client + "in 20 attempts");
+ return null;
+ }
+
+ protected static long waitForClusterToSync(int numDocs, CloudSolrClient clusterSolrClient) throws Exception {
+ return waitForClusterToSync(numDocs, clusterSolrClient, "*:*");
+ }
+
+ protected static long waitForClusterToSync(int numDocs, CloudSolrClient clusterSolrClient, String query) throws Exception {
+ long start = System.nanoTime();
+ QueryResponse response = null;
+ while (System.nanoTime() - start <= TimeUnit.NANOSECONDS.convert(120, TimeUnit.SECONDS)) {
+ clusterSolrClient.commit();
+ response = clusterSolrClient.query(new SolrQuery(query));
+ if (response.getResults().getNumFound() == numDocs) {
+ break;
+ }
+ Thread.sleep(1000);
+ }
+ return response != null ? response.getResults().getNumFound() : 0;
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrVersionReplicationTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrVersionReplicationTest.java b/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrVersionReplicationTest.java
new file mode 100644
index 0000000..ff9afe2
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrVersionReplicationTest.java
@@ -0,0 +1,307 @@
+/*
+ * 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.cdcr;
+
+import java.lang.invoke.MethodHandles;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.impl.CloudSolrClient;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+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.params.CommonParams;
+import org.apache.solr.common.util.StrUtils;
+import org.apache.solr.update.processor.CdcrUpdateProcessor;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class CdcrVersionReplicationTest extends BaseCdcrDistributedZkTest {
+
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static final String vfield = CommonParams.VERSION_FIELD;
+ SolrClient solrServer;
+
+ public CdcrVersionReplicationTest() {
+ schemaString = "schema15.xml"; // we need a string id
+ super.createTargetCollection = false;
+ }
+
+ SolrClient createClientRandomly() throws Exception {
+ int r = random().nextInt(100);
+
+ // testing the smart cloud client (requests to leaders) is more important than testing the forwarding logic
+ if (r < 80) {
+ return createCloudClient(SOURCE_COLLECTION);
+ }
+
+ if (r < 90) {
+ return createNewSolrServer(shardToJetty.get(SOURCE_COLLECTION).get(SHARD1).get(random().nextInt(2)).url);
+ }
+
+ return createNewSolrServer(shardToJetty.get(SOURCE_COLLECTION).get(SHARD2).get(random().nextInt(2)).url);
+ }
+
+ @Test
+ @ShardsFixed(num = 4)
+ public void testCdcrDocVersions() throws Exception {
+ SolrClient client = createClientRandomly();
+ try {
+ handle.clear();
+ handle.put("timestamp", SKIPVAL);
+
+ doTestCdcrDocVersions(client);
+
+ commit(SOURCE_COLLECTION); // work arround SOLR-5628
+ } finally {
+ client.close();
+ }
+ }
+
+ private void doTestCdcrDocVersions(SolrClient solrClient) throws Exception {
+ this.solrServer = solrClient;
+
+ log.info("### STARTING doCdcrTestDocVersions - Add commands, client: " + solrClient);
+
+ vadd("doc1", 10, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "10");
+ vadd("doc2", 11, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "11");
+ vadd("doc3", 10, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "10");
+ vadd("doc4", 11, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "11");
+ commit(SOURCE_COLLECTION);
+
+ // versions are preserved and verifiable both by query and by real-time get
+ doQuery(solrClient, "doc1,10,doc2,11,doc3,10,doc4,11", "q", "*:*");
+ doRealTimeGet("doc1,doc2,doc3,doc4", "10,11,10,11");
+
+ vadd("doc1", 5, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "5");
+ vadd("doc2", 10, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "10");
+ vadd("doc3", 9, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "9");
+ vadd("doc4", 8, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "8");
+
+ // lower versions are ignored
+ doRealTimeGet("doc1,doc2,doc3,doc4", "10,11,10,11");
+
+ vadd("doc1", 12, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "12");
+ vadd("doc2", 12, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "12");
+ vadd("doc3", 12, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "12");
+ vadd("doc4", 12, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "12");
+
+ // higher versions are accepted
+ doRealTimeGet("doc1,doc2,doc3,doc4", "12,12,12,12");
+
+ // non-cdcr update requests throw a version conflict exception for non-equal versions (optimistic locking feature)
+ vaddFail("doc1", 13, 409);
+ vaddFail("doc2", 13, 409);
+ vaddFail("doc3", 13, 409);
+
+ commit(SOURCE_COLLECTION);
+
+ // versions are still as they were
+ doQuery(solrClient, "doc1,12,doc2,12,doc3,12,doc4,12", "q", "*:*");
+
+ // query all shard replicas individually
+ doQueryShardReplica(SHARD1, "doc1,12,doc2,12,doc3,12,doc4,12", "q", "*:*");
+ doQueryShardReplica(SHARD2, "doc1,12,doc2,12,doc3,12,doc4,12", "q", "*:*");
+
+ // optimistic locking update
+ vadd("doc4", 12);
+ commit(SOURCE_COLLECTION);
+
+ QueryResponse rsp = solrClient.query(params("qt", "/get", "ids", "doc4"));
+ long version = (long) rsp.getResults().get(0).get(vfield);
+
+ // update accepted and a new version number was generated
+ assertTrue(version > 1_000_000_000_000l);
+
+ log.info("### STARTING doCdcrTestDocVersions - Delete commands");
+
+ // send a delete update with an older version number
+ vdelete("doc1", 5, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "5");
+ // must ignore the delete
+ doRealTimeGet("doc1", "12");
+
+ // send a delete update with a higher version number
+ vdelete("doc1", 13, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "13");
+ // must be deleted
+ doRealTimeGet("doc1", "");
+
+ // send a delete update with a higher version number
+ vdelete("doc4", version + 1, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "" + (version + 1));
+ // must be deleted
+ doRealTimeGet("doc4", "");
+
+ commit(SOURCE_COLLECTION);
+
+ // query each shard replica individually
+ doQueryShardReplica(SHARD1, "doc2,12,doc3,12", "q", "*:*");
+ doQueryShardReplica(SHARD2, "doc2,12,doc3,12", "q", "*:*");
+
+ // version conflict thanks to optimistic locking
+ if (solrClient instanceof CloudSolrClient) // TODO: it seems that optimistic locking doesn't work with forwarding, test with shard2 client
+ vdeleteFail("doc2", 50, 409);
+
+ // cleanup after ourselves for the next run
+ // deleteByQuery should work as usual with the CDCR_UPDATE param
+ doDeleteByQuery("id:doc*", CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, Long.toString(1));
+ commit(SOURCE_COLLECTION);
+
+ // deleteByQuery with a version lower than anything else should have no effect
+ doQuery(solrClient, "doc2,12,doc3,12", "q", "*:*");
+
+ doDeleteByQuery("id:doc*", CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, Long.toString(51));
+ commit(SOURCE_COLLECTION);
+
+ // deleteByQuery with a version higher than everything else should delete all remaining docs
+ doQuery(solrClient, "", "q", "*:*");
+
+ // check that replicas are as expected too
+ doQueryShardReplica(SHARD1, "", "q", "*:*");
+ doQueryShardReplica(SHARD2, "", "q", "*:*");
+ }
+
+
+ // ------------------ auxiliary methods ------------------
+
+
+ void doQueryShardReplica(String shard, String expectedDocs, String... queryParams) throws Exception {
+ for (CloudJettyRunner jetty : shardToJetty.get(SOURCE_COLLECTION).get(shard)) {
+ doQuery(jetty.client, expectedDocs, queryParams);
+ }
+ }
+
+ void vdelete(String id, long version, String... params) throws Exception {
+ UpdateRequest req = new UpdateRequest();
+ req.deleteById(id);
+ req.setParam(vfield, Long.toString(version));
+
+ for (int i = 0; i < params.length; i += 2) {
+ req.setParam(params[i], params[i + 1]);
+ }
+ solrServer.request(req);
+ }
+
+ void vdeleteFail(String id, long version, int errCode, String... params) throws Exception {
+ boolean failed = false;
+ try {
+ vdelete(id, version, params);
+ } catch (SolrException e) {
+ failed = true;
+ if (e.getCause() instanceof SolrException && e.getCause() != e) {
+ e = (SolrException) e.getCause();
+ }
+ assertEquals(errCode, e.code());
+ } catch (SolrServerException ex) {
+ Throwable t = ex.getCause();
+ if (t instanceof SolrException) {
+ failed = true;
+ SolrException exception = (SolrException) t;
+ assertEquals(errCode, exception.code());
+ }
+ } catch (Exception e) {
+ log.error("ERROR", e);
+ }
+ assertTrue(failed);
+ }
+
+ void vadd(String id, long version, String... params) throws Exception {
+ UpdateRequest req = new UpdateRequest();
+ req.add(sdoc("id", id, vfield, version));
+ for (int i = 0; i < params.length; i += 2) {
+ req.setParam(params[i], params[i + 1]);
+ }
+ solrServer.request(req);
+ }
+
+ void vaddFail(String id, long version, int errCode, String... params) throws Exception {
+ boolean failed = false;
+ try {
+ vadd(id, version, params);
+ } catch (SolrException e) {
+ failed = true;
+ if (e.getCause() instanceof SolrException && e.getCause() != e) {
+ e = (SolrException) e.getCause();
+ }
+ assertEquals(errCode, e.code());
+ } catch (SolrServerException ex) {
+ Throwable t = ex.getCause();
+ if (t instanceof SolrException) {
+ failed = true;
+ SolrException exception = (SolrException) t;
+ assertEquals(errCode, exception.code());
+ }
+ } catch (Exception e) {
+ log.error("ERROR", e);
+ }
+ assertTrue(failed);
+ }
+
+ void doQuery(SolrClient ss, String expectedDocs, String... queryParams) throws Exception {
+
+ List<String> strs = StrUtils.splitSmart(expectedDocs, ",", true);
+ Map<String, Object> expectedIds = new HashMap<>();
+ for (int i = 0; i < strs.size(); i += 2) {
+ String id = strs.get(i);
+ String vS = strs.get(i + 1);
+ Long v = Long.valueOf(vS);
+ expectedIds.put(id, v);
+ }
+
+ QueryResponse rsp = ss.query(params(queryParams));
+ Map<String, Object> obtainedIds = new HashMap<>();
+ for (SolrDocument doc : rsp.getResults()) {
+ obtainedIds.put((String) doc.get("id"), doc.get(vfield));
+ }
+
+ assertEquals(expectedIds, obtainedIds);
+ }
+
+
+ void doRealTimeGet(String ids, String versions) throws Exception {
+ Map<String, Object> expectedIds = new HashMap<>();
+ List<String> strs = StrUtils.splitSmart(ids, ",", true);
+ List<String> verS = StrUtils.splitSmart(versions, ",", true);
+ for (int i = 0; i < strs.size(); i++) {
+ if (!verS.isEmpty()) {
+ expectedIds.put(strs.get(i), Long.valueOf(verS.get(i)));
+ }
+ }
+
+ QueryResponse rsp = solrServer.query(params("qt", "/get", "ids", ids));
+ Map<String, Object> obtainedIds = new HashMap<>();
+ for (SolrDocument doc : rsp.getResults()) {
+ obtainedIds.put((String) doc.get("id"), doc.get(vfield));
+ }
+
+ assertEquals(expectedIds, obtainedIds);
+ }
+
+ void doDeleteByQuery(String q, String... reqParams) throws Exception {
+ UpdateRequest req = new UpdateRequest();
+ req.deleteByQuery(q);
+ req.setParams(params(reqParams));
+ req.process(solrServer);
+ }
+
+}
+
[2/4] lucene-solr:branch_7x: SOLR-11003: Support bi-directional
syncing of cdcr clusters. We still only support actively into one cluster
cluster,
but have the ability to switch indexing clusters and cdcr will replicate
correctly
Posted by va...@apache.org.
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test/org/apache/solr/cloud/cdcr/BaseCdcrDistributedZkTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/cdcr/BaseCdcrDistributedZkTest.java b/solr/core/src/test/org/apache/solr/cloud/cdcr/BaseCdcrDistributedZkTest.java
new file mode 100644
index 0000000..e3a00a8
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/cdcr/BaseCdcrDistributedZkTest.java
@@ -0,0 +1,907 @@
+/*
+ * 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.cdcr;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrQuery;
+import org.apache.solr.client.solrj.SolrRequest;
+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.QueryRequest;
+import org.apache.solr.client.solrj.response.CollectionAdminResponse;
+import org.apache.solr.cloud.AbstractDistribZkTestBase;
+import org.apache.solr.cloud.AbstractZkTestCase;
+import org.apache.solr.cloud.ChaosMonkey;
+import org.apache.solr.cloud.OverseerCollectionMessageHandler;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.cloud.ClusterState;
+import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.Slice;
+import org.apache.solr.common.cloud.SolrZkClient;
+import org.apache.solr.common.cloud.ZkCoreNodeProps;
+import org.apache.solr.common.cloud.ZkNodeProps;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.params.CollectionParams;
+import org.apache.solr.common.params.CommonParams;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.IOUtils;
+import org.apache.solr.common.util.NamedList;
+import org.apache.solr.common.util.StrUtils;
+import org.apache.solr.common.util.Utils;
+import org.apache.solr.core.CoreDescriptor;
+import org.apache.solr.core.SolrCore;
+import org.apache.solr.handler.CdcrParams;
+import org.apache.solr.util.TimeOut;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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.SHARDS_PROP;
+import static org.apache.solr.common.cloud.ZkStateReader.CLUSTER_PROPS;
+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.handler.admin.CoreAdminHandler.COMPLETED;
+import static org.apache.solr.handler.admin.CoreAdminHandler.RESPONSE_STATUS;
+
+/**
+ * <p>
+ * Abstract class for CDCR unit testing. This class emulates two clusters, a source and target, by using different
+ * collections in the same SolrCloud cluster. Therefore, the two clusters will share the same Zookeeper cluster. In
+ * real scenario, the two collections/clusters will likely have their own zookeeper cluster.
+ * </p>
+ * <p>
+ * This class will automatically create two collections, the source and the target. Each collection will have
+ * {@link #shardCount} shards, and {@link #replicationFactor} replicas per shard. One jetty instance will
+ * be created per core.
+ * </p>
+ * <p>
+ * The source and target collection can be reinitialised at will by calling {@link #clearSourceCollection()} and
+ * {@link #clearTargetCollection()}. After reinitialisation, a collection will have a new fresh index and update log.
+ * </p>
+ * <p>
+ * Servers can be restarted at will by calling
+ * {@link #restartServer(BaseCdcrDistributedZkTest.CloudJettyRunner)} or
+ * {@link #restartServers(java.util.List)}.
+ * </p>
+ * <p>
+ * The creation of the target collection can be disabled with the flag {@link #createTargetCollection};
+ * </p>
+ * <p>
+ * NB: We cannot use multiple cores per jetty instance, as jetty will load only one core when restarting. It seems
+ * that this is a limitation of the {@link org.apache.solr.client.solrj.embedded.JettySolrRunner}. This class
+ * tries to ensure that there always is one single core per jetty instance.
+ * </p>
+ */
+public class BaseCdcrDistributedZkTest extends AbstractDistribZkTestBase {
+
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ protected int shardCount = 2;
+ protected int replicationFactor = 2;
+ protected boolean createTargetCollection = true;
+
+ private static final String CDCR_PATH = "/cdcr";
+
+ protected static final String SOURCE_COLLECTION = "source_collection";
+ protected static final String TARGET_COLLECTION = "target_collection";
+
+ public static final String SHARD1 = "shard1";
+ public static final String SHARD2 = "shard2";
+
+ @Override
+ protected String getCloudSolrConfig() {
+ return "solrconfig-cdcr.xml";
+ }
+
+ @Override
+ public void distribSetUp() throws Exception {
+ super.distribSetUp();
+
+ if (shardCount > 0) {
+ System.setProperty("numShards", Integer.toString(shardCount));
+ } else {
+ System.clearProperty("numShards");
+ }
+
+ if (isSSLMode()) {
+ System.clearProperty("urlScheme");
+ ZkStateReader zkStateReader = new ZkStateReader(zkServer.getZkAddress(),
+ AbstractZkTestCase.TIMEOUT, AbstractZkTestCase.TIMEOUT);
+ try {
+ zkStateReader.getZkClient().create(ZkStateReader.CLUSTER_PROPS,
+ Utils.toJSON(Collections.singletonMap("urlScheme", "https")),
+ CreateMode.PERSISTENT, true);
+ } catch (KeeperException.NodeExistsException e) {
+ ZkNodeProps props = ZkNodeProps.load(zkStateReader.getZkClient().getData(ZkStateReader.CLUSTER_PROPS,
+ null, null, true));
+ props = props.plus("urlScheme", "https");
+ zkStateReader.getZkClient().setData(CLUSTER_PROPS, Utils.toJSON(props), true);
+ } finally {
+ zkStateReader.close();
+ }
+ }
+ }
+
+ @Override
+ protected void createServers(int numServers) throws Exception {
+ }
+
+ @BeforeClass
+ public static void beforeClass() {
+ System.setProperty("solrcloud.update.delay", "0");
+ }
+
+ @AfterClass
+ public static void afterClass() throws Exception {
+ System.clearProperty("solrcloud.update.delay");
+ }
+
+ @Before
+ public void baseBefore() throws Exception {
+ this.createSourceCollection();
+ if (this.createTargetCollection) this.createTargetCollection();
+ RandVal.uniqueValues = new HashSet(); //reset random values
+ }
+
+ @After
+ public void baseAfter() throws Exception {
+ for (List<CloudJettyRunner> runners : cloudJettys.values()) {
+ for (CloudJettyRunner runner : runners) {
+ runner.client.close();
+ }
+ }
+ destroyServers();
+ }
+
+ protected CloudSolrClient createCloudClient(String defaultCollection) {
+ CloudSolrClient server = getCloudSolrClient(zkServer.getZkAddress(), random().nextBoolean());
+ server.setParallelUpdates(random().nextBoolean());
+ if (defaultCollection != null) server.setDefaultCollection(defaultCollection);
+ return server;
+ }
+
+ protected void printLayout() throws Exception {
+ SolrZkClient zkClient = new SolrZkClient(zkServer.getZkHost(), AbstractZkTestCase.TIMEOUT);
+ zkClient.printLayoutToStdOut();
+ zkClient.close();
+ }
+
+ protected SolrInputDocument getDoc(Object... fields) throws Exception {
+ SolrInputDocument doc = new SolrInputDocument();
+ addFields(doc, fields);
+ return doc;
+ }
+
+ protected void index(String collection, SolrInputDocument doc) throws IOException, SolrServerException {
+ CloudSolrClient client = createCloudClient(collection);
+ try {
+ client.add(doc);
+ client.commit(true, true);
+ } finally {
+ client.close();
+ }
+ }
+
+ protected void index(String collection, List<SolrInputDocument> docs) throws IOException, SolrServerException {
+ CloudSolrClient client = createCloudClient(collection);
+ try {
+ client.add(docs);
+ client.commit(true, true);
+ } finally {
+ client.close();
+ }
+ }
+
+ protected void deleteById(String collection, List<String> ids) throws IOException, SolrServerException {
+ CloudSolrClient client = createCloudClient(collection);
+ try {
+ client.deleteById(ids);
+ client.commit(true, true);
+ } finally {
+ client.close();
+ }
+ }
+
+ protected void deleteByQuery(String collection, String q) throws IOException, SolrServerException {
+ CloudSolrClient client = createCloudClient(collection);
+ try {
+ client.deleteByQuery(q);
+ client.commit(true, true);
+ } finally {
+ client.close();
+ }
+ }
+
+ /**
+ * Invokes a commit on the given collection.
+ */
+ protected void commit(String collection) throws IOException, SolrServerException {
+ CloudSolrClient client = createCloudClient(collection);
+ try {
+ client.commit(true, true);
+ } finally {
+ client.close();
+ }
+ }
+
+ /**
+ * Assert the number of documents in a given collection
+ */
+ protected void assertNumDocs(int expectedNumDocs, String collection)
+ throws SolrServerException, IOException, InterruptedException {
+ CloudSolrClient client = createCloudClient(collection);
+ try {
+ int cnt = 30; // timeout after 15 seconds
+ AssertionError lastAssertionError = null;
+ while (cnt > 0) {
+ try {
+ assertEquals(expectedNumDocs, client.query(new SolrQuery("*:*")).getResults().getNumFound());
+ return;
+ }
+ catch (AssertionError e) {
+ lastAssertionError = e;
+ cnt--;
+ Thread.sleep(500);
+ }
+ }
+ throw new AssertionError("Timeout while trying to assert number of documents @ " + collection, lastAssertionError);
+ } finally {
+ client.close();
+ }
+ }
+
+ /**
+ * Invokes a CDCR action on a given node.
+ */
+ protected NamedList invokeCdcrAction(CloudJettyRunner jetty, CdcrParams.CdcrAction action) throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+ params.set(CommonParams.ACTION, action.toString());
+
+ SolrRequest request = new QueryRequest(params);
+ request.setPath(CDCR_PATH);
+
+ return jetty.client.request(request);
+ }
+
+ protected void waitForCdcrStateReplication(String collection) throws Exception {
+ log.info("Wait for CDCR state to replicate - collection: " + collection);
+
+ int cnt = 30;
+ while (cnt > 0) {
+ NamedList status = null;
+ boolean allEquals = true;
+ for (CloudJettyRunner jetty : cloudJettys.get(collection)) { // check all replicas
+ NamedList rsp = invokeCdcrAction(jetty, CdcrParams.CdcrAction.STATUS);
+ if (status == null) {
+ status = (NamedList) rsp.get(CdcrParams.CdcrAction.STATUS.toLower());
+ continue;
+ }
+ allEquals &= status.equals(rsp.get(CdcrParams.CdcrAction.STATUS.toLower()));
+ }
+
+ if (allEquals) {
+ break;
+ }
+ else {
+ if (cnt == 0) {
+ throw new RuntimeException("Timeout waiting for CDCR state to replicate: collection="+collection);
+ }
+ cnt--;
+ Thread.sleep(500);
+ }
+ }
+
+ log.info("CDCR state is identical across nodes - collection: " + collection);
+ }
+
+ /**
+ * Assert the state of CDCR on each nodes of the given collection.
+ */
+ protected void assertState(String collection, CdcrParams.ProcessState processState, CdcrParams.BufferState bufferState)
+ throws Exception {
+ this.waitForCdcrStateReplication(collection); // ensure that cdcr state is replicated and stable
+ for (CloudJettyRunner jetty : cloudJettys.get(collection)) { // check all replicas
+ NamedList rsp = invokeCdcrAction(jetty, CdcrParams.CdcrAction.STATUS);
+ NamedList status = (NamedList) rsp.get(CdcrParams.CdcrAction.STATUS.toLower());
+ assertEquals(processState.toLower(), status.get(CdcrParams.ProcessState.getParam()));
+ assertEquals(bufferState.toLower(), status.get(CdcrParams.BufferState.getParam()));
+ }
+ }
+
+ /**
+ * A mapping between collection and node names. This is used when creating the collection in
+ * {@link #createCollection(String)}.
+ */
+ private Map<String, List<String>> collectionToNodeNames = new HashMap<>();
+
+ /**
+ * Starts the servers, saves and associates the node names to the source collection,
+ * and finally creates the source collection.
+ */
+ private void createSourceCollection() throws Exception {
+ List<String> nodeNames = this.startServers(shardCount * replicationFactor);
+ this.collectionToNodeNames.put(SOURCE_COLLECTION, nodeNames);
+ this.createCollection(SOURCE_COLLECTION);
+ this.waitForRecoveriesToFinish(SOURCE_COLLECTION, true);
+ this.updateMappingsFromZk(SOURCE_COLLECTION);
+ }
+
+ /**
+ * Clear the source collection. It will delete then create the collection through the collection API.
+ * The collection will have a new fresh index, i.e., including a new update log.
+ */
+ protected void clearSourceCollection() throws Exception {
+ this.deleteCollection(SOURCE_COLLECTION);
+ this.waitForCollectionToDisappear(SOURCE_COLLECTION);
+ this.createCollection(SOURCE_COLLECTION);
+ this.waitForRecoveriesToFinish(SOURCE_COLLECTION, true);
+ this.updateMappingsFromZk(SOURCE_COLLECTION);
+ }
+
+ /**
+ * Starts the servers, saves and associates the node names to the target collection,
+ * and finally creates the target collection.
+ */
+ private void createTargetCollection() throws Exception {
+ List<String> nodeNames = this.startServers(shardCount * replicationFactor);
+ this.collectionToNodeNames.put(TARGET_COLLECTION, nodeNames);
+ this.createCollection(TARGET_COLLECTION);
+ this.waitForRecoveriesToFinish(TARGET_COLLECTION, true);
+ this.updateMappingsFromZk(TARGET_COLLECTION);
+ }
+
+ /**
+ * Clear the source collection. It will delete then create the collection through the collection API.
+ * The collection will have a new fresh index, i.e., including a new update log.
+ */
+ protected void clearTargetCollection() throws Exception {
+ this.deleteCollection(TARGET_COLLECTION);
+ this.waitForCollectionToDisappear(TARGET_COLLECTION);
+ this.createCollection(TARGET_COLLECTION);
+ this.waitForRecoveriesToFinish(TARGET_COLLECTION, true);
+ this.updateMappingsFromZk(TARGET_COLLECTION);
+ }
+
+ /**
+ * Create a new collection through the Collection API. It enforces the use of one max shard per node.
+ * It will define the nodes to spread the new collection across by using the mapping {@link #collectionToNodeNames},
+ * to ensure that a node will not host more than one core (which will create problem when trying to restart servers).
+ */
+ private void createCollection(String name) throws Exception {
+ CloudSolrClient client = createCloudClient(null);
+ try {
+ // Create the target collection
+ Map<String, List<Integer>> collectionInfos = new HashMap<>();
+ int maxShardsPerNode = 1;
+
+ StringBuilder sb = new StringBuilder();
+ for (String nodeName : collectionToNodeNames.get(name)) {
+ sb.append(nodeName);
+ sb.append(',');
+ }
+ sb.deleteCharAt(sb.length() - 1);
+
+ createCollection(collectionInfos, name, shardCount, replicationFactor, maxShardsPerNode, client, sb.toString());
+ } finally {
+ client.close();
+ }
+ }
+
+ private CollectionAdminResponse createCollection(Map<String, List<Integer>> collectionInfos,
+ String collectionName, int numShards, int replicationFactor,
+ int maxShardsPerNode, SolrClient client, String createNodeSetStr)
+ throws SolrServerException, IOException {
+ return createCollection(collectionInfos, collectionName,
+ Utils.makeMap(
+ NUM_SLICES, numShards,
+ REPLICATION_FACTOR, replicationFactor,
+ CREATE_NODE_SET, createNodeSetStr,
+ MAX_SHARDS_PER_NODE, maxShardsPerNode),
+ client, "conf1");
+ }
+
+ private CollectionAdminResponse createCollection(Map<String, List<Integer>> collectionInfos, String collectionName,
+ Map<String, Object> collectionProps, SolrClient client,
+ String confSetName)
+ throws SolrServerException, IOException {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+ params.set("action", CollectionParams.CollectionAction.CREATE.toString());
+ for (Map.Entry<String, Object> entry : collectionProps.entrySet()) {
+ if (entry.getValue() != null) params.set(entry.getKey(), String.valueOf(entry.getValue()));
+ }
+ Integer numShards = (Integer) collectionProps.get(NUM_SLICES);
+ if (numShards == null) {
+ String shardNames = (String) collectionProps.get(SHARDS_PROP);
+ numShards = StrUtils.splitSmart(shardNames, ',').size();
+ }
+ Integer replicationFactor = (Integer) collectionProps.get(REPLICATION_FACTOR);
+ if (replicationFactor == null) {
+ replicationFactor = (Integer) OverseerCollectionMessageHandler.COLL_PROPS.get(REPLICATION_FACTOR);
+ }
+
+ if (confSetName != null) {
+ params.set("collection.configName", confSetName);
+ }
+
+ List<Integer> list = new ArrayList<>();
+ list.add(numShards);
+ list.add(replicationFactor);
+ if (collectionInfos != null) {
+ collectionInfos.put(collectionName, list);
+ }
+ params.set("name", collectionName);
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/admin/collections");
+
+ CollectionAdminResponse res = new CollectionAdminResponse();
+ res.setResponse(client.request(request));
+ return res;
+ }
+
+ /**
+ * Delete a collection through the Collection API.
+ */
+ protected CollectionAdminResponse deleteCollection(String collectionName) throws Exception {
+ SolrClient client = createCloudClient(null);
+ CollectionAdminResponse res;
+
+ try {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+ params.set("action", CollectionParams.CollectionAction.DELETE.toString());
+ params.set("name", collectionName);
+ QueryRequest request = new QueryRequest(params);
+ request.setPath("/admin/collections");
+
+ res = new CollectionAdminResponse();
+ res.setResponse(client.request(request));
+ } catch (Exception e) {
+ log.warn("Error while deleting the collection " + collectionName, e);
+ return new CollectionAdminResponse();
+ } finally {
+ client.close();
+ }
+
+ return res;
+ }
+
+ private void waitForCollectionToDisappear(String collection) throws Exception {
+ CloudSolrClient client = this.createCloudClient(null);
+ try {
+ client.connect();
+ ZkStateReader zkStateReader = client.getZkStateReader();
+ AbstractDistribZkTestBase.waitForCollectionToDisappear(collection, zkStateReader, false, true, 15);
+ } finally {
+ client.close();
+ }
+ }
+
+ private void waitForRecoveriesToFinish(String collection, boolean verbose) throws Exception {
+ CloudSolrClient client = this.createCloudClient(null);
+ try {
+ client.connect();
+ ZkStateReader zkStateReader = client.getZkStateReader();
+ super.waitForRecoveriesToFinish(collection, zkStateReader, verbose);
+ } finally {
+ client.close();
+ }
+ }
+
+ /**
+ * Asserts that the collection has the correct number of shards and replicas
+ */
+ protected void assertCollectionExpectations(String collectionName) throws Exception {
+ CloudSolrClient client = this.createCloudClient(null);
+ try {
+ client.connect();
+ ClusterState clusterState = client.getZkStateReader().getClusterState();
+
+ assertTrue("Could not find new collection " + collectionName, clusterState.hasCollection(collectionName));
+ Map<String, Slice> shards = clusterState.getCollection(collectionName).getSlicesMap();
+ // did we find expectedSlices shards/shards?
+ assertEquals("Found new collection " + collectionName + ", but mismatch on number of shards.", shardCount, shards.size());
+ int totalShards = 0;
+ for (String shardName : shards.keySet()) {
+ totalShards += shards.get(shardName).getReplicas().size();
+ }
+ int expectedTotalShards = shardCount * replicationFactor;
+ assertEquals("Found new collection " + collectionName + " with correct number of shards, but mismatch on number " +
+ "of shards.", expectedTotalShards, totalShards);
+ } finally {
+ client.close();
+ }
+ }
+
+ /**
+ * Restart a server.
+ */
+ protected void restartServer(CloudJettyRunner server) throws Exception {
+ // it seems we need to set the collection property to have the jetty properly restarted
+ System.setProperty("collection", server.collection);
+ JettySolrRunner jetty = server.jetty;
+ ChaosMonkey.stop(jetty);
+ ChaosMonkey.start(jetty);
+ System.clearProperty("collection");
+ waitForRecoveriesToFinish(server.collection, true);
+ updateMappingsFromZk(server.collection); // must update the mapping as the core node name might have changed
+ }
+
+ /**
+ * Restarts a list of servers.
+ */
+ protected void restartServers(List<CloudJettyRunner> servers) throws Exception {
+ for (CloudJettyRunner server : servers) {
+ this.restartServer(server);
+ }
+ }
+
+ private List<JettySolrRunner> jettys = new ArrayList<>();
+
+ /**
+ * Creates and starts a given number of servers.
+ */
+ protected List<String> startServers(int nServer) throws Exception {
+ String temporaryCollection = "tmp_collection";
+
+ for (int i = 1; i <= nServer; i++) {
+ // give everyone there own solrhome
+ File jettyDir = createTempDir("jetty").toFile();
+ jettyDir.mkdirs();
+ setupJettySolrHome(jettyDir);
+ JettySolrRunner jetty = createJetty(jettyDir, null, "shard" + i);
+ jettys.add(jetty);
+ }
+
+ try (SolrClient client = createCloudClient(temporaryCollection)) {
+ assertEquals(0, CollectionAdminRequest
+ .createCollection(temporaryCollection, "conf1", shardCount, 1)
+ .setCreateNodeSet("")
+ .process(client).getStatus());
+ for (int i = 0; i < jettys.size(); i++) {
+ assertTrue(CollectionAdminRequest
+ .addReplicaToShard(temporaryCollection, "shard"+((i % shardCount) + 1))
+ .setNode(jettys.get(i).getNodeName())
+ .process(client).isSuccess());
+ }
+ }
+
+ ZkStateReader zkStateReader = jettys.get(0).getCoreContainer().getZkController().getZkStateReader();
+
+ // now wait till we see the leader for each shard
+ for (int i = 1; i <= shardCount; i++) {
+ this.printLayout();
+ zkStateReader.getLeaderRetry(temporaryCollection, "shard" + i, 15000);
+ }
+
+ // store the node names
+ List<String> nodeNames = new ArrayList<>();
+ for (Slice shard : zkStateReader.getClusterState().getCollection(temporaryCollection).getSlices()) {
+ for (Replica replica : shard.getReplicas()) {
+ nodeNames.add(replica.getNodeName());
+ }
+ }
+
+ this.waitForRecoveriesToFinish(temporaryCollection,zkStateReader, true);
+ // delete the temporary collection - we will create our own collections later
+ this.deleteCollection(temporaryCollection);
+ this.waitForCollectionToDisappear(temporaryCollection);
+ System.clearProperty("collection");
+
+ return nodeNames;
+ }
+
+ @Override
+ protected void destroyServers() throws Exception {
+ for (JettySolrRunner runner : jettys) {
+ try {
+ ChaosMonkey.stop(runner);
+ } catch (Exception e) {
+ log.error("", e);
+ }
+ }
+
+ jettys.clear();
+ }
+
+ /**
+ * Mapping from collection to jettys
+ */
+ protected Map<String, List<CloudJettyRunner>> cloudJettys = new HashMap<>();
+
+ /**
+ * Mapping from collection/shard to jettys
+ */
+ protected Map<String, Map<String, List<CloudJettyRunner>>> shardToJetty = new HashMap<>();
+
+ /**
+ * Mapping from collection/shard leader to jettys
+ */
+ protected Map<String, Map<String, CloudJettyRunner>> shardToLeaderJetty = new HashMap<>();
+
+ /**
+ * Updates the mappings between the jetty's instances and the zookeeper cluster state.
+ */
+ protected void updateMappingsFromZk(String collection) throws Exception {
+ List<CloudJettyRunner> cloudJettys = new ArrayList<>();
+ Map<String, List<CloudJettyRunner>> shardToJetty = new HashMap<>();
+ Map<String, CloudJettyRunner> shardToLeaderJetty = new HashMap<>();
+
+ CloudSolrClient cloudClient = this.createCloudClient(null);
+ try {
+ cloudClient.connect();
+ ZkStateReader zkStateReader = cloudClient.getZkStateReader();
+ ClusterState clusterState = zkStateReader.getClusterState();
+ DocCollection coll = clusterState.getCollection(collection);
+
+ for (JettySolrRunner jetty : jettys) {
+ int port = jetty.getLocalPort();
+ if (port == -1) {
+ throw new RuntimeException("Cannot find the port for jetty");
+ }
+
+ nextJetty:
+ for (Slice shard : coll.getSlices()) {
+ Set<Map.Entry<String, Replica>> entries = shard.getReplicasMap().entrySet();
+ for (Map.Entry<String, Replica> entry : entries) {
+ Replica replica = entry.getValue();
+ if (replica.getStr(ZkStateReader.BASE_URL_PROP).contains(":" + port)) {
+ if (!shardToJetty.containsKey(shard.getName())) {
+ shardToJetty.put(shard.getName(), new ArrayList<CloudJettyRunner>());
+ }
+ boolean isLeader = shard.getLeader() == replica;
+ CloudJettyRunner cjr = new CloudJettyRunner(jetty, replica, collection, shard.getName(), entry.getKey());
+ shardToJetty.get(shard.getName()).add(cjr);
+ if (isLeader) {
+ shardToLeaderJetty.put(shard.getName(), cjr);
+ }
+ cloudJettys.add(cjr);
+ break nextJetty;
+ }
+ }
+ }
+ }
+
+ List<CloudJettyRunner> oldRunners = this.cloudJettys.putIfAbsent(collection, cloudJettys);
+ if (oldRunners != null) {
+ // must close resources for the old entries
+ for (CloudJettyRunner oldRunner : oldRunners) {
+ IOUtils.closeQuietly(oldRunner.client);
+ }
+ }
+
+ this.cloudJettys.put(collection, cloudJettys);
+ this.shardToJetty.put(collection, shardToJetty);
+ this.shardToLeaderJetty.put(collection, shardToLeaderJetty);
+ } finally {
+ cloudClient.close();
+ }
+ }
+
+ /**
+ * Wrapper around a {@link org.apache.solr.client.solrj.embedded.JettySolrRunner} to map the jetty
+ * instance to various information of the cloud cluster, such as the collection and shard
+ * that is served by the jetty instance, the node name, core node name, url, etc.
+ */
+ public static class CloudJettyRunner {
+
+ public JettySolrRunner jetty;
+ public String nodeName;
+ public String coreNodeName;
+ public String url;
+ public SolrClient client;
+ public Replica info;
+ public String shard;
+ public String collection;
+
+ public CloudJettyRunner(JettySolrRunner jetty, Replica replica,
+ String collection, String shard, String coreNodeName) {
+ this.jetty = jetty;
+ this.info = replica;
+ this.collection = collection;
+
+ Properties nodeProperties = jetty.getNodeProperties();
+
+ // we need to update the jetty's shard so that it registers itself to the right shard when restarted
+ this.shard = shard;
+ nodeProperties.setProperty(CoreDescriptor.CORE_SHARD, this.shard);
+
+ // we need to update the jetty's shard so that it registers itself under the right core name when restarted
+ this.coreNodeName = coreNodeName;
+ nodeProperties.setProperty(CoreDescriptor.CORE_NODE_NAME, this.coreNodeName);
+
+ this.nodeName = replica.getNodeName();
+
+ ZkCoreNodeProps coreNodeProps = new ZkCoreNodeProps(info);
+ this.url = coreNodeProps.getCoreUrl();
+
+ // strip the trailing slash as this can cause issues when executing requests
+ this.client = createNewSolrServer(this.url.substring(0, this.url.length() - 1));
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((url == null) ? 0 : url.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) return true;
+ if (obj == null) return false;
+ if (getClass() != obj.getClass()) return false;
+ CloudJettyRunner other = (CloudJettyRunner) obj;
+ if (url == null) {
+ if (other.url != null) return false;
+ } else if (!url.equals(other.url)) return false;
+ return true;
+ }
+
+ @Override
+ public String toString() {
+ return "CloudJettyRunner [url=" + url + "]";
+ }
+
+ }
+
+ protected static SolrClient createNewSolrServer(String baseUrl) {
+ try {
+ // setup the server...
+ HttpSolrClient s = getHttpSolrClient(baseUrl, DEFAULT_CONNECTION_TIMEOUT);
+ return s;
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+
+ protected void waitForBootstrapToComplete(String collectionName, String shardId) throws Exception {
+ NamedList rsp;// we need to wait until bootstrap is complete otherwise the replicator thread will never start
+ TimeOut timeOut = new TimeOut(60, TimeUnit.SECONDS);
+ while (!timeOut.hasTimedOut()) {
+ rsp = invokeCdcrAction(shardToLeaderJetty.get(collectionName).get(shardId), CdcrParams.CdcrAction.BOOTSTRAP_STATUS);
+ if (rsp.get(RESPONSE_STATUS).toString().equals(COMPLETED)) {
+ break;
+ }
+ Thread.sleep(1000);
+ }
+ }
+
+ protected void waitForReplicationToComplete(String collectionName, String shardId) throws Exception {
+ int cnt = 15;
+ while (cnt > 0) {
+ log.info("Checking queue size @ {}:{}", collectionName, shardId);
+ long size = this.getQueueSize(collectionName, shardId);
+ if (size == 0) { // if we received -1, it means that the log reader is not yet initialised, we should wait
+ return;
+ }
+ log.info("Waiting for replication to complete. Queue size: {} @ {}:{}", size, collectionName, shardId);
+ cnt--;
+ Thread.sleep(1000); // wait a bit for the replication to complete
+ }
+ throw new RuntimeException("Timeout waiting for CDCR replication to complete @" + collectionName + ":" + shardId);
+ }
+
+ protected long getQueueSize(String collectionName, String shardId) throws Exception {
+ NamedList rsp = this.invokeCdcrAction(shardToLeaderJetty.get(collectionName).get(shardId), CdcrParams.CdcrAction.QUEUES);
+ NamedList host = (NamedList) ((NamedList) rsp.get(CdcrParams.QUEUES)).getVal(0);
+ NamedList status = (NamedList) host.get(TARGET_COLLECTION);
+ return (Long) status.get(CdcrParams.QUEUE_SIZE);
+ }
+
+ protected CollectionInfo collectInfo(String collection) throws Exception {
+ CollectionInfo info = new CollectionInfo(collection);
+ for (String shard : shardToJetty.get(collection).keySet()) {
+ List<CloudJettyRunner> jettyRunners = shardToJetty.get(collection).get(shard);
+ for (CloudJettyRunner jettyRunner : jettyRunners) {
+ for (SolrCore core : jettyRunner.jetty.getCoreContainer().getCores()) {
+ info.addCore(core, shard, shardToLeaderJetty.get(collection).containsValue(jettyRunner));
+ }
+ }
+ }
+
+ return info;
+ }
+
+ protected static class CollectionInfo {
+
+ List<CoreInfo> coreInfos = new ArrayList<>();
+
+ String collection;
+
+ CollectionInfo(String collection) {
+ this.collection = collection;
+ }
+
+ /**
+ * @return Returns a map shard -> list of cores
+ */
+ Map<String, List<CoreInfo>> getShardToCoresMap() {
+ Map<String, List<CoreInfo>> map = new HashMap<>();
+ for (CoreInfo info : coreInfos) {
+ List<CoreInfo> list = map.get(info.shard);
+ if (list == null) {
+ list = new ArrayList<>();
+ map.put(info.shard, list);
+ }
+ list.add(info);
+ }
+ return map;
+ }
+
+ CoreInfo getLeader(String shard) {
+ List<CoreInfo> coreInfos = getShardToCoresMap().get(shard);
+ for (CoreInfo info : coreInfos) {
+ if (info.isLeader) {
+ return info;
+ }
+ }
+ assertTrue(String.format(Locale.ENGLISH, "There is no leader for collection %s shard %s", collection, shard), false);
+ return null;
+ }
+
+ List<CoreInfo> getReplicas(String shard) {
+ List<CoreInfo> coreInfos = getShardToCoresMap().get(shard);
+ coreInfos.remove(getLeader(shard));
+ return coreInfos;
+ }
+
+ void addCore(SolrCore core, String shard, boolean isLeader) throws Exception {
+ CoreInfo info = new CoreInfo();
+ info.collectionName = core.getName();
+ info.shard = shard;
+ info.isLeader = isLeader;
+ info.ulogDir = core.getUpdateHandler().getUpdateLog().getLogDir();
+
+ this.coreInfos.add(info);
+ }
+
+ public static class CoreInfo {
+ String collectionName;
+ String shard;
+ boolean isLeader;
+ String ulogDir;
+ }
+
+ }
+
+}
+
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrBidirectionalTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrBidirectionalTest.java b/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrBidirectionalTest.java
new file mode 100644
index 0000000..70cb5eb
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrBidirectionalTest.java
@@ -0,0 +1,234 @@
+/*
+ * 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.cdcr;
+
+import java.lang.invoke.MethodHandles;
+import java.util.concurrent.TimeUnit;
+
+import com.carrotsearch.ant.tasks.junit4.dependencies.com.google.common.collect.ImmutableMap;
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.SolrQuery;
+import org.apache.solr.client.solrj.impl.CloudSolrClient;
+import org.apache.solr.client.solrj.request.AbstractUpdateRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.client.solrj.response.QueryResponse;
+import org.apache.solr.cloud.MiniSolrCloudCluster;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.CommonParams;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.handler.CdcrParams;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CdcrBidirectionalTest extends SolrTestCaseJ4 {
+
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ @Test
+ public void testBiDir() throws Exception {
+ MiniSolrCloudCluster cluster2 = new MiniSolrCloudCluster(1, createTempDir("cdcr-cluster2"), buildJettyConfig("/solr"));
+ cluster2.waitForAllNodes(30);
+ MiniSolrCloudCluster cluster1 = new MiniSolrCloudCluster(1, createTempDir("cdcr-cluster1"), buildJettyConfig("/solr"));
+ cluster1.waitForAllNodes(30);
+ try {
+ log.info("cluster2 zkHost = " + cluster2.getZkServer().getZkAddress());
+ System.setProperty("cdcr.cluster2.zkHost", cluster2.getZkServer().getZkAddress());
+
+ log.info("cluster1 zkHost = " + cluster1.getZkServer().getZkAddress());
+ System.setProperty("cdcr.cluster1.zkHost", cluster1.getZkServer().getZkAddress());
+
+
+ cluster1.uploadConfigSet(configset("cdcr-cluster1"), "cdcr-cluster1");
+ CollectionAdminRequest.createCollection("cdcr-cluster1", "cdcr-cluster1", 2, 1)
+ .withProperty("solr.directoryFactory", "solr.StandardDirectoryFactory")
+ .setMaxShardsPerNode(2)
+ .process(cluster1.getSolrClient());
+ CloudSolrClient cluster1SolrClient = cluster1.getSolrClient();
+ cluster1SolrClient.setDefaultCollection("cdcr-cluster1");
+
+ cluster2.uploadConfigSet(configset("cdcr-cluster2"), "cdcr-cluster2");
+ CollectionAdminRequest.createCollection("cdcr-cluster2", "cdcr-cluster2", 2, 1)
+ .withProperty("solr.directoryFactory", "solr.StandardDirectoryFactory")
+ .setMaxShardsPerNode(2)
+ .process(cluster2.getSolrClient());
+ CloudSolrClient cluster2SolrClient = cluster2.getSolrClient();
+ cluster2SolrClient.setDefaultCollection("cdcr-cluster2");
+
+ UpdateRequest req = null;
+
+ CdcrTestsUtil.cdcrStart(cluster1SolrClient);
+ Thread.sleep(2000);
+
+ // ADD operation on cluster 1
+ int docs = (TEST_NIGHTLY ? 100 : 10);
+ int numDocs_c1 = 0;
+ for (int k = 0; k < docs; k++) {
+ req = new UpdateRequest();
+ for (; numDocs_c1 < (k + 1) * 100; numDocs_c1++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", "cluster1_" + numDocs_c1);
+ doc.addField("xyz", numDocs_c1);
+ req.add(doc);
+ }
+ req.setAction(AbstractUpdateRequest.ACTION.COMMIT, true, true);
+ log.info("Adding " + docs + " docs with commit=true, numDocs=" + numDocs_c1);
+ req.process(cluster1SolrClient);
+ }
+
+ QueryResponse response = cluster1SolrClient.query(new SolrQuery("*:*"));
+ assertEquals("cluster 1 docs mismatch", numDocs_c1, response.getResults().getNumFound());
+
+ assertEquals("cluster 2 docs mismatch", numDocs_c1, CdcrTestsUtil.waitForClusterToSync(numDocs_c1, cluster2SolrClient));
+
+ CdcrTestsUtil.cdcrStart(cluster2SolrClient); // FULL BI-DIRECTIONAL CDCR FORWARDING ON
+ Thread.sleep(2000);
+
+ // ADD operation on cluster 2
+ int numDocs_c2 = 0;
+ for (int k = 0; k < docs; k++) {
+ req = new UpdateRequest();
+ for (; numDocs_c2 < (k + 1) * 100; numDocs_c2++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", "cluster2_" + numDocs_c2);
+ doc.addField("xyz", numDocs_c2);
+ req.add(doc);
+ }
+ req.setAction(AbstractUpdateRequest.ACTION.COMMIT, true, true);
+ log.info("Adding " + docs + " docs with commit=true, numDocs=" + numDocs_c2);
+ req.process(cluster2SolrClient);
+ }
+
+ int numDocs = numDocs_c1 + numDocs_c2;
+
+ response = cluster2SolrClient.query(new SolrQuery("*:*"));
+ assertEquals("cluster 2 docs mismatch", numDocs, response.getResults().getNumFound());
+
+ assertEquals("cluster 1 docs mismatch", numDocs, CdcrTestsUtil.waitForClusterToSync(numDocs, cluster1SolrClient));
+
+ // logging cdcr clusters queue response
+ response = CdcrTestsUtil.getCdcrQueue(cluster1SolrClient);
+ log.info("Cdcr cluster1 queue response: " + response.getResponse());
+ response = CdcrTestsUtil.getCdcrQueue(cluster2SolrClient);
+ log.info("Cdcr cluster2 queue response: " + response.getResponse());
+
+ // lets find and keep the maximum version assigned by cluster1 & cluster2 across all our updates
+
+ long maxVersion_c1 = Math.min((long)CdcrTestsUtil.getFingerPrintMaxVersion(cluster1SolrClient, "shard1", numDocs),
+ (long)CdcrTestsUtil.getFingerPrintMaxVersion(cluster1SolrClient, "shard2", numDocs));
+ long maxVersion_c2 = Math.min((long)CdcrTestsUtil.getFingerPrintMaxVersion(cluster2SolrClient, "shard1", numDocs),
+ (long)CdcrTestsUtil.getFingerPrintMaxVersion(cluster2SolrClient, "shard2", numDocs));
+
+ ModifiableSolrParams params = new ModifiableSolrParams();
+ params.set(CommonParams.ACTION, CdcrParams.CdcrAction.COLLECTIONCHECKPOINT.toString());
+ params.set(CommonParams.QT, "/cdcr");
+ response = cluster2SolrClient.query(params);
+ Long checkpoint_2 = (Long) response.getResponse().get(CdcrParams.CHECKPOINT);
+ assertNotNull(checkpoint_2);
+
+ params = new ModifiableSolrParams();
+ params.set(CommonParams.ACTION, CdcrParams.CdcrAction.COLLECTIONCHECKPOINT.toString());
+ params.set(CommonParams.QT, "/cdcr");
+ response = cluster1SolrClient.query(params);
+ Long checkpoint_1 = (Long) response.getResponse().get(CdcrParams.CHECKPOINT);
+ assertNotNull(checkpoint_1);
+
+ log.info("v1: " + maxVersion_c1 + "\t" + "v2: " + maxVersion_c2 + "\t" +
+ "checkpoint1: " + checkpoint_1 + "\t" + "checkpoint2: " + checkpoint_2);
+
+ assertEquals("COLLECTIONCHECKPOINT from cluster2 should have returned the maximum " +
+ "version across all updates made to cluster1", maxVersion_c1, checkpoint_2.longValue());
+ assertEquals("COLLECTIONCHECKPOINT from cluster1 should have returned the maximum " +
+ "version across all updates made to cluster2", maxVersion_c2, checkpoint_1.longValue());
+ assertEquals("max versions of updates in both clusters should be same", maxVersion_c1, maxVersion_c2);
+
+ // DELETE BY QUERY
+ String deleteByQuery = "id:cluster1_" +String.valueOf(random().nextInt(numDocs_c1));
+ response = cluster1SolrClient.query(new SolrQuery(deleteByQuery));
+ assertEquals("should match exactly one doc", 1, response.getResults().getNumFound());
+ cluster1SolrClient.deleteByQuery(deleteByQuery);
+ cluster1SolrClient.commit();
+ numDocs--;
+ numDocs_c1--;
+
+ response = cluster1SolrClient.query(new SolrQuery("*:*"));
+ assertEquals("cluster 1 docs mismatch", numDocs, response.getResults().getNumFound());
+ assertEquals("cluster 2 docs mismatch", numDocs, CdcrTestsUtil.waitForClusterToSync(numDocs, cluster2SolrClient));
+
+ // DELETE BY ID
+ SolrInputDocument doc;
+ String delete_id_query = "cluster2_" + random().nextInt(numDocs_c2);
+ cluster2SolrClient.deleteById(delete_id_query);
+ cluster2SolrClient.commit();
+ numDocs--;
+ numDocs_c2--;
+ response = cluster2SolrClient.query(new SolrQuery("*:*"));
+ assertEquals("cluster 2 docs mismatch", numDocs, response.getResults().getNumFound());
+ assertEquals("cluster 1 docs mismatch", numDocs, CdcrTestsUtil.waitForClusterToSync(numDocs, cluster1SolrClient));
+
+ // ATOMIC UPDATES
+ req = new UpdateRequest();
+ doc = new SolrInputDocument();
+ ImmutableMap.of("", "");
+ String atomicUpdateId = "cluster2_" + random().nextInt(numDocs_c2);
+ doc.addField("id", atomicUpdateId);
+ doc.addField("xyz", ImmutableMap.of("delete", ""));
+ doc.addField("abc", ImmutableMap.of("set", "ABC"));
+ req.add(doc);
+ req.process(cluster2SolrClient);
+ cluster2SolrClient.commit();
+
+ String atomicQuery = "id:" + atomicUpdateId;
+ response = cluster2SolrClient.query(new SolrQuery(atomicQuery));
+ assertEquals("cluster 2 wrong doc", "ABC", response.getResults().get(0).get("abc"));
+ assertEquals("cluster 1 wrong doc", "ABC", getDocFieldValue(cluster1SolrClient, atomicQuery, "ABC"));
+
+
+ // logging cdcr clusters queue response
+ response = CdcrTestsUtil.getCdcrQueue(cluster1SolrClient);
+ log.info("Cdcr cluster1 queue response at end of testcase: " + response.getResponse());
+ response = CdcrTestsUtil.getCdcrQueue(cluster2SolrClient);
+ log.info("Cdcr cluster2 queue response at end of testcase: " + response.getResponse());
+
+ CdcrTestsUtil.cdcrStop(cluster1SolrClient);
+ CdcrTestsUtil.cdcrStop(cluster2SolrClient);
+ } finally {
+ if (cluster1 != null) {
+ cluster1.shutdown();
+ }
+ if (cluster2 != null) {
+ cluster2.shutdown();
+ }
+ }
+ }
+
+ private String getDocFieldValue(CloudSolrClient clusterSolrClient, String query, String match) throws Exception {
+ long start = System.nanoTime();
+ QueryResponse response = null;
+ while (System.nanoTime() - start <= TimeUnit.NANOSECONDS.convert(120, TimeUnit.SECONDS)) {
+ clusterSolrClient.commit();
+ response = clusterSolrClient.query(new SolrQuery(query));
+ if (match.equals(response.getResults().get(0).get("abc"))) {
+ break;
+ }
+ Thread.sleep(1000);
+ }
+ return response != null ? (String) response.getResults().get(0).get("abc") : "";
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrBootstrapTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrBootstrapTest.java b/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrBootstrapTest.java
new file mode 100644
index 0000000..2cabb50
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrBootstrapTest.java
@@ -0,0 +1,302 @@
+/*
+ * 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.cdcr;
+
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.util.LinkedHashMap;
+
+import org.apache.lucene.store.FSDirectory;
+import org.apache.solr.SolrTestCaseJ4;
+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.request.AbstractUpdateRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.client.solrj.response.QueryResponse;
+import org.apache.solr.cloud.AbstractDistribZkTestBase;
+import org.apache.solr.cloud.MiniSolrCloudCluster;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.CommonParams;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.handler.CdcrParams;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CdcrBootstrapTest extends SolrTestCaseJ4 {
+
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ /**
+ * Starts a source cluster with no CDCR configuration, indexes enough documents such that
+ * the at least one old tlog is closed and thrown away so that the source cluster does not have
+ * all updates available in tlogs only.
+ * <p>
+ * Then we start a target cluster with CDCR configuration and we change the source cluster configuration
+ * to use CDCR (i.e. CdcrUpdateLog, CdcrRequestHandler and CdcrUpdateProcessor) and restart it.
+ * <p>
+ * We test that all updates eventually make it to the target cluster and that the collectioncheckpoint
+ * call returns the same version as the last update indexed on the source.
+ */
+ @Test
+ public void testConvertClusterToCdcrAndBootstrap() throws Exception {
+ // start the target first so that we know its zkhost
+ MiniSolrCloudCluster target = new MiniSolrCloudCluster(1, createTempDir("cdcr-target"), buildJettyConfig("/solr"));
+ try {
+ target.waitForAllNodes(30);
+ log.info("Target zkHost = " + target.getZkServer().getZkAddress());
+ System.setProperty("cdcr.target.zkHost", target.getZkServer().getZkAddress());
+
+ // start a cluster with no cdcr
+ MiniSolrCloudCluster source = new MiniSolrCloudCluster(1, createTempDir("cdcr-source"), buildJettyConfig("/solr"));
+ try {
+ source.waitForAllNodes(30);
+ source.uploadConfigSet(configset("cdcr-source-disabled"), "cdcr-source");
+
+ // create a collection with the cdcr-source-disabled configset
+ CollectionAdminRequest.createCollection("cdcr-source", "cdcr-source", 1, 1)
+ // todo investigate why this is necessary??? because by default it selects a ram directory which deletes the tlogs on reloads?
+ .withProperty("solr.directoryFactory", "solr.StandardDirectoryFactory")
+ .process(source.getSolrClient());
+
+ CloudSolrClient sourceSolrClient = source.getSolrClient();
+ int docs = (TEST_NIGHTLY ? 100 : 10);
+ int numDocs = indexDocs(sourceSolrClient, "cdcr-source", docs);
+
+ QueryResponse response = sourceSolrClient.query(new SolrQuery("*:*"));
+ assertEquals("", numDocs, response.getResults().getNumFound());
+
+ // lets find and keep the maximum version assigned by source cluster across all our updates
+ long maxVersion = Long.MIN_VALUE;
+ ModifiableSolrParams params = new ModifiableSolrParams();
+ params.set(CommonParams.QT, "/get");
+ params.set("getVersions", numDocs);
+ params.set("fingerprint", true);
+ response = sourceSolrClient.query(params);
+ maxVersion = (long)(((LinkedHashMap)response.getResponse().get("fingerprint")).get("maxVersionEncountered"));
+
+// upload the cdcr-enabled config and restart source cluster
+ source.uploadConfigSet(configset("cdcr-source"), "cdcr-source");
+ JettySolrRunner runner = source.stopJettySolrRunner(0);
+ source.startJettySolrRunner(runner);
+ assertTrue(runner.isRunning());
+ AbstractDistribZkTestBase.waitForRecoveriesToFinish("cdcr-source", source.getSolrClient().getZkStateReader(), true, true, 330);
+
+ response = sourceSolrClient.query(new SolrQuery("*:*"));
+ assertEquals("Document mismatch on source after restart", numDocs, response.getResults().getNumFound());
+
+ // setup the target cluster
+ target.uploadConfigSet(configset("cdcr-target"), "cdcr-target");
+ CollectionAdminRequest.createCollection("cdcr-target", "cdcr-target", 1, 1)
+ .process(target.getSolrClient());
+ CloudSolrClient targetSolrClient = target.getSolrClient();
+ targetSolrClient.setDefaultCollection("cdcr-target");
+ Thread.sleep(1000);
+
+ CdcrTestsUtil.cdcrStart(targetSolrClient);
+ CdcrTestsUtil.cdcrStart(sourceSolrClient);
+
+ response = CdcrTestsUtil.getCdcrQueue(sourceSolrClient);
+ log.info("Cdcr queue response: " + response.getResponse());
+ long foundDocs = CdcrTestsUtil.waitForClusterToSync(numDocs, targetSolrClient);
+ assertEquals("Document mismatch on target after sync", numDocs, foundDocs);
+
+ params = new ModifiableSolrParams();
+ params.set(CommonParams.ACTION, CdcrParams.CdcrAction.COLLECTIONCHECKPOINT.toString());
+ params.set(CommonParams.QT, "/cdcr");
+ response = targetSolrClient.query(params);
+ Long checkpoint = (Long) response.getResponse().get(CdcrParams.CHECKPOINT);
+ assertNotNull(checkpoint);
+ assertEquals("COLLECTIONCHECKPOINT from target cluster should have returned the maximum " +
+ "version across all updates made to source", maxVersion, checkpoint.longValue());
+ } finally {
+ source.shutdown();
+ }
+ } finally {
+ target.shutdown();
+ }
+ }
+
+ private int indexDocs(CloudSolrClient sourceSolrClient, String collection, int batches) throws IOException, SolrServerException {
+ sourceSolrClient.setDefaultCollection(collection);
+ int numDocs = 0;
+ for (int k = 0; k < batches; k++) {
+ UpdateRequest req = new UpdateRequest();
+ for (; numDocs < (k + 1) * 100; numDocs++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", "source_" + numDocs);
+ doc.addField("xyz", numDocs);
+ req.add(doc);
+ }
+ req.setAction(AbstractUpdateRequest.ACTION.COMMIT, true, true);
+ req.process(sourceSolrClient);
+ }
+ log.info("Adding numDocs=" + numDocs);
+ return numDocs;
+ }
+ /**
+ * This test start cdcr source, adds data,starts target cluster, verifies replication,
+ * stops cdcr replication and buffering, adds more data, re-enables cdcr and verify replication
+ */
+ public void testBootstrapWithSourceCluster() throws Exception {
+ // start the target first so that we know its zkhost
+ MiniSolrCloudCluster target = new MiniSolrCloudCluster(1, createTempDir("cdcr-target"), buildJettyConfig("/solr"));
+ try {
+ target.waitForAllNodes(30);
+ System.out.println("Target zkHost = " + target.getZkServer().getZkAddress());
+ System.setProperty("cdcr.target.zkHost", target.getZkServer().getZkAddress());
+
+ MiniSolrCloudCluster source = new MiniSolrCloudCluster(1, createTempDir("cdcr-source"), buildJettyConfig("/solr"));
+ try {
+ source.waitForAllNodes(30);
+ source.uploadConfigSet(configset("cdcr-source"), "cdcr-source");
+
+ CollectionAdminRequest.createCollection("cdcr-source", "cdcr-source", 1, 1)
+ .withProperty("solr.directoryFactory", "solr.StandardDirectoryFactory")
+ .process(source.getSolrClient());
+
+ CloudSolrClient sourceSolrClient = source.getSolrClient();
+ int docs = (TEST_NIGHTLY ? 100 : 10);
+ int numDocs = indexDocs(sourceSolrClient, "cdcr-source", docs);
+
+ QueryResponse response = sourceSolrClient.query(new SolrQuery("*:*"));
+ assertEquals("", numDocs, response.getResults().getNumFound());
+
+ // setup the target cluster
+ target.uploadConfigSet(configset("cdcr-target"), "cdcr-target");
+ CollectionAdminRequest.createCollection("cdcr-target", "cdcr-target", 1, 1)
+ .process(target.getSolrClient());
+ CloudSolrClient targetSolrClient = target.getSolrClient();
+ targetSolrClient.setDefaultCollection("cdcr-target");
+
+ CdcrTestsUtil.cdcrStart(targetSolrClient);
+ CdcrTestsUtil.cdcrStart(sourceSolrClient);
+
+ response = CdcrTestsUtil.getCdcrQueue(sourceSolrClient);
+ log.info("Cdcr queue response: " + response.getResponse());
+ long foundDocs = CdcrTestsUtil.waitForClusterToSync(numDocs, targetSolrClient);
+ assertEquals("Document mismatch on target after sync", numDocs, foundDocs);
+
+ int total_tlogs_in_index = FSDirectory.open(target.getBaseDir().resolve("node1").
+ resolve("cdcr-target_shard1_replica_n1").resolve("data").
+ resolve("tlog")).listAll().length;
+
+ assertEquals("tlogs count should be ZERO",0, total_tlogs_in_index);
+
+ CdcrTestsUtil.cdcrStop(sourceSolrClient);
+ CdcrTestsUtil.cdcrDisableBuffer(sourceSolrClient);
+
+ int c = 0;
+ for (int k = 0; k < 10; k++) {
+ UpdateRequest req = new UpdateRequest();
+ for (; c < (k + 1) * 100; c++, numDocs++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", "source_" + numDocs);
+ doc.addField("xyz", numDocs);
+ req.add(doc);
+ }
+ req.setAction(AbstractUpdateRequest.ACTION.COMMIT, true, true);
+ log.info("Adding 100 docs with commit=true, numDocs=" + numDocs);
+ req.process(sourceSolrClient);
+ }
+
+ response = sourceSolrClient.query(new SolrQuery("*:*"));
+ assertEquals("", numDocs, response.getResults().getNumFound());
+
+ CdcrTestsUtil.cdcrStart(sourceSolrClient);
+ CdcrTestsUtil.cdcrEnableBuffer(sourceSolrClient);
+
+ foundDocs = CdcrTestsUtil.waitForClusterToSync(numDocs, targetSolrClient);
+ assertEquals("Document mismatch on target after sync", numDocs, foundDocs);
+
+ } finally {
+ source.shutdown();
+ }
+ } finally {
+ target.shutdown();
+ }
+ }
+
+ public void testBootstrapWithContinousIndexingOnSourceCluster() throws Exception {
+ // start the target first so that we know its zkhost
+ MiniSolrCloudCluster target = new MiniSolrCloudCluster(1, createTempDir("cdcr-target"), buildJettyConfig("/solr"));
+ target.waitForAllNodes(30);
+ try {
+ log.info("Target zkHost = " + target.getZkServer().getZkAddress());
+ System.setProperty("cdcr.target.zkHost", target.getZkServer().getZkAddress());
+
+ MiniSolrCloudCluster source = new MiniSolrCloudCluster(1, createTempDir("cdcr-source"), buildJettyConfig("/solr"));
+ try {
+ source.waitForAllNodes(30);
+ source.uploadConfigSet(configset("cdcr-source"), "cdcr-source");
+
+ CollectionAdminRequest.createCollection("cdcr-source", "cdcr-source", 1, 1)
+ .withProperty("solr.directoryFactory", "solr.StandardDirectoryFactory")
+ .process(source.getSolrClient());
+
+ CloudSolrClient sourceSolrClient = source.getSolrClient();
+ int docs = (TEST_NIGHTLY ? 100 : 10);
+ int numDocs = indexDocs(sourceSolrClient, "cdcr-source", docs);
+
+ QueryResponse response = sourceSolrClient.query(new SolrQuery("*:*"));
+ assertEquals("", numDocs, response.getResults().getNumFound());
+
+ // setup the target cluster
+ target.uploadConfigSet(configset("cdcr-target"), "cdcr-target");
+ CollectionAdminRequest.createCollection("cdcr-target", "cdcr-target", 1, 1)
+ .process(target.getSolrClient());
+ CloudSolrClient targetSolrClient = target.getSolrClient();
+ targetSolrClient.setDefaultCollection("cdcr-target");
+ Thread.sleep(1000);
+
+ CdcrTestsUtil.cdcrStart(targetSolrClient);
+ CdcrTestsUtil.cdcrStart(sourceSolrClient);
+ int c = 0;
+ for (int k = 0; k < docs; k++) {
+ UpdateRequest req = new UpdateRequest();
+ for (; c < (k + 1) * 100; c++, numDocs++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", "source_" + numDocs);
+ doc.addField("xyz", numDocs);
+ req.add(doc);
+ }
+ req.setAction(AbstractUpdateRequest.ACTION.COMMIT, true, true);
+ log.info("Adding " + docs + " docs with commit=true, numDocs=" + numDocs);
+ req.process(sourceSolrClient);
+ }
+
+ response = sourceSolrClient.query(new SolrQuery("*:*"));
+ assertEquals("", numDocs, response.getResults().getNumFound());
+
+ response = CdcrTestsUtil.getCdcrQueue(sourceSolrClient);
+ log.info("Cdcr queue response: " + response.getResponse());
+ long foundDocs = CdcrTestsUtil.waitForClusterToSync(numDocs, targetSolrClient);
+ assertEquals("Document mismatch on target after sync", numDocs, foundDocs);
+
+ } finally {
+ source.shutdown();
+ }
+ } finally {
+ target.shutdown();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrReplicationDistributedZkTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrReplicationDistributedZkTest.java b/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrReplicationDistributedZkTest.java
new file mode 100644
index 0000000..206a08f
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/cdcr/CdcrReplicationDistributedZkTest.java
@@ -0,0 +1,672 @@
+/*
+ * 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.cdcr;
+
+import java.io.File;
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.lucene.util.LuceneTestCase.BadApple;
+import org.apache.lucene.util.LuceneTestCase.Nightly;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.util.NamedList;
+import org.apache.solr.handler.CdcrParams;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Nightly
+@BadApple(bugUrl = "https://issues.apache.org/jira/browse/SOLR-10107")
+public class CdcrReplicationDistributedZkTest extends BaseCdcrDistributedZkTest {
+
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ @Override
+ public void distribSetUp() throws Exception {
+ schemaString = "schema15.xml"; // we need a string id
+ super.distribSetUp();
+ }
+
+ /**
+ * Checks that the test framework handles properly the creation and deletion of collections and the
+ * restart of servers.
+ */
+ @Test
+ @ShardsFixed(num = 4)
+ public void testDeleteCreateSourceCollection() throws Exception {
+ log.info("Indexing documents");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ docs.add(getDoc(id, Integer.toString(i)));
+ }
+ index(SOURCE_COLLECTION, docs);
+ index(TARGET_COLLECTION, docs);
+
+ assertNumDocs(10, SOURCE_COLLECTION);
+ assertNumDocs(10, TARGET_COLLECTION);
+
+ log.info("Restarting leader @ source_collection:shard1");
+
+ this.restartServer(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1));
+
+ assertNumDocs(10, SOURCE_COLLECTION);
+ assertNumDocs(10, TARGET_COLLECTION);
+
+ log.info("Clearing source_collection");
+
+ this.clearSourceCollection();
+
+ assertNumDocs(0, SOURCE_COLLECTION);
+ assertNumDocs(10, TARGET_COLLECTION);
+
+ log.info("Restarting leader @ target_collection:shard1");
+
+ this.restartServer(shardToLeaderJetty.get(TARGET_COLLECTION).get(SHARD1));
+
+ assertNumDocs(0, SOURCE_COLLECTION);
+ assertNumDocs(10, TARGET_COLLECTION);
+
+ log.info("Clearing target_collection");
+
+ this.clearTargetCollection();
+
+ assertNumDocs(0, SOURCE_COLLECTION);
+ assertNumDocs(0, TARGET_COLLECTION);
+
+ assertCollectionExpectations(SOURCE_COLLECTION);
+ assertCollectionExpectations(TARGET_COLLECTION);
+ }
+
+ @Test
+ @ShardsFixed(num = 4)
+ public void testTargetCollectionNotAvailable() throws Exception {
+ // send start action to first shard
+ NamedList rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
+ NamedList status = (NamedList) rsp.get(CdcrParams.CdcrAction.STATUS.toLower());
+ assertEquals(CdcrParams.ProcessState.STARTED.toLower(), status.get(CdcrParams.ProcessState.getParam()));
+
+ // check status
+ this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STARTED, CdcrParams.BufferState.ENABLED);
+
+ this.waitForBootstrapToComplete(TARGET_COLLECTION, SHARD2);
+
+ // sleep for a bit to ensure that replicator threads are started
+ Thread.sleep(3000);
+
+ // Kill all the servers of the target
+ this.deleteCollection(TARGET_COLLECTION);
+
+ // Index a few documents to trigger the replication
+ index(SOURCE_COLLECTION, getDoc(id, "a"));
+ index(SOURCE_COLLECTION, getDoc(id, "b"));
+ index(SOURCE_COLLECTION, getDoc(id, "c"));
+ index(SOURCE_COLLECTION, getDoc(id, "d"));
+ index(SOURCE_COLLECTION, getDoc(id, "e"));
+ index(SOURCE_COLLECTION, getDoc(id, "f"));
+
+ assertNumDocs(6, SOURCE_COLLECTION);
+
+ // we need to wait until the replicator thread is triggered
+ int cnt = 15; // timeout after 15 seconds
+ AssertionError lastAssertionError = null;
+ while (cnt > 0) {
+ try {
+ rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.ERRORS);
+ NamedList collections = (NamedList) ((NamedList) rsp.get(CdcrParams.ERRORS)).getVal(0);
+ NamedList errors = (NamedList) collections.get(TARGET_COLLECTION);
+ assertTrue(0 < (Long) errors.get(CdcrParams.CONSECUTIVE_ERRORS));
+ NamedList lastErrors = (NamedList) errors.get(CdcrParams.LAST);
+ assertNotNull(lastErrors);
+ assertTrue(0 < lastErrors.size());
+ return;
+ }
+ catch (AssertionError e) {
+ lastAssertionError = e;
+ cnt--;
+ Thread.sleep(1000);
+ }
+ }
+
+ throw new AssertionError("Timeout while trying to assert replication errors", lastAssertionError);
+ }
+
+ @Test
+ @ShardsFixed(num = 4)
+ public void testReplicationStartStop() throws Exception {
+ int start = 0;
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (; start < 10; start++) {
+ docs.add(getDoc(id, Integer.toString(start)));
+ }
+ index(SOURCE_COLLECTION, docs);
+
+ assertNumDocs(10, SOURCE_COLLECTION);
+ assertNumDocs(0, TARGET_COLLECTION);
+
+ this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
+ this.waitForCdcrStateReplication(SOURCE_COLLECTION);
+
+ this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD2);
+
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
+
+ commit(TARGET_COLLECTION);
+
+ assertNumDocs(10, SOURCE_COLLECTION);
+ assertNumDocs(10, TARGET_COLLECTION);
+
+ this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.STOP);
+ this.waitForCdcrStateReplication(SOURCE_COLLECTION);
+
+ docs.clear();
+ for (; start < 110; start++) {
+ docs.add(getDoc(id, Integer.toString(start)));
+ }
+ index(SOURCE_COLLECTION, docs);
+
+ assertNumDocs(110, SOURCE_COLLECTION);
+ assertNumDocs(10, TARGET_COLLECTION);
+
+ // Start again CDCR, the source cluster should reinitialise its log readers
+ // with the latest checkpoints
+
+ this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
+ this.waitForCdcrStateReplication(SOURCE_COLLECTION);
+
+ this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD2);
+
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
+
+ commit(TARGET_COLLECTION);
+
+ assertNumDocs(110, SOURCE_COLLECTION);
+ assertNumDocs(110, TARGET_COLLECTION);
+ }
+
+ /**
+ * Check that the replication manager is properly restarted after a node failure.
+ */
+ @Test
+ @ShardsFixed(num = 4)
+ public void testReplicationAfterRestart() throws Exception {
+ log.info("Starting CDCR");
+
+ // send start action to first shard
+ this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
+ this.waitForCdcrStateReplication(SOURCE_COLLECTION);
+
+ this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD2);
+
+ log.info("Indexing 10 documents");
+
+ int start = 0;
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (; start < 10; start++) {
+ docs.add(getDoc(id, Integer.toString(start)));
+ }
+ index(SOURCE_COLLECTION, docs);
+
+ log.info("Querying source collection");
+
+ assertNumDocs(10, SOURCE_COLLECTION);
+
+ log.info("Waiting for replication");
+
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
+
+ log.info("Querying target collection");
+
+ commit(TARGET_COLLECTION);
+ assertNumDocs(10, TARGET_COLLECTION);
+
+ log.info("Restarting shard1");
+
+ this.restartServers(shardToJetty.get(SOURCE_COLLECTION).get(SHARD1));
+
+ log.info("Indexing 100 documents");
+
+ docs.clear();
+ for (; start < 110; start++) {
+ docs.add(getDoc(id, Integer.toString(start)));
+ }
+ index(SOURCE_COLLECTION, docs);
+
+ log.info("Querying source collection");
+
+ assertNumDocs(110, SOURCE_COLLECTION);
+
+ log.info("Waiting for replication");
+
+ this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD2);
+
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
+
+ log.info("Querying target collection");
+
+ commit(TARGET_COLLECTION);
+ assertNumDocs(110, TARGET_COLLECTION);
+ }
+
+ /**
+ * Check that the replication manager is properly started after a change of leader.
+ * This test also checks that the log readers on the new leaders are initialised with
+ * the target's checkpoint.
+ */
+ @Test
+ @ShardsFixed(num = 4)
+ public void testReplicationAfterLeaderChange() throws Exception {
+ log.info("Starting CDCR");
+
+ // send start action to first shard
+ this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
+ this.waitForCdcrStateReplication(SOURCE_COLLECTION);
+
+ this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD2);
+
+ log.info("Indexing 10 documents");
+
+ int start = 0;
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (; start < 10; start++) {
+ docs.add(getDoc(id, Integer.toString(start)));
+ }
+ index(SOURCE_COLLECTION, docs);
+
+ log.info("Querying source collection");
+
+ assertNumDocs(10, SOURCE_COLLECTION);
+
+ log.info("Waiting for replication");
+
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
+
+ log.info("Querying target collection");
+
+ commit(TARGET_COLLECTION);
+ assertNumDocs(10, TARGET_COLLECTION);
+
+ log.info("Restarting target leaders");
+
+ // Close all the leaders, then restart them
+ this.restartServer(shardToLeaderJetty.get(TARGET_COLLECTION).get(SHARD1));
+ this.restartServer(shardToLeaderJetty.get(TARGET_COLLECTION).get(SHARD2));
+
+ log.info("Restarting source leaders");
+
+ // Close all the leaders, then restart them
+ this.restartServer(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1));
+ this.restartServer(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2));
+
+ log.info("Checking queue size of new source leaders");
+
+ // If the log readers of the new leaders are initialised with the target's checkpoint, the
+ // queue size must be inferior to the current number of documents indexed.
+ // The queue might be not completely empty since the new target checkpoint is probably not the
+ // last document received
+ assertTrue(this.getQueueSize(SOURCE_COLLECTION, SHARD1) < 10);
+ assertTrue(this.getQueueSize(SOURCE_COLLECTION, SHARD2) < 10);
+
+ log.info("Indexing 100 documents");
+
+ docs.clear();
+ for (; start < 110; start++) {
+ docs.add(getDoc(id, Integer.toString(start)));
+ }
+ index(SOURCE_COLLECTION, docs);
+
+ log.info("Querying source collection");
+
+ assertNumDocs(110, SOURCE_COLLECTION);
+
+ log.info("Waiting for replication");
+
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
+
+ log.info("Querying target collection");
+
+ commit(TARGET_COLLECTION);
+ assertNumDocs(110, TARGET_COLLECTION);
+ }
+
+ /**
+ * Check that the update logs are synchronised between leader and non-leader nodes
+ * when CDCR is on and buffer is disabled
+ */
+ @Test
+ @ShardsFixed(num = 4)
+ public void testUpdateLogSynchronisation() throws Exception {
+ this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
+ this.waitForCdcrStateReplication(SOURCE_COLLECTION);
+
+ for (int i = 0; i < 100; i++) {
+ // will perform a commit for every document and will create one tlog file per commit
+ index(SOURCE_COLLECTION, getDoc(id, Integer.toString(i)));
+ }
+
+ this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD2);
+
+ // wait a bit for the replication to complete
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
+
+ commit(TARGET_COLLECTION);
+
+ // Check that the replication was done properly
+ assertNumDocs(100, SOURCE_COLLECTION);
+ assertNumDocs(100, TARGET_COLLECTION);
+
+ // Get the number of tlog files on the replicas (should be equal to the number of documents indexed)
+ int nTlogs = getNumberOfTlogFilesOnReplicas(SOURCE_COLLECTION);
+
+ // Disable the buffer - ulog synch should start on non-leader nodes
+ this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.DISABLEBUFFER);
+ this.waitForCdcrStateReplication(SOURCE_COLLECTION);
+
+ int cnt = 15; // timeout after 15 seconds
+ while (cnt > 0) {
+ // Index a new document with a commit to trigger update log cleaning
+ index(SOURCE_COLLECTION, getDoc(id, Integer.toString(50)));
+
+ // Check the update logs on non-leader nodes, the number of tlog files should decrease
+ int n = getNumberOfTlogFilesOnReplicas(SOURCE_COLLECTION);
+ if (n < nTlogs) return;
+
+ cnt--;
+ Thread.sleep(1000);
+ }
+
+ throw new AssertionError("Timeout while trying to assert update logs @ source_collection");
+ }
+
+ /**
+ * Check that the buffer is always activated on non-leader nodes.
+ */
+ @Test
+ @ShardsFixed(num = 4)
+ public void testBufferOnNonLeader() throws Exception {
+ // buffering is enabled by default, so disable it
+ this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.DISABLEBUFFER);
+ this.waitForCdcrStateReplication(SOURCE_COLLECTION);
+
+ // Start CDCR
+ this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
+ this.waitForCdcrStateReplication(SOURCE_COLLECTION);
+
+ // Index documents
+ for (int i = 0; i < 200; i++) {
+ index(SOURCE_COLLECTION, getDoc(id, Integer.toString(i))); // will perform a commit for every document
+ }
+
+ // And immediately, close all the leaders, then restart them. It is likely that the replication will not be
+ // performed fully, and therefore be continued by the new leader
+ // At this stage, the new leader must have been elected
+ this.restartServer(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1));
+ this.restartServer(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2));
+
+ // wait a bit for the replication to complete
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
+
+ // Commit to make the documents visible on the target
+ commit(TARGET_COLLECTION);
+
+ // If the non-leader node were buffering updates, then the replication must be complete
+ assertNumDocs(200, SOURCE_COLLECTION);
+ assertNumDocs(200, TARGET_COLLECTION);
+ }
+
+ /**
+ * Check the ops statistics.
+ */
+ @Test
+ @ShardsFixed(num = 4)
+ public void testOps() throws Exception {
+ // Index documents
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 200; i++) {
+ docs.add(getDoc(id, Integer.toString(i)));
+ }
+ index(SOURCE_COLLECTION, docs);
+
+ // Start CDCR
+ this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
+ this.waitForCdcrStateReplication(SOURCE_COLLECTION);
+
+ // wait a bit for the replication to complete
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
+
+ NamedList rsp = this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.OPS);
+ NamedList collections = (NamedList) ((NamedList) rsp.get(CdcrParams.OPERATIONS_PER_SECOND)).getVal(0);
+ NamedList ops = (NamedList) collections.get(TARGET_COLLECTION);
+ double opsAll = (Double) ops.get(CdcrParams.COUNTER_ALL);
+ double opsAdds = (Double) ops.get(CdcrParams.COUNTER_ADDS);
+ assertTrue(opsAll > 0);
+ assertEquals(opsAll, opsAdds, 0);
+
+ double opsDeletes = (Double) ops.get(CdcrParams.COUNTER_DELETES);
+ assertEquals(0, opsDeletes, 0);
+ }
+
+ /**
+ * Check that batch updates with deletes
+ */
+ @Test
+ @ShardsFixed(num = 4)
+ public void testBatchAddsWithDelete() throws Exception {
+ // Index 50 documents
+ int start = 0;
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (; start < 50; start++) {
+ docs.add(getDoc(id, Integer.toString(start)));
+ }
+ index(SOURCE_COLLECTION, docs);
+
+ // Delete 10 documents: 10-19
+ List<String> ids = new ArrayList<>();
+ for (int id = 10; id < 20; id++) {
+ ids.add(Integer.toString(id));
+ }
+ deleteById(SOURCE_COLLECTION, ids);
+
+ // Index 10 documents
+ docs = new ArrayList<>();
+ for (; start < 60; start++) {
+ docs.add(getDoc(id, Integer.toString(start)));
+ }
+ index(SOURCE_COLLECTION, docs);
+
+ // Delete 1 document: 50
+ ids = new ArrayList<>();
+ ids.add(Integer.toString(50));
+ deleteById(SOURCE_COLLECTION, ids);
+
+ // Index 10 documents
+ docs = new ArrayList<>();
+ for (; start < 70; start++) {
+ docs.add(getDoc(id, Integer.toString(start)));
+ }
+ index(SOURCE_COLLECTION, docs);
+
+ // Start CDCR
+ this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
+ this.waitForCdcrStateReplication(SOURCE_COLLECTION);
+ this.waitForBootstrapToComplete(TARGET_COLLECTION, SHARD1);
+ this.waitForBootstrapToComplete(TARGET_COLLECTION, SHARD2);
+
+ // wait a bit for the replication to complete
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
+
+ commit(TARGET_COLLECTION);
+
+ // If the non-leader node were buffering updates, then the replication must be complete
+ assertNumDocs(59, SOURCE_COLLECTION);
+ assertNumDocs(59, TARGET_COLLECTION);
+ }
+
+ /**
+ * Checks that batches are correctly constructed when batch boundaries are reached.
+ */
+ @Test
+ @ShardsFixed(num = 4)
+ public void testBatchBoundaries() throws Exception {
+ this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
+ this.waitForCdcrStateReplication(SOURCE_COLLECTION);
+
+ log.info("Indexing documents");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 128; i++) { // should create two full batches (default batch = 64)
+ docs.add(getDoc(id, Integer.toString(i)));
+ }
+ index(SOURCE_COLLECTION, docs);
+
+ assertNumDocs(128, SOURCE_COLLECTION);
+
+ this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD2);
+
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
+
+ commit(TARGET_COLLECTION);
+
+ assertNumDocs(128, SOURCE_COLLECTION);
+ assertNumDocs(128, TARGET_COLLECTION);
+ }
+
+ /**
+ * Check resilience of replication with delete by query executed on targets
+ */
+ @Test
+ @ShardsFixed(num = 4)
+ public void testResilienceWithDeleteByQueryOnTarget() throws Exception {
+ // Index 50 documents
+ int start = 0;
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (; start < 50; start++) {
+ docs.add(getDoc(id, Integer.toString(start)));
+ }
+ index(SOURCE_COLLECTION, docs);
+
+ // Start CDCR
+ this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
+ this.waitForCdcrStateReplication(SOURCE_COLLECTION);
+
+ this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD2);
+
+ // wait a bit for the replication to complete
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
+
+ commit(TARGET_COLLECTION);
+
+ // If the non-leader node were buffering updates, then the replication must be complete
+ assertNumDocs(50, SOURCE_COLLECTION);
+ assertNumDocs(50, TARGET_COLLECTION);
+
+ deleteByQuery(SOURCE_COLLECTION, "*:*");
+ deleteByQuery(TARGET_COLLECTION, "*:*");
+
+ assertNumDocs(0, SOURCE_COLLECTION);
+ assertNumDocs(0, TARGET_COLLECTION);
+
+ docs.clear();
+ for (; start < 100; start++) {
+ docs.add(getDoc(id, Integer.toString(start)));
+ }
+ index(SOURCE_COLLECTION, docs);
+
+ // wait a bit for the replication to complete
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
+
+ commit(TARGET_COLLECTION);
+
+ assertNumDocs(50, SOURCE_COLLECTION);
+ assertNumDocs(50, TARGET_COLLECTION);
+
+ deleteByQuery(TARGET_COLLECTION, "*:*");
+
+ assertNumDocs(50, SOURCE_COLLECTION);
+ assertNumDocs(0, TARGET_COLLECTION);
+
+ // Restart CDCR
+ this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.STOP);
+ this.waitForCdcrStateReplication(SOURCE_COLLECTION);
+ this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
+ this.waitForCdcrStateReplication(SOURCE_COLLECTION);
+
+ docs.clear();
+ for (; start < 150; start++) {
+ docs.add(getDoc(id, Integer.toString(start)));
+ }
+ index(SOURCE_COLLECTION, docs);
+
+ // wait a bit for the replication to complete
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
+ this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
+
+ commit(TARGET_COLLECTION);
+
+ assertNumDocs(100, SOURCE_COLLECTION);
+ assertNumDocs(50, TARGET_COLLECTION);
+ }
+
+ private int numberOfFiles(String dir) {
+ File file = new File(dir);
+ if (!file.isDirectory()) {
+ assertTrue("Path to tlog " + dir + " does not exists or it's not a directory.", false);
+ }
+ log.debug("Update log dir {} contains: {}", dir, file.listFiles());
+ return file.listFiles().length;
+ }
+
+ private int getNumberOfTlogFilesOnReplicas(String collection) throws Exception {
+ CollectionInfo info = collectInfo(collection);
+ Map<String, List<CollectionInfo.CoreInfo>> shardToCoresMap = info.getShardToCoresMap();
+
+ int count = 0;
+
+ for (String shard : shardToCoresMap.keySet()) {
+ for (int i = 0; i < replicationFactor - 1; i++) {
+ count += numberOfFiles(info.getReplicas(shard).get(i).ulogDir);
+ }
+ }
+
+ return count;
+ }
+
+}
+
[3/4] lucene-solr:branch_7x: SOLR-11003: Support bi-directional
syncing of cdcr clusters. We still only support actively into one cluster
cluster,
but have the ability to switch indexing clusters and cdcr will replicate
correctly
Posted by va...@apache.org.
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test/org/apache/solr/cloud/CdcrReplicationDistributedZkTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/CdcrReplicationDistributedZkTest.java b/solr/core/src/test/org/apache/solr/cloud/CdcrReplicationDistributedZkTest.java
deleted file mode 100644
index 35592ff..0000000
--- a/solr/core/src/test/org/apache/solr/cloud/CdcrReplicationDistributedZkTest.java
+++ /dev/null
@@ -1,672 +0,0 @@
-/*
- * 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 org.apache.lucene.util.LuceneTestCase.BadApple;
-import org.apache.lucene.util.LuceneTestCase.Nightly;
-import org.apache.solr.common.SolrInputDocument;
-import org.apache.solr.common.util.NamedList;
-import org.apache.solr.handler.CdcrParams;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
-import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-@Nightly
-@BadApple(bugUrl = "https://issues.apache.org/jira/browse/SOLR-10107")
-public class CdcrReplicationDistributedZkTest extends BaseCdcrDistributedZkTest {
-
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- @Override
- public void distribSetUp() throws Exception {
- schemaString = "schema15.xml"; // we need a string id
- super.distribSetUp();
- }
-
- /**
- * Checks that the test framework handles properly the creation and deletion of collections and the
- * restart of servers.
- */
- @Test
- @ShardsFixed(num = 4)
- public void testDeleteCreateSourceCollection() throws Exception {
- log.info("Indexing documents");
-
- List<SolrInputDocument> docs = new ArrayList<>();
- for (int i = 0; i < 10; i++) {
- docs.add(getDoc(id, Integer.toString(i)));
- }
- index(SOURCE_COLLECTION, docs);
- index(TARGET_COLLECTION, docs);
-
- assertNumDocs(10, SOURCE_COLLECTION);
- assertNumDocs(10, TARGET_COLLECTION);
-
- log.info("Restarting leader @ source_collection:shard1");
-
- this.restartServer(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1));
-
- assertNumDocs(10, SOURCE_COLLECTION);
- assertNumDocs(10, TARGET_COLLECTION);
-
- log.info("Clearing source_collection");
-
- this.clearSourceCollection();
-
- assertNumDocs(0, SOURCE_COLLECTION);
- assertNumDocs(10, TARGET_COLLECTION);
-
- log.info("Restarting leader @ target_collection:shard1");
-
- this.restartServer(shardToLeaderJetty.get(TARGET_COLLECTION).get(SHARD1));
-
- assertNumDocs(0, SOURCE_COLLECTION);
- assertNumDocs(10, TARGET_COLLECTION);
-
- log.info("Clearing target_collection");
-
- this.clearTargetCollection();
-
- assertNumDocs(0, SOURCE_COLLECTION);
- assertNumDocs(0, TARGET_COLLECTION);
-
- assertCollectionExpectations(SOURCE_COLLECTION);
- assertCollectionExpectations(TARGET_COLLECTION);
- }
-
- @Test
- @ShardsFixed(num = 4)
- public void testTargetCollectionNotAvailable() throws Exception {
- // send start action to first shard
- NamedList rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
- NamedList status = (NamedList) rsp.get(CdcrParams.CdcrAction.STATUS.toLower());
- assertEquals(CdcrParams.ProcessState.STARTED.toLower(), status.get(CdcrParams.ProcessState.getParam()));
-
- // check status
- this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STARTED, CdcrParams.BufferState.ENABLED);
-
- this.waitForBootstrapToComplete(TARGET_COLLECTION, SHARD2);
-
- // sleep for a bit to ensure that replicator threads are started
- Thread.sleep(3000);
-
- // Kill all the servers of the target
- this.deleteCollection(TARGET_COLLECTION);
-
- // Index a few documents to trigger the replication
- index(SOURCE_COLLECTION, getDoc(id, "a"));
- index(SOURCE_COLLECTION, getDoc(id, "b"));
- index(SOURCE_COLLECTION, getDoc(id, "c"));
- index(SOURCE_COLLECTION, getDoc(id, "d"));
- index(SOURCE_COLLECTION, getDoc(id, "e"));
- index(SOURCE_COLLECTION, getDoc(id, "f"));
-
- assertNumDocs(6, SOURCE_COLLECTION);
-
- // we need to wait until the replicator thread is triggered
- int cnt = 15; // timeout after 15 seconds
- AssertionError lastAssertionError = null;
- while (cnt > 0) {
- try {
- rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.ERRORS);
- NamedList collections = (NamedList) ((NamedList) rsp.get(CdcrParams.ERRORS)).getVal(0);
- NamedList errors = (NamedList) collections.get(TARGET_COLLECTION);
- assertTrue(0 < (Long) errors.get(CdcrParams.CONSECUTIVE_ERRORS));
- NamedList lastErrors = (NamedList) errors.get(CdcrParams.LAST);
- assertNotNull(lastErrors);
- assertTrue(0 < lastErrors.size());
- return;
- }
- catch (AssertionError e) {
- lastAssertionError = e;
- cnt--;
- Thread.sleep(1000);
- }
- }
-
- throw new AssertionError("Timeout while trying to assert replication errors", lastAssertionError);
- }
-
- @Test
- @ShardsFixed(num = 4)
- public void testReplicationStartStop() throws Exception {
- int start = 0;
- List<SolrInputDocument> docs = new ArrayList<>();
- for (; start < 10; start++) {
- docs.add(getDoc(id, Integer.toString(start)));
- }
- index(SOURCE_COLLECTION, docs);
-
- assertNumDocs(10, SOURCE_COLLECTION);
- assertNumDocs(0, TARGET_COLLECTION);
-
- this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
- this.waitForCdcrStateReplication(SOURCE_COLLECTION);
-
- this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD2);
-
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
-
- commit(TARGET_COLLECTION);
-
- assertNumDocs(10, SOURCE_COLLECTION);
- assertNumDocs(10, TARGET_COLLECTION);
-
- this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.STOP);
- this.waitForCdcrStateReplication(SOURCE_COLLECTION);
-
- docs.clear();
- for (; start < 110; start++) {
- docs.add(getDoc(id, Integer.toString(start)));
- }
- index(SOURCE_COLLECTION, docs);
-
- assertNumDocs(110, SOURCE_COLLECTION);
- assertNumDocs(10, TARGET_COLLECTION);
-
- // Start again CDCR, the source cluster should reinitialise its log readers
- // with the latest checkpoints
-
- this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
- this.waitForCdcrStateReplication(SOURCE_COLLECTION);
-
- this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD2);
-
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
-
- commit(TARGET_COLLECTION);
-
- assertNumDocs(110, SOURCE_COLLECTION);
- assertNumDocs(110, TARGET_COLLECTION);
- }
-
- /**
- * Check that the replication manager is properly restarted after a node failure.
- */
- @Test
- @ShardsFixed(num = 4)
- public void testReplicationAfterRestart() throws Exception {
- log.info("Starting CDCR");
-
- // send start action to first shard
- this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
- this.waitForCdcrStateReplication(SOURCE_COLLECTION);
-
- this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD2);
-
- log.info("Indexing 10 documents");
-
- int start = 0;
- List<SolrInputDocument> docs = new ArrayList<>();
- for (; start < 10; start++) {
- docs.add(getDoc(id, Integer.toString(start)));
- }
- index(SOURCE_COLLECTION, docs);
-
- log.info("Querying source collection");
-
- assertNumDocs(10, SOURCE_COLLECTION);
-
- log.info("Waiting for replication");
-
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
-
- log.info("Querying target collection");
-
- commit(TARGET_COLLECTION);
- assertNumDocs(10, TARGET_COLLECTION);
-
- log.info("Restarting shard1");
-
- this.restartServers(shardToJetty.get(SOURCE_COLLECTION).get(SHARD1));
-
- log.info("Indexing 100 documents");
-
- docs.clear();
- for (; start < 110; start++) {
- docs.add(getDoc(id, Integer.toString(start)));
- }
- index(SOURCE_COLLECTION, docs);
-
- log.info("Querying source collection");
-
- assertNumDocs(110, SOURCE_COLLECTION);
-
- log.info("Waiting for replication");
-
- this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD2);
-
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
-
- log.info("Querying target collection");
-
- commit(TARGET_COLLECTION);
- assertNumDocs(110, TARGET_COLLECTION);
- }
-
- /**
- * Check that the replication manager is properly started after a change of leader.
- * This test also checks that the log readers on the new leaders are initialised with
- * the target's checkpoint.
- */
- @Test
- @ShardsFixed(num = 4)
- public void testReplicationAfterLeaderChange() throws Exception {
- log.info("Starting CDCR");
-
- // send start action to first shard
- this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
- this.waitForCdcrStateReplication(SOURCE_COLLECTION);
-
- this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD2);
-
- log.info("Indexing 10 documents");
-
- int start = 0;
- List<SolrInputDocument> docs = new ArrayList<>();
- for (; start < 10; start++) {
- docs.add(getDoc(id, Integer.toString(start)));
- }
- index(SOURCE_COLLECTION, docs);
-
- log.info("Querying source collection");
-
- assertNumDocs(10, SOURCE_COLLECTION);
-
- log.info("Waiting for replication");
-
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
-
- log.info("Querying target collection");
-
- commit(TARGET_COLLECTION);
- assertNumDocs(10, TARGET_COLLECTION);
-
- log.info("Restarting target leaders");
-
- // Close all the leaders, then restart them
- this.restartServer(shardToLeaderJetty.get(TARGET_COLLECTION).get(SHARD1));
- this.restartServer(shardToLeaderJetty.get(TARGET_COLLECTION).get(SHARD2));
-
- log.info("Restarting source leaders");
-
- // Close all the leaders, then restart them
- this.restartServer(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1));
- this.restartServer(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2));
-
- log.info("Checking queue size of new source leaders");
-
- // If the log readers of the new leaders are initialised with the target's checkpoint, the
- // queue size must be inferior to the current number of documents indexed.
- // The queue might be not completely empty since the new target checkpoint is probably not the
- // last document received
- assertTrue(this.getQueueSize(SOURCE_COLLECTION, SHARD1) < 10);
- assertTrue(this.getQueueSize(SOURCE_COLLECTION, SHARD2) < 10);
-
- log.info("Indexing 100 documents");
-
- docs.clear();
- for (; start < 110; start++) {
- docs.add(getDoc(id, Integer.toString(start)));
- }
- index(SOURCE_COLLECTION, docs);
-
- log.info("Querying source collection");
-
- assertNumDocs(110, SOURCE_COLLECTION);
-
- log.info("Waiting for replication");
-
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
-
- log.info("Querying target collection");
-
- commit(TARGET_COLLECTION);
- assertNumDocs(110, TARGET_COLLECTION);
- }
-
- /**
- * Check that the update logs are synchronised between leader and non-leader nodes
- * when CDCR is on and buffer is disabled
- */
- @Test
- @ShardsFixed(num = 4)
- public void testUpdateLogSynchronisation() throws Exception {
- this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
- this.waitForCdcrStateReplication(SOURCE_COLLECTION);
-
- for (int i = 0; i < 100; i++) {
- // will perform a commit for every document and will create one tlog file per commit
- index(SOURCE_COLLECTION, getDoc(id, Integer.toString(i)));
- }
-
- this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD2);
-
- // wait a bit for the replication to complete
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
-
- commit(TARGET_COLLECTION);
-
- // Check that the replication was done properly
- assertNumDocs(100, SOURCE_COLLECTION);
- assertNumDocs(100, TARGET_COLLECTION);
-
- // Get the number of tlog files on the replicas (should be equal to the number of documents indexed)
- int nTlogs = getNumberOfTlogFilesOnReplicas(SOURCE_COLLECTION);
-
- // Disable the buffer - ulog synch should start on non-leader nodes
- this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.DISABLEBUFFER);
- this.waitForCdcrStateReplication(SOURCE_COLLECTION);
-
- int cnt = 15; // timeout after 15 seconds
- while (cnt > 0) {
- // Index a new document with a commit to trigger update log cleaning
- index(SOURCE_COLLECTION, getDoc(id, Integer.toString(50)));
-
- // Check the update logs on non-leader nodes, the number of tlog files should decrease
- int n = getNumberOfTlogFilesOnReplicas(SOURCE_COLLECTION);
- if (n < nTlogs) return;
-
- cnt--;
- Thread.sleep(1000);
- }
-
- throw new AssertionError("Timeout while trying to assert update logs @ source_collection");
- }
-
- /**
- * Check that the buffer is always activated on non-leader nodes.
- */
- @Test
- @ShardsFixed(num = 4)
- public void testBufferOnNonLeader() throws Exception {
- // buffering is enabled by default, so disable it
- this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.DISABLEBUFFER);
- this.waitForCdcrStateReplication(SOURCE_COLLECTION);
-
- // Start CDCR
- this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
- this.waitForCdcrStateReplication(SOURCE_COLLECTION);
-
- // Index documents
- for (int i = 0; i < 200; i++) {
- index(SOURCE_COLLECTION, getDoc(id, Integer.toString(i))); // will perform a commit for every document
- }
-
- // And immediately, close all the leaders, then restart them. It is likely that the replication will not be
- // performed fully, and therefore be continued by the new leader
- // At this stage, the new leader must have been elected
- this.restartServer(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1));
- this.restartServer(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2));
-
- // wait a bit for the replication to complete
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
-
- // Commit to make the documents visible on the target
- commit(TARGET_COLLECTION);
-
- // If the non-leader node were buffering updates, then the replication must be complete
- assertNumDocs(200, SOURCE_COLLECTION);
- assertNumDocs(200, TARGET_COLLECTION);
- }
-
- /**
- * Check the ops statistics.
- */
- @Test
- @ShardsFixed(num = 4)
- public void testOps() throws Exception {
- // Index documents
- List<SolrInputDocument> docs = new ArrayList<>();
- for (int i = 0; i < 200; i++) {
- docs.add(getDoc(id, Integer.toString(i)));
- }
- index(SOURCE_COLLECTION, docs);
-
- // Start CDCR
- this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
- this.waitForCdcrStateReplication(SOURCE_COLLECTION);
-
- // wait a bit for the replication to complete
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
-
- NamedList rsp = this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.OPS);
- NamedList collections = (NamedList) ((NamedList) rsp.get(CdcrParams.OPERATIONS_PER_SECOND)).getVal(0);
- NamedList ops = (NamedList) collections.get(TARGET_COLLECTION);
- double opsAll = (Double) ops.get(CdcrParams.COUNTER_ALL);
- double opsAdds = (Double) ops.get(CdcrParams.COUNTER_ADDS);
- assertTrue(opsAll > 0);
- assertEquals(opsAll, opsAdds, 0);
-
- double opsDeletes = (Double) ops.get(CdcrParams.COUNTER_DELETES);
- assertEquals(0, opsDeletes, 0);
- }
-
- /**
- * Check that batch updates with deletes
- */
- @Test
- @ShardsFixed(num = 4)
- public void testBatchAddsWithDelete() throws Exception {
- // Index 50 documents
- int start = 0;
- List<SolrInputDocument> docs = new ArrayList<>();
- for (; start < 50; start++) {
- docs.add(getDoc(id, Integer.toString(start)));
- }
- index(SOURCE_COLLECTION, docs);
-
- // Delete 10 documents: 10-19
- List<String> ids = new ArrayList<>();
- for (int id = 10; id < 20; id++) {
- ids.add(Integer.toString(id));
- }
- deleteById(SOURCE_COLLECTION, ids);
-
- // Index 10 documents
- docs = new ArrayList<>();
- for (; start < 60; start++) {
- docs.add(getDoc(id, Integer.toString(start)));
- }
- index(SOURCE_COLLECTION, docs);
-
- // Delete 1 document: 50
- ids = new ArrayList<>();
- ids.add(Integer.toString(50));
- deleteById(SOURCE_COLLECTION, ids);
-
- // Index 10 documents
- docs = new ArrayList<>();
- for (; start < 70; start++) {
- docs.add(getDoc(id, Integer.toString(start)));
- }
- index(SOURCE_COLLECTION, docs);
-
- // Start CDCR
- this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
- this.waitForCdcrStateReplication(SOURCE_COLLECTION);
- this.waitForBootstrapToComplete(TARGET_COLLECTION, SHARD1);
- this.waitForBootstrapToComplete(TARGET_COLLECTION, SHARD2);
-
- // wait a bit for the replication to complete
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
-
- commit(TARGET_COLLECTION);
-
- // If the non-leader node were buffering updates, then the replication must be complete
- assertNumDocs(59, SOURCE_COLLECTION);
- assertNumDocs(59, TARGET_COLLECTION);
- }
-
- /**
- * Checks that batches are correctly constructed when batch boundaries are reached.
- */
- @Test
- @ShardsFixed(num = 4)
- public void testBatchBoundaries() throws Exception {
- this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
- this.waitForCdcrStateReplication(SOURCE_COLLECTION);
-
- log.info("Indexing documents");
-
- List<SolrInputDocument> docs = new ArrayList<>();
- for (int i = 0; i < 128; i++) { // should create two full batches (default batch = 64)
- docs.add(getDoc(id, Integer.toString(i)));
- }
- index(SOURCE_COLLECTION, docs);
-
- assertNumDocs(128, SOURCE_COLLECTION);
-
- this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD2);
-
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
-
- commit(TARGET_COLLECTION);
-
- assertNumDocs(128, SOURCE_COLLECTION);
- assertNumDocs(128, TARGET_COLLECTION);
- }
-
- /**
- * Check resilience of replication with delete by query executed on targets
- */
- @Test
- @ShardsFixed(num = 4)
- public void testResilienceWithDeleteByQueryOnTarget() throws Exception {
- // Index 50 documents
- int start = 0;
- List<SolrInputDocument> docs = new ArrayList<>();
- for (; start < 50; start++) {
- docs.add(getDoc(id, Integer.toString(start)));
- }
- index(SOURCE_COLLECTION, docs);
-
- // Start CDCR
- this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
- this.waitForCdcrStateReplication(SOURCE_COLLECTION);
-
- this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForBootstrapToComplete(SOURCE_COLLECTION, SHARD2);
-
- // wait a bit for the replication to complete
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
-
- commit(TARGET_COLLECTION);
-
- // If the non-leader node were buffering updates, then the replication must be complete
- assertNumDocs(50, SOURCE_COLLECTION);
- assertNumDocs(50, TARGET_COLLECTION);
-
- deleteByQuery(SOURCE_COLLECTION, "*:*");
- deleteByQuery(TARGET_COLLECTION, "*:*");
-
- assertNumDocs(0, SOURCE_COLLECTION);
- assertNumDocs(0, TARGET_COLLECTION);
-
- docs.clear();
- for (; start < 100; start++) {
- docs.add(getDoc(id, Integer.toString(start)));
- }
- index(SOURCE_COLLECTION, docs);
-
- // wait a bit for the replication to complete
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
-
- commit(TARGET_COLLECTION);
-
- assertNumDocs(50, SOURCE_COLLECTION);
- assertNumDocs(50, TARGET_COLLECTION);
-
- deleteByQuery(TARGET_COLLECTION, "*:*");
-
- assertNumDocs(50, SOURCE_COLLECTION);
- assertNumDocs(0, TARGET_COLLECTION);
-
- // Restart CDCR
- this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.STOP);
- this.waitForCdcrStateReplication(SOURCE_COLLECTION);
- this.invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
- this.waitForCdcrStateReplication(SOURCE_COLLECTION);
-
- docs.clear();
- for (; start < 150; start++) {
- docs.add(getDoc(id, Integer.toString(start)));
- }
- index(SOURCE_COLLECTION, docs);
-
- // wait a bit for the replication to complete
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD1);
- this.waitForReplicationToComplete(SOURCE_COLLECTION, SHARD2);
-
- commit(TARGET_COLLECTION);
-
- assertNumDocs(100, SOURCE_COLLECTION);
- assertNumDocs(50, TARGET_COLLECTION);
- }
-
- private int numberOfFiles(String dir) {
- File file = new File(dir);
- if (!file.isDirectory()) {
- assertTrue("Path to tlog " + dir + " does not exists or it's not a directory.", false);
- }
- log.debug("Update log dir {} contains: {}", dir, file.listFiles());
- return file.listFiles().length;
- }
-
- private int getNumberOfTlogFilesOnReplicas(String collection) throws Exception {
- CollectionInfo info = collectInfo(collection);
- Map<String, List<CollectionInfo.CoreInfo>> shardToCoresMap = info.getShardToCoresMap();
-
- int count = 0;
-
- for (String shard : shardToCoresMap.keySet()) {
- for (int i = 0; i < replicationFactor - 1; i++) {
- count += numberOfFiles(info.getReplicas(shard).get(i).ulogDir);
- }
- }
-
- return count;
- }
-
-}
-
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test/org/apache/solr/cloud/CdcrReplicationHandlerTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/CdcrReplicationHandlerTest.java b/solr/core/src/test/org/apache/solr/cloud/CdcrReplicationHandlerTest.java
deleted file mode 100644
index 1663949..0000000
--- a/solr/core/src/test/org/apache/solr/cloud/CdcrReplicationHandlerTest.java
+++ /dev/null
@@ -1,330 +0,0 @@
-/*
- * 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 org.apache.lucene.util.LuceneTestCase.Nightly;
-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.impl.CloudSolrClient;
-import org.apache.solr.common.SolrInputDocument;
-import org.apache.solr.util.DefaultSolrThreadFactory;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
-import java.io.IOException;
-import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
-/**
- * This class is testing the cdcr extension to the {@link org.apache.solr.handler.ReplicationHandler} and
- * {@link org.apache.solr.handler.IndexFetcher}.
- */
-@Nightly
-public class CdcrReplicationHandlerTest extends BaseCdcrDistributedZkTest {
-
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- @Override
- public void distribSetUp() throws Exception {
- schemaString = "schema15.xml"; // we need a string id
- createTargetCollection = false; // we do not need the target cluster
- shardCount = 1; // we need only one shard
- // we need a persistent directory, otherwise the UpdateHandler will erase existing tlog files after restarting a node
- System.setProperty("solr.directoryFactory", "solr.StandardDirectoryFactory");
- super.distribSetUp();
- }
-
- /**
- * Test the scenario where the slave is killed from the start. The replication
- * strategy should fetch all the missing tlog files from the leader.
- */
- @Test
- @ShardsFixed(num = 2)
- public void testFullReplication() throws Exception {
- List<CloudJettyRunner> slaves = this.getShardToSlaveJetty(SOURCE_COLLECTION, SHARD1);
- ChaosMonkey.stop(slaves.get(0).jetty);
-
- for (int i = 0; i < 10; i++) {
- List<SolrInputDocument> docs = new ArrayList<>();
- for (int j = i * 10; j < (i * 10) + 10; j++) {
- docs.add(getDoc(id, Integer.toString(j)));
- }
- index(SOURCE_COLLECTION, docs);
- }
-
- assertNumDocs(100, SOURCE_COLLECTION);
-
- // Restart the slave node to trigger Replication strategy
- this.restartServer(slaves.get(0));
-
- this.assertUpdateLogsEquals(SOURCE_COLLECTION, 10);
- }
-
- /**
- * Test the scenario where the slave is killed before receiving all the documents. The replication
- * strategy should fetch all the missing tlog files from the leader.
- */
- @Test
- @ShardsFixed(num = 2)
- public void testPartialReplication() throws Exception {
- for (int i = 0; i < 5; i++) {
- List<SolrInputDocument> docs = new ArrayList<>();
- for (int j = i * 20; j < (i * 20) + 20; j++) {
- docs.add(getDoc(id, Integer.toString(j)));
- }
- index(SOURCE_COLLECTION, docs);
- }
-
- List<CloudJettyRunner> slaves = this.getShardToSlaveJetty(SOURCE_COLLECTION, SHARD1);
- ChaosMonkey.stop(slaves.get(0).jetty);
-
- for (int i = 5; i < 10; i++) {
- List<SolrInputDocument> docs = new ArrayList<>();
- for (int j = i * 20; j < (i * 20) + 20; j++) {
- docs.add(getDoc(id, Integer.toString(j)));
- }
- index(SOURCE_COLLECTION, docs);
- }
-
- assertNumDocs(200, SOURCE_COLLECTION);
-
- // Restart the slave node to trigger Replication strategy
- this.restartServer(slaves.get(0));
-
- // at this stage, the slave should have replicated the 5 missing tlog files
- this.assertUpdateLogsEquals(SOURCE_COLLECTION, 10);
- }
-
- /**
- * Test the scenario where the slave is killed before receiving a commit. This creates a truncated tlog
- * file on the slave node. The replication strategy should detect this truncated file, and fetch the
- * non-truncated file from the leader.
- */
- @Test
- @ShardsFixed(num = 2)
- public void testPartialReplicationWithTruncatedTlog() throws Exception {
- CloudSolrClient client = createCloudClient(SOURCE_COLLECTION);
- List<CloudJettyRunner> slaves = this.getShardToSlaveJetty(SOURCE_COLLECTION, SHARD1);
-
- try {
- for (int i = 0; i < 10; i++) {
- for (int j = i * 20; j < (i * 20) + 20; j++) {
- client.add(getDoc(id, Integer.toString(j)));
-
- // Stop the slave in the middle of a batch to create a truncated tlog on the slave
- if (j == 45) {
- ChaosMonkey.stop(slaves.get(0).jetty);
- }
-
- }
- commit(SOURCE_COLLECTION);
- }
- } finally {
- client.close();
- }
-
- assertNumDocs(200, SOURCE_COLLECTION);
-
- // Restart the slave node to trigger Replication recovery
- this.restartServer(slaves.get(0));
-
- // at this stage, the slave should have replicated the 5 missing tlog files
- this.assertUpdateLogsEquals(SOURCE_COLLECTION, 10);
- }
-
- /**
- * Test the scenario where the slave first recovered with a PeerSync strategy, then with a Replication strategy.
- * The PeerSync strategy will generate a single tlog file for all the missing updates on the slave node.
- * If a Replication strategy occurs at a later stage, it should remove this tlog file generated by PeerSync
- * and fetch the corresponding tlog files from the leader.
- */
- @Test
- @ShardsFixed(num = 2)
- public void testPartialReplicationAfterPeerSync() throws Exception {
- for (int i = 0; i < 5; i++) {
- List<SolrInputDocument> docs = new ArrayList<>();
- for (int j = i * 10; j < (i * 10) + 10; j++) {
- docs.add(getDoc(id, Integer.toString(j)));
- }
- index(SOURCE_COLLECTION, docs);
- }
-
- List<CloudJettyRunner> slaves = this.getShardToSlaveJetty(SOURCE_COLLECTION, SHARD1);
- ChaosMonkey.stop(slaves.get(0).jetty);
-
- for (int i = 5; i < 10; i++) {
- List<SolrInputDocument> docs = new ArrayList<>();
- for (int j = i * 10; j < (i * 10) + 10; j++) {
- docs.add(getDoc(id, Integer.toString(j)));
- }
- index(SOURCE_COLLECTION, docs);
- }
-
- assertNumDocs(100, SOURCE_COLLECTION);
-
- // Restart the slave node to trigger PeerSync recovery
- // (the update windows between leader and slave is small enough)
- this.restartServer(slaves.get(0));
-
- ChaosMonkey.stop(slaves.get(0).jetty);
-
- for (int i = 10; i < 15; i++) {
- List<SolrInputDocument> docs = new ArrayList<>();
- for (int j = i * 20; j < (i * 20) + 20; j++) {
- docs.add(getDoc(id, Integer.toString(j)));
- }
- index(SOURCE_COLLECTION, docs);
- }
-
- // restart the slave node to trigger Replication recovery
- this.restartServer(slaves.get(0));
-
- // at this stage, the slave should have replicated the 5 missing tlog files
- this.assertUpdateLogsEquals(SOURCE_COLLECTION, 15);
- }
-
- /**
- * Test the scenario where the slave is killed while the leader is still receiving updates.
- * The slave should buffer updates while in recovery, then replay them at the end of the recovery.
- * If updates were properly buffered and replayed, then the slave should have the same number of documents
- * than the leader. This checks if cdcr tlog replication interferes with buffered updates - SOLR-8263.
- */
- @Test
- @ShardsFixed(num = 2)
- public void testReplicationWithBufferedUpdates() throws Exception {
- List<CloudJettyRunner> slaves = this.getShardToSlaveJetty(SOURCE_COLLECTION, SHARD1);
-
- AtomicInteger numDocs = new AtomicInteger(0);
- ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(new DefaultSolrThreadFactory("cdcr-test-update-scheduler"));
- executor.scheduleWithFixedDelay(new UpdateThread(numDocs), 10, 10, TimeUnit.MILLISECONDS);
-
- // Restart the slave node to trigger Replication strategy
- this.restartServer(slaves.get(0));
-
- // shutdown the update thread and wait for its completion
- executor.shutdown();
- executor.awaitTermination(500, TimeUnit.MILLISECONDS);
-
- // check that we have the expected number of documents in the cluster
- assertNumDocs(numDocs.get(), SOURCE_COLLECTION);
-
- // check that we have the expected number of documents on the slave
- assertNumDocs(numDocs.get(), slaves.get(0));
- }
-
- private void assertNumDocs(int expectedNumDocs, CloudJettyRunner jetty)
- throws InterruptedException, IOException, SolrServerException {
- SolrClient client = createNewSolrServer(jetty.url);
- try {
- int cnt = 30; // timeout after 15 seconds
- AssertionError lastAssertionError = null;
- while (cnt > 0) {
- try {
- assertEquals(expectedNumDocs, client.query(new SolrQuery("*:*")).getResults().getNumFound());
- return;
- }
- catch (AssertionError e) {
- lastAssertionError = e;
- cnt--;
- Thread.sleep(500);
- }
- }
- throw new AssertionError("Timeout while trying to assert number of documents @ " + jetty.url, lastAssertionError);
- } finally {
- client.close();
- }
- }
-
- private class UpdateThread implements Runnable {
-
- private AtomicInteger numDocs;
-
- private UpdateThread(AtomicInteger numDocs) {
- this.numDocs = numDocs;
- }
-
- @Override
- public void run() {
- try {
- List<SolrInputDocument> docs = new ArrayList<>();
- for (int j = numDocs.get(); j < (numDocs.get() + 10); j++) {
- docs.add(getDoc(id, Integer.toString(j)));
- }
- index(SOURCE_COLLECTION, docs);
- numDocs.getAndAdd(10);
- log.info("Sent batch of {} updates - numDocs:{}", docs.size(), numDocs);
- }
- catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- }
-
- private List<CloudJettyRunner> getShardToSlaveJetty(String collection, String shard) {
- List<CloudJettyRunner> jetties = new ArrayList<>(shardToJetty.get(collection).get(shard));
- CloudJettyRunner leader = shardToLeaderJetty.get(collection).get(shard);
- jetties.remove(leader);
- return jetties;
- }
-
- /**
- * Asserts that the update logs are in sync between the leader and slave. The leader and the slaves
- * must have identical tlog files.
- */
- protected void assertUpdateLogsEquals(String collection, int numberOfTLogs) throws Exception {
- CollectionInfo info = collectInfo(collection);
- Map<String, List<CollectionInfo.CoreInfo>> shardToCoresMap = info.getShardToCoresMap();
-
- for (String shard : shardToCoresMap.keySet()) {
- Map<Long, Long> leaderFilesMeta = this.getFilesMeta(info.getLeader(shard).ulogDir);
- Map<Long, Long> slaveFilesMeta = this.getFilesMeta(info.getReplicas(shard).get(0).ulogDir);
-
- assertEquals("Incorrect number of tlog files on the leader", numberOfTLogs, leaderFilesMeta.size());
- assertEquals("Incorrect number of tlog files on the slave", numberOfTLogs, slaveFilesMeta.size());
-
- for (Long leaderFileVersion : leaderFilesMeta.keySet()) {
- assertTrue("Slave is missing a tlog for version " + leaderFileVersion, slaveFilesMeta.containsKey(leaderFileVersion));
- assertEquals("Slave's tlog file size differs for version " + leaderFileVersion, leaderFilesMeta.get(leaderFileVersion), slaveFilesMeta.get(leaderFileVersion));
- }
- }
- }
-
- private Map<Long, Long> getFilesMeta(String dir) {
- File file = new File(dir);
- if (!file.isDirectory()) {
- assertTrue("Path to tlog " + dir + " does not exists or it's not a directory.", false);
- }
-
- Map<Long, Long> filesMeta = new HashMap<>();
- for (File tlogFile : file.listFiles()) {
- filesMeta.put(Math.abs(Long.parseLong(tlogFile.getName().substring(tlogFile.getName().lastIndexOf('.') + 1))), tlogFile.length());
- }
- return filesMeta;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test/org/apache/solr/cloud/CdcrRequestHandlerTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/CdcrRequestHandlerTest.java b/solr/core/src/test/org/apache/solr/cloud/CdcrRequestHandlerTest.java
deleted file mode 100644
index c74079c..0000000
--- a/solr/core/src/test/org/apache/solr/cloud/CdcrRequestHandlerTest.java
+++ /dev/null
@@ -1,153 +0,0 @@
-/*
- * 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 org.apache.lucene.util.LuceneTestCase.Nightly;
-import org.apache.solr.common.util.NamedList;
-import org.apache.solr.handler.CdcrParams;
-import org.junit.Test;
-
-@Nightly
-public class CdcrRequestHandlerTest extends BaseCdcrDistributedZkTest {
-
- @Override
- public void distribSetUp() throws Exception {
- schemaString = "schema15.xml"; // we need a string id
- createTargetCollection = false; // we do not need the target cluster
- super.distribSetUp();
- }
-
- // check that the life-cycle state is properly synchronised across nodes
- @Test
- @ShardsFixed(num = 2)
- public void testLifeCycleActions() throws Exception {
- // check initial status
- this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STOPPED, CdcrParams.BufferState.ENABLED);
-
- // send start action to first shard
- NamedList rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.START);
- NamedList status = (NamedList) rsp.get(CdcrParams.CdcrAction.STATUS.toLower());
- assertEquals(CdcrParams.ProcessState.STARTED.toLower(), status.get(CdcrParams.ProcessState.getParam()));
-
- // check status
- this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STARTED, CdcrParams.BufferState.ENABLED);
-
- // Restart the leader of shard 1
- this.restartServer(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1));
-
- // check status - the node that died should have picked up the original state
- this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STARTED, CdcrParams.BufferState.ENABLED);
-
- // send stop action to second shard
- rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.STOP);
- status = (NamedList) rsp.get(CdcrParams.CdcrAction.STATUS.toLower());
- assertEquals(CdcrParams.ProcessState.STOPPED.toLower(), status.get(CdcrParams.ProcessState.getParam()));
-
- // check status
- this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STOPPED, CdcrParams.BufferState.ENABLED);
- }
-
- // check the checkpoint API
- @Test
- @ShardsFixed(num = 2)
- public void testCheckpointActions() throws Exception {
- // initial request on an empty index, must return -1
- NamedList rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.COLLECTIONCHECKPOINT);
- assertEquals(-1l, rsp.get(CdcrParams.CHECKPOINT));
-
- index(SOURCE_COLLECTION, getDoc(id, "a")); // shard 2
-
- // only one document indexed in shard 2, the checkpoint must be still -1
- rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.COLLECTIONCHECKPOINT);
- assertEquals(-1l, rsp.get(CdcrParams.CHECKPOINT));
-
- index(SOURCE_COLLECTION, getDoc(id, "b")); // shard 1
-
- // a second document indexed in shard 1, the checkpoint must come from shard 2
- rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.COLLECTIONCHECKPOINT);
- long checkpoint1 = (Long) rsp.get(CdcrParams.CHECKPOINT);
- long expected = (Long) invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.SHARDCHECKPOINT).get(CdcrParams.CHECKPOINT);
- assertEquals(expected, checkpoint1);
-
- index(SOURCE_COLLECTION, getDoc(id, "c")); // shard 1
-
- // a third document indexed in shard 1, the checkpoint must still come from shard 2
- rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.COLLECTIONCHECKPOINT);
- assertEquals(checkpoint1, rsp.get(CdcrParams.CHECKPOINT));
-
- index(SOURCE_COLLECTION, getDoc(id, "d")); // shard 2
-
- // a fourth document indexed in shard 2, the checkpoint must come from shard 1
- rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.COLLECTIONCHECKPOINT);
- long checkpoint2 = (Long) rsp.get(CdcrParams.CHECKPOINT);
- expected = (Long) invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.SHARDCHECKPOINT).get(CdcrParams.CHECKPOINT);
- assertEquals(expected, checkpoint2);
-
- // send a delete by query
- deleteByQuery(SOURCE_COLLECTION, "*:*");
-
- // all the checkpoints must come from the DBQ
- rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.COLLECTIONCHECKPOINT);
- long checkpoint3 = (Long) rsp.get(CdcrParams.CHECKPOINT);
- assertTrue(checkpoint3 > 0); // ensure that checkpoints from deletes are in absolute form
- checkpoint3 = (Long) invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.SHARDCHECKPOINT).get(CdcrParams.CHECKPOINT);
- assertTrue(checkpoint3 > 0); // ensure that checkpoints from deletes are in absolute form
- checkpoint3 = (Long) invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.SHARDCHECKPOINT).get(CdcrParams.CHECKPOINT);
- assertTrue(checkpoint3 > 0); // ensure that checkpoints from deletes are in absolute form
-
- // replication never started, lastProcessedVersion should be -1 for both shards
- rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.LASTPROCESSEDVERSION);
- long lastVersion = (Long) rsp.get(CdcrParams.LAST_PROCESSED_VERSION);
- assertEquals(-1l, lastVersion);
-
- rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.LASTPROCESSEDVERSION);
- lastVersion = (Long) rsp.get(CdcrParams.LAST_PROCESSED_VERSION);
- assertEquals(-1l, lastVersion);
- }
-
- // check that the buffer state is properly synchronised across nodes
- @Test
- @ShardsFixed(num = 2)
- public void testBufferActions() throws Exception {
- // check initial status
- this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STOPPED, CdcrParams.BufferState.ENABLED);
-
- // send disable buffer action to first shard
- NamedList rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1), CdcrParams.CdcrAction.DISABLEBUFFER);
- NamedList status = (NamedList) rsp.get(CdcrParams.CdcrAction.STATUS.toLower());
- assertEquals(CdcrParams.BufferState.DISABLED.toLower(), status.get(CdcrParams.BufferState.getParam()));
-
- // check status
- this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STOPPED, CdcrParams.BufferState.DISABLED);
-
- // Restart the leader of shard 1
- this.restartServer(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD1));
-
- // check status
- this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STOPPED, CdcrParams.BufferState.DISABLED);
-
- // send enable buffer action to second shard
- rsp = invokeCdcrAction(shardToLeaderJetty.get(SOURCE_COLLECTION).get(SHARD2), CdcrParams.CdcrAction.ENABLEBUFFER);
- status = (NamedList) rsp.get(CdcrParams.CdcrAction.STATUS.toLower());
- assertEquals(CdcrParams.BufferState.ENABLED.toLower(), status.get(CdcrParams.BufferState.getParam()));
-
- // check status
- this.assertState(SOURCE_COLLECTION, CdcrParams.ProcessState.STOPPED, CdcrParams.BufferState.ENABLED);
- }
-
-}
-
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test/org/apache/solr/cloud/CdcrVersionReplicationTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/CdcrVersionReplicationTest.java b/solr/core/src/test/org/apache/solr/cloud/CdcrVersionReplicationTest.java
deleted file mode 100644
index 59d3818..0000000
--- a/solr/core/src/test/org/apache/solr/cloud/CdcrVersionReplicationTest.java
+++ /dev/null
@@ -1,307 +0,0 @@
-/*
- * 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 org.apache.solr.client.solrj.SolrClient;
-import org.apache.solr.client.solrj.SolrServerException;
-import org.apache.solr.client.solrj.impl.CloudSolrClient;
-import org.apache.solr.client.solrj.request.UpdateRequest;
-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.params.CommonParams;
-import org.apache.solr.common.util.StrUtils;
-import org.apache.solr.update.processor.CdcrUpdateProcessor;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.lang.invoke.MethodHandles;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-
-public class CdcrVersionReplicationTest extends BaseCdcrDistributedZkTest {
-
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- private static final String vfield = CommonParams.VERSION_FIELD;
- SolrClient solrServer;
-
- public CdcrVersionReplicationTest() {
- schemaString = "schema15.xml"; // we need a string id
- super.createTargetCollection = false;
- }
-
- SolrClient createClientRandomly() throws Exception {
- int r = random().nextInt(100);
-
- // testing the smart cloud client (requests to leaders) is more important than testing the forwarding logic
- if (r < 80) {
- return createCloudClient(SOURCE_COLLECTION);
- }
-
- if (r < 90) {
- return createNewSolrServer(shardToJetty.get(SOURCE_COLLECTION).get(SHARD1).get(random().nextInt(2)).url);
- }
-
- return createNewSolrServer(shardToJetty.get(SOURCE_COLLECTION).get(SHARD2).get(random().nextInt(2)).url);
- }
-
- @Test
- @ShardsFixed(num = 4)
- public void testCdcrDocVersions() throws Exception {
- SolrClient client = createClientRandomly();
- try {
- handle.clear();
- handle.put("timestamp", SKIPVAL);
-
- doTestCdcrDocVersions(client);
-
- commit(SOURCE_COLLECTION); // work arround SOLR-5628
- } finally {
- client.close();
- }
- }
-
- private void doTestCdcrDocVersions(SolrClient solrClient) throws Exception {
- this.solrServer = solrClient;
-
- log.info("### STARTING doCdcrTestDocVersions - Add commands, client: " + solrClient);
-
- vadd("doc1", 10, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "10");
- vadd("doc2", 11, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "11");
- vadd("doc3", 10, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "10");
- vadd("doc4", 11, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "11");
- commit(SOURCE_COLLECTION);
-
- // versions are preserved and verifiable both by query and by real-time get
- doQuery(solrClient, "doc1,10,doc2,11,doc3,10,doc4,11", "q", "*:*");
- doRealTimeGet("doc1,doc2,doc3,doc4", "10,11,10,11");
-
- vadd("doc1", 5, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "5");
- vadd("doc2", 10, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "10");
- vadd("doc3", 9, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "9");
- vadd("doc4", 8, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "8");
-
- // lower versions are ignored
- doRealTimeGet("doc1,doc2,doc3,doc4", "10,11,10,11");
-
- vadd("doc1", 12, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "12");
- vadd("doc2", 12, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "12");
- vadd("doc3", 12, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "12");
- vadd("doc4", 12, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "12");
-
- // higher versions are accepted
- doRealTimeGet("doc1,doc2,doc3,doc4", "12,12,12,12");
-
- // non-cdcr update requests throw a version conflict exception for non-equal versions (optimistic locking feature)
- vaddFail("doc1", 13, 409);
- vaddFail("doc2", 13, 409);
- vaddFail("doc3", 13, 409);
-
- commit(SOURCE_COLLECTION);
-
- // versions are still as they were
- doQuery(solrClient, "doc1,12,doc2,12,doc3,12,doc4,12", "q", "*:*");
-
- // query all shard replicas individually
- doQueryShardReplica(SHARD1, "doc1,12,doc2,12,doc3,12,doc4,12", "q", "*:*");
- doQueryShardReplica(SHARD2, "doc1,12,doc2,12,doc3,12,doc4,12", "q", "*:*");
-
- // optimistic locking update
- vadd("doc4", 12);
- commit(SOURCE_COLLECTION);
-
- QueryResponse rsp = solrClient.query(params("qt", "/get", "ids", "doc4"));
- long version = (long) rsp.getResults().get(0).get(vfield);
-
- // update accepted and a new version number was generated
- assertTrue(version > 1_000_000_000_000l);
-
- log.info("### STARTING doCdcrTestDocVersions - Delete commands");
-
- // send a delete update with an older version number
- vdelete("doc1", 5, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "5");
- // must ignore the delete
- doRealTimeGet("doc1", "12");
-
- // send a delete update with a higher version number
- vdelete("doc1", 13, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "13");
- // must be deleted
- doRealTimeGet("doc1", "");
-
- // send a delete update with a higher version number
- vdelete("doc4", version + 1, CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, "" + (version + 1));
- // must be deleted
- doRealTimeGet("doc4", "");
-
- commit(SOURCE_COLLECTION);
-
- // query each shard replica individually
- doQueryShardReplica(SHARD1, "doc2,12,doc3,12", "q", "*:*");
- doQueryShardReplica(SHARD2, "doc2,12,doc3,12", "q", "*:*");
-
- // version conflict thanks to optimistic locking
- if (solrClient instanceof CloudSolrClient) // TODO: it seems that optimistic locking doesn't work with forwarding, test with shard2 client
- vdeleteFail("doc2", 50, 409);
-
- // cleanup after ourselves for the next run
- // deleteByQuery should work as usual with the CDCR_UPDATE param
- doDeleteByQuery("id:doc*", CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, Long.toString(1));
- commit(SOURCE_COLLECTION);
-
- // deleteByQuery with a version lower than anything else should have no effect
- doQuery(solrClient, "doc2,12,doc3,12", "q", "*:*");
-
- doDeleteByQuery("id:doc*", CdcrUpdateProcessor.CDCR_UPDATE, "", vfield, Long.toString(51));
- commit(SOURCE_COLLECTION);
-
- // deleteByQuery with a version higher than everything else should delete all remaining docs
- doQuery(solrClient, "", "q", "*:*");
-
- // check that replicas are as expected too
- doQueryShardReplica(SHARD1, "", "q", "*:*");
- doQueryShardReplica(SHARD2, "", "q", "*:*");
- }
-
-
- // ------------------ auxiliary methods ------------------
-
-
- void doQueryShardReplica(String shard, String expectedDocs, String... queryParams) throws Exception {
- for (CloudJettyRunner jetty : shardToJetty.get(SOURCE_COLLECTION).get(shard)) {
- doQuery(jetty.client, expectedDocs, queryParams);
- }
- }
-
- void vdelete(String id, long version, String... params) throws Exception {
- UpdateRequest req = new UpdateRequest();
- req.deleteById(id);
- req.setParam(vfield, Long.toString(version));
-
- for (int i = 0; i < params.length; i += 2) {
- req.setParam(params[i], params[i + 1]);
- }
- solrServer.request(req);
- }
-
- void vdeleteFail(String id, long version, int errCode, String... params) throws Exception {
- boolean failed = false;
- try {
- vdelete(id, version, params);
- } catch (SolrException e) {
- failed = true;
- if (e.getCause() instanceof SolrException && e.getCause() != e) {
- e = (SolrException) e.getCause();
- }
- assertEquals(errCode, e.code());
- } catch (SolrServerException ex) {
- Throwable t = ex.getCause();
- if (t instanceof SolrException) {
- failed = true;
- SolrException exception = (SolrException) t;
- assertEquals(errCode, exception.code());
- }
- } catch (Exception e) {
- log.error("ERROR", e);
- }
- assertTrue(failed);
- }
-
- void vadd(String id, long version, String... params) throws Exception {
- UpdateRequest req = new UpdateRequest();
- req.add(sdoc("id", id, vfield, version));
- for (int i = 0; i < params.length; i += 2) {
- req.setParam(params[i], params[i + 1]);
- }
- solrServer.request(req);
- }
-
- void vaddFail(String id, long version, int errCode, String... params) throws Exception {
- boolean failed = false;
- try {
- vadd(id, version, params);
- } catch (SolrException e) {
- failed = true;
- if (e.getCause() instanceof SolrException && e.getCause() != e) {
- e = (SolrException) e.getCause();
- }
- assertEquals(errCode, e.code());
- } catch (SolrServerException ex) {
- Throwable t = ex.getCause();
- if (t instanceof SolrException) {
- failed = true;
- SolrException exception = (SolrException) t;
- assertEquals(errCode, exception.code());
- }
- } catch (Exception e) {
- log.error("ERROR", e);
- }
- assertTrue(failed);
- }
-
- void doQuery(SolrClient ss, String expectedDocs, String... queryParams) throws Exception {
-
- List<String> strs = StrUtils.splitSmart(expectedDocs, ",", true);
- Map<String, Object> expectedIds = new HashMap<>();
- for (int i = 0; i < strs.size(); i += 2) {
- String id = strs.get(i);
- String vS = strs.get(i + 1);
- Long v = Long.valueOf(vS);
- expectedIds.put(id, v);
- }
-
- QueryResponse rsp = ss.query(params(queryParams));
- Map<String, Object> obtainedIds = new HashMap<>();
- for (SolrDocument doc : rsp.getResults()) {
- obtainedIds.put((String) doc.get("id"), doc.get(vfield));
- }
-
- assertEquals(expectedIds, obtainedIds);
- }
-
-
- void doRealTimeGet(String ids, String versions) throws Exception {
- Map<String, Object> expectedIds = new HashMap<>();
- List<String> strs = StrUtils.splitSmart(ids, ",", true);
- List<String> verS = StrUtils.splitSmart(versions, ",", true);
- for (int i = 0; i < strs.size(); i++) {
- if (!verS.isEmpty()) {
- expectedIds.put(strs.get(i), Long.valueOf(verS.get(i)));
- }
- }
-
- QueryResponse rsp = solrServer.query(params("qt", "/get", "ids", ids));
- Map<String, Object> obtainedIds = new HashMap<>();
- for (SolrDocument doc : rsp.getResults()) {
- obtainedIds.put((String) doc.get("id"), doc.get(vfield));
- }
-
- assertEquals(expectedIds, obtainedIds);
- }
-
- void doDeleteByQuery(String q, String... reqParams) throws Exception {
- UpdateRequest req = new UpdateRequest();
- req.deleteByQuery(q);
- req.setParams(params(reqParams));
- req.process(solrServer);
- }
-
-}
-
[4/4] lucene-solr:branch_7x: SOLR-11003: Support bi-directional
syncing of cdcr clusters. We still only support actively into one cluster
cluster,
but have the ability to switch indexing clusters and cdcr will replicate
correctly
Posted by va...@apache.org.
SOLR-11003: Support bi-directional syncing of cdcr clusters. We still only support actively into one cluster cluster,
but have the ability to switch indexing clusters and cdcr will replicate correctly
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/a30c92a7
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/a30c92a7
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/a30c92a7
Branch: refs/heads/branch_7x
Commit: a30c92a79cd75d3c4fe21829e70292f8314afd4f
Parents: 1af62a6
Author: Varun Thacker <va...@apache.org>
Authored: Thu Nov 2 16:17:14 2017 -0700
Committer: Varun Thacker <va...@apache.org>
Committed: Mon Nov 6 08:13:08 2017 -0800
----------------------------------------------------------------------
solr/CHANGES.txt | 3 +
.../org/apache/solr/handler/CdcrReplicator.java | 28 +
.../apache/solr/update/CdcrTransactionLog.java | 144 +++
.../org/apache/solr/update/TransactionLog.java | 2 +-
.../cdcr-cluster1/conf/managed-schema | 29 +
.../cdcr-cluster1/conf/solrconfig.xml | 80 ++
.../cdcr-cluster2/conf/managed-schema | 29 +
.../cdcr-cluster2/conf/solrconfig.xml | 80 ++
.../solr/cloud/BaseCdcrDistributedZkTest.java | 903 ------------------
.../apache/solr/cloud/CdcrBootstrapTest.java | 356 --------
.../cloud/CdcrReplicationDistributedZkTest.java | 672 --------------
.../solr/cloud/CdcrReplicationHandlerTest.java | 330 -------
.../solr/cloud/CdcrRequestHandlerTest.java | 153 ----
.../solr/cloud/CdcrVersionReplicationTest.java | 307 -------
.../cloud/cdcr/BaseCdcrDistributedZkTest.java | 907 +++++++++++++++++++
.../solr/cloud/cdcr/CdcrBidirectionalTest.java | 234 +++++
.../solr/cloud/cdcr/CdcrBootstrapTest.java | 302 ++++++
.../cdcr/CdcrReplicationDistributedZkTest.java | 672 ++++++++++++++
.../cloud/cdcr/CdcrReplicationHandlerTest.java | 331 +++++++
.../solr/cloud/cdcr/CdcrRequestHandlerTest.java | 153 ++++
.../apache/solr/cloud/cdcr/CdcrTestsUtil.java | 111 +++
.../cloud/cdcr/CdcrVersionReplicationTest.java | 307 +++++++
22 files changed, 3411 insertions(+), 2722 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/CHANGES.txt
----------------------------------------------------------------------
diff --git a/solr/CHANGES.txt b/solr/CHANGES.txt
index 4af2f4c..15d74de 100644
--- a/solr/CHANGES.txt
+++ b/solr/CHANGES.txt
@@ -48,6 +48,9 @@ New Features
* SOLR-11438: Solr should return rf when min_rf is specified for deletes as well as adds (Erick Erickson)
+* SOLR-11003: Support bi-directional syncing of cdcr clusters. We still only support actively into one cluster cluster,
+ but have the ability to switch indexing clusters and cdcr will replicate correctly. (Amrit Sarkar, Varun Thacker)
+
Bug Fixes
----------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/java/org/apache/solr/handler/CdcrReplicator.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/handler/CdcrReplicator.java b/solr/core/src/java/org/apache/solr/handler/CdcrReplicator.java
index a151a43..5dca0d8 100644
--- a/solr/core/src/java/org/apache/solr/handler/CdcrReplicator.java
+++ b/solr/core/src/java/org/apache/solr/handler/CdcrReplicator.java
@@ -77,6 +77,10 @@ public class CdcrReplicator implements Runnable {
Object o = subReader.next();
if (o == null) break; // we have reached the end of the update logs, we should close the batch
+ if (isTargetCluster(o)) {
+ continue;
+ }
+
if (isDelete(o)) {
/*
@@ -140,6 +144,30 @@ public class CdcrReplicator implements Runnable {
state.resetConsecutiveErrors();
}
+ /** check whether the update read from TLog is received from source
+ * or received via solr client
+ */
+ private boolean isTargetCluster(Object o) {
+ List entry = (List) o;
+ int operationAndFlags = (Integer) entry.get(0);
+ int oper = operationAndFlags & UpdateLog.OPERATION_MASK;
+ Boolean isTarget = false;
+ if (oper == UpdateLog.DELETE_BY_QUERY || oper == UpdateLog.DELETE) {
+ if (entry.size() == 4) { //back-combat - skip for previous versions
+ isTarget = (Boolean) entry.get(entry.size() - 1);
+ }
+ } else if (oper == UpdateLog.UPDATE_INPLACE) {
+ if (entry.size() == 6) { //back-combat - skip for previous versions
+ isTarget = (Boolean) entry.get(entry.size() - 2);
+ }
+ } else if (oper == UpdateLog.ADD) {
+ if (entry.size() == 4) { //back-combat - skip for previous versions
+ isTarget = (Boolean) entry.get(entry.size() - 2);
+ }
+ }
+ return isTarget;
+ }
+
private boolean isDelete(Object o) {
List entry = (List) o;
int operationAndFlags = (Integer) entry.get(0);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/java/org/apache/solr/update/CdcrTransactionLog.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/update/CdcrTransactionLog.java b/solr/core/src/java/org/apache/solr/update/CdcrTransactionLog.java
index ce99200..f959e14 100644
--- a/solr/core/src/java/org/apache/solr/update/CdcrTransactionLog.java
+++ b/solr/core/src/java/org/apache/solr/update/CdcrTransactionLog.java
@@ -24,10 +24,13 @@ import java.nio.channels.Channels;
import java.nio.file.Files;
import java.util.Collection;
+import org.apache.lucene.util.BytesRef;
import org.apache.solr.common.SolrException;
+import org.apache.solr.common.SolrInputDocument;
import org.apache.solr.common.util.FastOutputStream;
import org.apache.solr.common.util.JavaBinCodec;
import org.apache.solr.common.util.ObjectReleaseTracker;
+import org.apache.solr.update.processor.CdcrUpdateProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -105,6 +108,147 @@ public class CdcrTransactionLog extends TransactionLog {
}
@Override
+ public long write(AddUpdateCommand cmd, long prevPointer, int flags) {
+ assert (-1 <= prevPointer && (cmd.isInPlaceUpdate() || (-1 == prevPointer)));
+
+ LogCodec codec = new LogCodec(resolver);
+ SolrInputDocument sdoc = cmd.getSolrInputDocument();
+
+ try {
+ checkWriteHeader(codec, sdoc);
+
+ // adaptive buffer sizing
+ int bufSize = lastAddSize; // unsynchronized access of lastAddSize should be fine
+ bufSize = Math.min(1024*1024, bufSize+(bufSize>>3)+256);
+
+ MemOutputStream out = new MemOutputStream(new byte[bufSize]);
+ codec.init(out);
+ if (cmd.isInPlaceUpdate()) {
+ codec.writeTag(JavaBinCodec.ARR, 6);
+ codec.writeInt(UpdateLog.UPDATE_INPLACE | flags); // should just take one byte
+ codec.writeLong(cmd.getVersion());
+ codec.writeLong(prevPointer);
+ codec.writeLong(cmd.prevVersion);
+ if (cmd.getReq().getParamString().contains(CdcrUpdateProcessor.CDCR_UPDATE)) {
+ // if the update is received via cdcr source; add boolean entry
+ // CdcrReplicator.isTargetCluster() checks that particular boolean to accept or discard the update
+ // to forward to its own target cluster
+ codec.writePrimitive(true);
+ } else {
+ codec.writePrimitive(false);
+ }
+ codec.writeSolrInputDocument(cmd.getSolrInputDocument());
+
+ } else {
+ codec.writeTag(JavaBinCodec.ARR, 4);
+ codec.writeInt(UpdateLog.ADD | flags); // should just take one byte
+ codec.writeLong(cmd.getVersion());
+ if (cmd.getReq().getParamString().contains(CdcrUpdateProcessor.CDCR_UPDATE)) {
+ // if the update is received via cdcr source; add extra boolean entry
+ // CdcrReplicator.isTargetCluster() checks that particular boolean to accept or discard the update
+ // to forward to its own target cluster
+ codec.writePrimitive(true);
+ } else {
+ codec.writePrimitive(false);
+ }
+ codec.writeSolrInputDocument(cmd.getSolrInputDocument());
+ }
+ lastAddSize = (int)out.size();
+
+ synchronized (this) {
+ long pos = fos.size(); // if we had flushed, this should be equal to channel.position()
+ assert pos != 0;
+
+ /***
+ System.out.println("###writing at " + pos + " fos.size()=" + fos.size() + " raf.length()=" + raf.length());
+ if (pos != fos.size()) {
+ throw new RuntimeException("ERROR" + "###writing at " + pos + " fos.size()=" + fos.size() + " raf.length()=" + raf.length());
+ }
+ ***/
+
+ out.writeAll(fos);
+ endRecord(pos);
+ // fos.flushBuffer(); // flush later
+ return pos;
+ }
+
+ } catch (IOException e) {
+ // TODO: reset our file pointer back to "pos", the start of this record.
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Error logging add", e);
+ }
+ }
+
+ @Override
+ public long writeDelete(DeleteUpdateCommand cmd, int flags) {
+ LogCodec codec = new LogCodec(resolver);
+
+ try {
+ checkWriteHeader(codec, null);
+
+ BytesRef br = cmd.getIndexedId();
+
+ MemOutputStream out = new MemOutputStream(new byte[20 + br.length]);
+ codec.init(out);
+ codec.writeTag(JavaBinCodec.ARR, 4);
+ codec.writeInt(UpdateLog.DELETE | flags); // should just take one byte
+ codec.writeLong(cmd.getVersion());
+ codec.writeByteArray(br.bytes, br.offset, br.length);
+ if (cmd.getReq().getParamString().contains(CdcrUpdateProcessor.CDCR_UPDATE)) {
+ // if the update is received via cdcr source; add extra boolean entry
+ // CdcrReplicator.isTargetCluster() checks that particular boolean to accept or discard the update
+ // to forward to its own target cluster
+ codec.writePrimitive(true);
+ } else {
+ codec.writePrimitive(false);
+ }
+
+ synchronized (this) {
+ long pos = fos.size(); // if we had flushed, this should be equal to channel.position()
+ assert pos != 0;
+ out.writeAll(fos);
+ endRecord(pos);
+ // fos.flushBuffer(); // flush later
+ return pos;
+ }
+
+ } catch (IOException e) {
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, e);
+ }
+ }
+
+ @Override
+ public long writeDeleteByQuery(DeleteUpdateCommand cmd, int flags) {
+ LogCodec codec = new LogCodec(resolver);
+ try {
+ checkWriteHeader(codec, null);
+
+ MemOutputStream out = new MemOutputStream(new byte[20 + (cmd.query.length())]);
+ codec.init(out);
+ codec.writeTag(JavaBinCodec.ARR, 4);
+ codec.writeInt(UpdateLog.DELETE_BY_QUERY | flags); // should just take one byte
+ codec.writeLong(cmd.getVersion());
+ codec.writeStr(cmd.query);
+ if (cmd.getReq().getParamString().contains(CdcrUpdateProcessor.CDCR_UPDATE)) {
+ // if the update is received via cdcr source; add extra boolean entry
+ // CdcrReplicator.isTargetCluster() checks that particular boolean to accept or discard the update
+ // to forward to its own target cluster
+ codec.writePrimitive(true);
+ } else {
+ codec.writePrimitive(false);
+ }
+ synchronized (this) {
+ long pos = fos.size(); // if we had flushed, this should be equal to channel.position()
+ out.writeAll(fos);
+ endRecord(pos);
+ // fos.flushBuffer(); // flush later
+ return pos;
+ }
+ } catch (IOException e) {
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, e);
+ }
+ }
+
+ @Override
public long writeCommit(CommitUpdateCommand cmd, int flags) {
LogCodec codec = new LogCodec(resolver);
synchronized (this) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/java/org/apache/solr/update/TransactionLog.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/update/TransactionLog.java b/solr/core/src/java/org/apache/solr/update/TransactionLog.java
index 73328cf..7daf44f 100644
--- a/solr/core/src/java/org/apache/solr/update/TransactionLog.java
+++ b/solr/core/src/java/org/apache/solr/update/TransactionLog.java
@@ -326,7 +326,7 @@ public class TransactionLog implements Closeable {
numRecords++;
}
- private void checkWriteHeader(LogCodec codec, SolrInputDocument optional) throws IOException {
+ protected void checkWriteHeader(LogCodec codec, SolrInputDocument optional) throws IOException {
// Unsynchronized access. We can get away with an unsynchronized access here
// since we will never get a false non-zero when the position is in fact 0.
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test-files/solr/configsets/cdcr-cluster1/conf/managed-schema
----------------------------------------------------------------------
diff --git a/solr/core/src/test-files/solr/configsets/cdcr-cluster1/conf/managed-schema b/solr/core/src/test-files/solr/configsets/cdcr-cluster1/conf/managed-schema
new file mode 100644
index 0000000..2df6c0a
--- /dev/null
+++ b/solr/core/src/test-files/solr/configsets/cdcr-cluster1/conf/managed-schema
@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!--
+ 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.
+-->
+<schema name="minimal" version="1.1">
+ <types>
+ <fieldType name="string" class="solr.StrField"/>
+ <fieldType name="long" class="${solr.tests.LongFieldType}" docValues="${solr.tests.numeric.dv}" precisionStep="0" positionIncrementGap="0"/>
+ </types>
+ <fields>
+ <field name="id" type="string" indexed="true" stored="true"/>
+ <field name="_version_" type="long" indexed="true" stored="true"/>
+ <dynamicField name="*" type="string" indexed="true" stored="true"/>
+ </fields>
+ <uniqueKey>id</uniqueKey>
+</schema>
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test-files/solr/configsets/cdcr-cluster1/conf/solrconfig.xml
----------------------------------------------------------------------
diff --git a/solr/core/src/test-files/solr/configsets/cdcr-cluster1/conf/solrconfig.xml b/solr/core/src/test-files/solr/configsets/cdcr-cluster1/conf/solrconfig.xml
new file mode 100644
index 0000000..da548c4
--- /dev/null
+++ b/solr/core/src/test-files/solr/configsets/cdcr-cluster1/conf/solrconfig.xml
@@ -0,0 +1,80 @@
+<?xml version="1.0" ?>
+
+<!--
+ 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.
+-->
+
+<!-- This is a "kitchen sink" config file that tests can use.
+ When writting a new test, feel free to add *new* items (plugins,
+ config options, etc...) as long as they don't break any existing
+ tests. if you need to test something esoteric please add a new
+ "solrconfig-your-esoteric-purpose.xml" config file.
+
+ Note in particular that this test is used by MinimalSchemaTest so
+ Anything added to this file needs to work correctly even if there
+ is now uniqueKey or defaultSearch Field.
+ -->
+
+<config>
+
+ <dataDir>${solr.data.dir:}</dataDir>
+
+ <directoryFactory name="DirectoryFactory"
+ class="${solr.directoryFactory:solr.NRTCachingDirectoryFactory}"/>
+
+ <luceneMatchVersion>${tests.luceneMatchVersion:LATEST}</luceneMatchVersion>
+
+ <updateRequestProcessorChain name="cdcr-processor-chain">
+ <processor class="solr.CdcrUpdateProcessorFactory"/>
+ <processor class="solr.RunUpdateProcessorFactory"/>
+ </updateRequestProcessorChain>
+
+ <requestHandler name="/cdcr" class="solr.CdcrRequestHandler">
+ <lst name="replica">
+ <str name="zkHost">${cdcr.cluster2.zkHost}</str>
+ <str name="source">cdcr-cluster1</str>
+ <str name="target">cdcr-cluster2</str>
+ </lst>
+ <lst name="replicator">
+ <str name="threadPoolSize">1</str>
+ <str name="schedule">1000</str>
+ <str name="batchSize">1000</str>
+ </lst>
+ <lst name="updateLogSynchronizer">
+ <str name="schedule">1000</str>
+ </lst>
+ </requestHandler>
+
+ <updateHandler class="solr.DirectUpdateHandler2">
+ <updateLog class="solr.CdcrUpdateLog">
+ <str name="dir">${solr.ulog.dir:}</str>
+ </updateLog>
+ </updateHandler>
+
+ <requestHandler name="/select" class="solr.SearchHandler" />
+
+ <initParams path="/update/**,/query,/select,/tvrh,/elevate,/spell,/browse">
+ <lst name="defaults">
+ <str name="df">_text_</str>
+ </lst>
+ </initParams>
+
+ <requestHandler name="/update" class="solr.UpdateRequestHandler">
+ <lst name="defaults">
+ <str name="update.chain">cdcr-processor-chain</str>
+ </lst>
+ </requestHandler>
+</config>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test-files/solr/configsets/cdcr-cluster2/conf/managed-schema
----------------------------------------------------------------------
diff --git a/solr/core/src/test-files/solr/configsets/cdcr-cluster2/conf/managed-schema b/solr/core/src/test-files/solr/configsets/cdcr-cluster2/conf/managed-schema
new file mode 100644
index 0000000..2df6c0a
--- /dev/null
+++ b/solr/core/src/test-files/solr/configsets/cdcr-cluster2/conf/managed-schema
@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!--
+ 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.
+-->
+<schema name="minimal" version="1.1">
+ <types>
+ <fieldType name="string" class="solr.StrField"/>
+ <fieldType name="long" class="${solr.tests.LongFieldType}" docValues="${solr.tests.numeric.dv}" precisionStep="0" positionIncrementGap="0"/>
+ </types>
+ <fields>
+ <field name="id" type="string" indexed="true" stored="true"/>
+ <field name="_version_" type="long" indexed="true" stored="true"/>
+ <dynamicField name="*" type="string" indexed="true" stored="true"/>
+ </fields>
+ <uniqueKey>id</uniqueKey>
+</schema>
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test-files/solr/configsets/cdcr-cluster2/conf/solrconfig.xml
----------------------------------------------------------------------
diff --git a/solr/core/src/test-files/solr/configsets/cdcr-cluster2/conf/solrconfig.xml b/solr/core/src/test-files/solr/configsets/cdcr-cluster2/conf/solrconfig.xml
new file mode 100644
index 0000000..8e26d45
--- /dev/null
+++ b/solr/core/src/test-files/solr/configsets/cdcr-cluster2/conf/solrconfig.xml
@@ -0,0 +1,80 @@
+<?xml version="1.0" ?>
+
+<!--
+ 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.
+-->
+
+<!-- This is a "kitchen sink" config file that tests can use.
+ When writting a new test, feel free to add *new* items (plugins,
+ config options, etc...) as long as they don't break any existing
+ tests. if you need to test something esoteric please add a new
+ "solrconfig-your-esoteric-purpose.xml" config file.
+
+ Note in particular that this test is used by MinimalSchemaTest so
+ Anything added to this file needs to work correctly even if there
+ is now uniqueKey or defaultSearch Field.
+ -->
+
+<config>
+
+ <dataDir>${solr.data.dir:}</dataDir>
+
+ <directoryFactory name="DirectoryFactory"
+ class="${solr.directoryFactory:solr.NRTCachingDirectoryFactory}"/>
+
+ <luceneMatchVersion>${tests.luceneMatchVersion:LATEST}</luceneMatchVersion>
+
+ <updateRequestProcessorChain name="cdcr-processor-chain">
+ <processor class="solr.CdcrUpdateProcessorFactory"/>
+ <processor class="solr.RunUpdateProcessorFactory"/>
+ </updateRequestProcessorChain>
+
+ <requestHandler name="/cdcr" class="solr.CdcrRequestHandler">
+ <lst name="replica">
+ <str name="zkHost">${cdcr.cluster1.zkHost}</str>
+ <str name="source">cdcr-cluster2</str>
+ <str name="target">cdcr-cluster1</str>
+ </lst>
+ <lst name="replicator">
+ <str name="threadPoolSize">1</str>
+ <str name="schedule">1000</str>
+ <str name="batchSize">1000</str>
+ </lst>
+ <lst name="updateLogSynchronizer">
+ <str name="schedule">1000</str>
+ </lst>
+ </requestHandler>
+
+ <updateHandler class="solr.DirectUpdateHandler2">
+ <updateLog class="solr.CdcrUpdateLog">
+ <str name="dir">${solr.ulog.dir:}</str>
+ </updateLog>
+ </updateHandler>
+
+ <requestHandler name="/select" class="solr.SearchHandler" />
+
+ <initParams path="/update/**,/query,/select,/tvrh,/elevate,/spell,/browse">
+ <lst name="defaults">
+ <str name="df">_text_</str>
+ </lst>
+ </initParams>
+
+ <requestHandler name="/update" class="solr.UpdateRequestHandler">
+ <lst name="defaults">
+ <str name="update.chain">cdcr-processor-chain</str>
+ </lst>
+ </requestHandler>
+</config>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test/org/apache/solr/cloud/BaseCdcrDistributedZkTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/BaseCdcrDistributedZkTest.java b/solr/core/src/test/org/apache/solr/cloud/BaseCdcrDistributedZkTest.java
deleted file mode 100644
index fc563aa..0000000
--- a/solr/core/src/test/org/apache/solr/cloud/BaseCdcrDistributedZkTest.java
+++ /dev/null
@@ -1,903 +0,0 @@
-/*
- * 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.File;
-import java.io.IOException;
-import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.solr.client.solrj.SolrClient;
-import org.apache.solr.client.solrj.SolrQuery;
-import org.apache.solr.client.solrj.SolrRequest;
-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.QueryRequest;
-import org.apache.solr.client.solrj.response.CollectionAdminResponse;
-import org.apache.solr.common.SolrInputDocument;
-import org.apache.solr.common.cloud.ClusterState;
-import org.apache.solr.common.cloud.DocCollection;
-import org.apache.solr.common.cloud.Replica;
-import org.apache.solr.common.cloud.Slice;
-import org.apache.solr.common.cloud.SolrZkClient;
-import org.apache.solr.common.cloud.ZkCoreNodeProps;
-import org.apache.solr.common.cloud.ZkNodeProps;
-import org.apache.solr.common.cloud.ZkStateReader;
-import org.apache.solr.common.params.CollectionParams;
-import org.apache.solr.common.params.CommonParams;
-import org.apache.solr.common.params.ModifiableSolrParams;
-import org.apache.solr.common.util.IOUtils;
-import org.apache.solr.common.util.NamedList;
-import org.apache.solr.common.util.StrUtils;
-import org.apache.solr.common.util.Utils;
-import org.apache.solr.core.CoreDescriptor;
-import org.apache.solr.core.SolrCore;
-import org.apache.solr.handler.CdcrParams;
-import org.apache.solr.util.TimeOut;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException;
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-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.SHARDS_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.CLUSTER_PROPS;
-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.handler.admin.CoreAdminHandler.COMPLETED;
-import static org.apache.solr.handler.admin.CoreAdminHandler.RESPONSE_STATUS;
-
-/**
- * <p>
- * Abstract class for CDCR unit testing. This class emulates two clusters, a source and target, by using different
- * collections in the same SolrCloud cluster. Therefore, the two clusters will share the same Zookeeper cluster. In
- * real scenario, the two collections/clusters will likely have their own zookeeper cluster.
- * </p>
- * <p>
- * This class will automatically create two collections, the source and the target. Each collection will have
- * {@link #shardCount} shards, and {@link #replicationFactor} replicas per shard. One jetty instance will
- * be created per core.
- * </p>
- * <p>
- * The source and target collection can be reinitialised at will by calling {@link #clearSourceCollection()} and
- * {@link #clearTargetCollection()}. After reinitialisation, a collection will have a new fresh index and update log.
- * </p>
- * <p>
- * Servers can be restarted at will by calling
- * {@link #restartServer(BaseCdcrDistributedZkTest.CloudJettyRunner)} or
- * {@link #restartServers(java.util.List)}.
- * </p>
- * <p>
- * The creation of the target collection can be disabled with the flag {@link #createTargetCollection};
- * </p>
- * <p>
- * NB: We cannot use multiple cores per jetty instance, as jetty will load only one core when restarting. It seems
- * that this is a limitation of the {@link org.apache.solr.client.solrj.embedded.JettySolrRunner}. This class
- * tries to ensure that there always is one single core per jetty instance.
- * </p>
- */
-public class BaseCdcrDistributedZkTest extends AbstractDistribZkTestBase {
-
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- protected int shardCount = 2;
- protected int replicationFactor = 2;
- protected boolean createTargetCollection = true;
-
- private static final String CDCR_PATH = "/cdcr";
-
- protected static final String SOURCE_COLLECTION = "source_collection";
- protected static final String TARGET_COLLECTION = "target_collection";
-
- public static final String SHARD1 = "shard1";
- public static final String SHARD2 = "shard2";
-
- @Override
- protected String getCloudSolrConfig() {
- return "solrconfig-cdcr.xml";
- }
-
- @Override
- public void distribSetUp() throws Exception {
- super.distribSetUp();
-
- if (shardCount > 0) {
- System.setProperty("numShards", Integer.toString(shardCount));
- } else {
- System.clearProperty("numShards");
- }
-
- if (isSSLMode()) {
- System.clearProperty("urlScheme");
- ZkStateReader zkStateReader = new ZkStateReader(zkServer.getZkAddress(),
- AbstractZkTestCase.TIMEOUT, AbstractZkTestCase.TIMEOUT);
- try {
- zkStateReader.getZkClient().create(ZkStateReader.CLUSTER_PROPS,
- Utils.toJSON(Collections.singletonMap("urlScheme", "https")),
- CreateMode.PERSISTENT, true);
- } catch (KeeperException.NodeExistsException e) {
- ZkNodeProps props = ZkNodeProps.load(zkStateReader.getZkClient().getData(ZkStateReader.CLUSTER_PROPS,
- null, null, true));
- props = props.plus("urlScheme", "https");
- zkStateReader.getZkClient().setData(CLUSTER_PROPS, Utils.toJSON(props), true);
- } finally {
- zkStateReader.close();
- }
- }
- }
-
- @Override
- protected void createServers(int numServers) throws Exception {
- }
-
- @BeforeClass
- public static void beforeClass() {
- System.setProperty("solrcloud.update.delay", "0");
- }
-
- @AfterClass
- public static void afterClass() throws Exception {
- System.clearProperty("solrcloud.update.delay");
- }
-
- @Before
- public void baseBefore() throws Exception {
- this.createSourceCollection();
- if (this.createTargetCollection) this.createTargetCollection();
- RandVal.uniqueValues = new HashSet(); //reset random values
- }
-
- @After
- public void baseAfter() throws Exception {
- for (List<CloudJettyRunner> runners : cloudJettys.values()) {
- for (CloudJettyRunner runner : runners) {
- runner.client.close();
- }
- }
- destroyServers();
- }
-
- protected CloudSolrClient createCloudClient(String defaultCollection) {
- CloudSolrClient server = getCloudSolrClient(zkServer.getZkAddress(), random().nextBoolean());
- server.setParallelUpdates(random().nextBoolean());
- if (defaultCollection != null) server.setDefaultCollection(defaultCollection);
- return server;
- }
-
- protected void printLayout() throws Exception {
- SolrZkClient zkClient = new SolrZkClient(zkServer.getZkHost(), AbstractZkTestCase.TIMEOUT);
- zkClient.printLayoutToStdOut();
- zkClient.close();
- }
-
- protected SolrInputDocument getDoc(Object... fields) throws Exception {
- SolrInputDocument doc = new SolrInputDocument();
- addFields(doc, fields);
- return doc;
- }
-
- protected void index(String collection, SolrInputDocument doc) throws IOException, SolrServerException {
- CloudSolrClient client = createCloudClient(collection);
- try {
- client.add(doc);
- client.commit(true, true);
- } finally {
- client.close();
- }
- }
-
- protected void index(String collection, List<SolrInputDocument> docs) throws IOException, SolrServerException {
- CloudSolrClient client = createCloudClient(collection);
- try {
- client.add(docs);
- client.commit(true, true);
- } finally {
- client.close();
- }
- }
-
- protected void deleteById(String collection, List<String> ids) throws IOException, SolrServerException {
- CloudSolrClient client = createCloudClient(collection);
- try {
- client.deleteById(ids);
- client.commit(true, true);
- } finally {
- client.close();
- }
- }
-
- protected void deleteByQuery(String collection, String q) throws IOException, SolrServerException {
- CloudSolrClient client = createCloudClient(collection);
- try {
- client.deleteByQuery(q);
- client.commit(true, true);
- } finally {
- client.close();
- }
- }
-
- /**
- * Invokes a commit on the given collection.
- */
- protected void commit(String collection) throws IOException, SolrServerException {
- CloudSolrClient client = createCloudClient(collection);
- try {
- client.commit(true, true);
- } finally {
- client.close();
- }
- }
-
- /**
- * Assert the number of documents in a given collection
- */
- protected void assertNumDocs(int expectedNumDocs, String collection)
- throws SolrServerException, IOException, InterruptedException {
- CloudSolrClient client = createCloudClient(collection);
- try {
- int cnt = 30; // timeout after 15 seconds
- AssertionError lastAssertionError = null;
- while (cnt > 0) {
- try {
- assertEquals(expectedNumDocs, client.query(new SolrQuery("*:*")).getResults().getNumFound());
- return;
- }
- catch (AssertionError e) {
- lastAssertionError = e;
- cnt--;
- Thread.sleep(500);
- }
- }
- throw new AssertionError("Timeout while trying to assert number of documents @ " + collection, lastAssertionError);
- } finally {
- client.close();
- }
- }
-
- /**
- * Invokes a CDCR action on a given node.
- */
- protected NamedList invokeCdcrAction(CloudJettyRunner jetty, CdcrParams.CdcrAction action) throws Exception {
- ModifiableSolrParams params = new ModifiableSolrParams();
- params.set(CommonParams.ACTION, action.toString());
-
- SolrRequest request = new QueryRequest(params);
- request.setPath(CDCR_PATH);
-
- return jetty.client.request(request);
- }
-
- protected void waitForCdcrStateReplication(String collection) throws Exception {
- log.info("Wait for CDCR state to replicate - collection: " + collection);
-
- int cnt = 30;
- while (cnt > 0) {
- NamedList status = null;
- boolean allEquals = true;
- for (CloudJettyRunner jetty : cloudJettys.get(collection)) { // check all replicas
- NamedList rsp = invokeCdcrAction(jetty, CdcrParams.CdcrAction.STATUS);
- if (status == null) {
- status = (NamedList) rsp.get(CdcrParams.CdcrAction.STATUS.toLower());
- continue;
- }
- allEquals &= status.equals(rsp.get(CdcrParams.CdcrAction.STATUS.toLower()));
- }
-
- if (allEquals) {
- break;
- }
- else {
- if (cnt == 0) {
- throw new RuntimeException("Timeout waiting for CDCR state to replicate: collection="+collection);
- }
- cnt--;
- Thread.sleep(500);
- }
- }
-
- log.info("CDCR state is identical across nodes - collection: " + collection);
- }
-
- /**
- * Assert the state of CDCR on each nodes of the given collection.
- */
- protected void assertState(String collection, CdcrParams.ProcessState processState, CdcrParams.BufferState bufferState)
- throws Exception {
- this.waitForCdcrStateReplication(collection); // ensure that cdcr state is replicated and stable
- for (CloudJettyRunner jetty : cloudJettys.get(collection)) { // check all replicas
- NamedList rsp = invokeCdcrAction(jetty, CdcrParams.CdcrAction.STATUS);
- NamedList status = (NamedList) rsp.get(CdcrParams.CdcrAction.STATUS.toLower());
- assertEquals(processState.toLower(), status.get(CdcrParams.ProcessState.getParam()));
- assertEquals(bufferState.toLower(), status.get(CdcrParams.BufferState.getParam()));
- }
- }
-
- /**
- * A mapping between collection and node names. This is used when creating the collection in
- * {@link #createCollection(String)}.
- */
- private Map<String, List<String>> collectionToNodeNames = new HashMap<>();
-
- /**
- * Starts the servers, saves and associates the node names to the source collection,
- * and finally creates the source collection.
- */
- private void createSourceCollection() throws Exception {
- List<String> nodeNames = this.startServers(shardCount * replicationFactor);
- this.collectionToNodeNames.put(SOURCE_COLLECTION, nodeNames);
- this.createCollection(SOURCE_COLLECTION);
- this.waitForRecoveriesToFinish(SOURCE_COLLECTION, true);
- this.updateMappingsFromZk(SOURCE_COLLECTION);
- }
-
- /**
- * Clear the source collection. It will delete then create the collection through the collection API.
- * The collection will have a new fresh index, i.e., including a new update log.
- */
- protected void clearSourceCollection() throws Exception {
- this.deleteCollection(SOURCE_COLLECTION);
- this.waitForCollectionToDisappear(SOURCE_COLLECTION);
- this.createCollection(SOURCE_COLLECTION);
- this.waitForRecoveriesToFinish(SOURCE_COLLECTION, true);
- this.updateMappingsFromZk(SOURCE_COLLECTION);
- }
-
- /**
- * Starts the servers, saves and associates the node names to the target collection,
- * and finally creates the target collection.
- */
- private void createTargetCollection() throws Exception {
- List<String> nodeNames = this.startServers(shardCount * replicationFactor);
- this.collectionToNodeNames.put(TARGET_COLLECTION, nodeNames);
- this.createCollection(TARGET_COLLECTION);
- this.waitForRecoveriesToFinish(TARGET_COLLECTION, true);
- this.updateMappingsFromZk(TARGET_COLLECTION);
- }
-
- /**
- * Clear the source collection. It will delete then create the collection through the collection API.
- * The collection will have a new fresh index, i.e., including a new update log.
- */
- protected void clearTargetCollection() throws Exception {
- this.deleteCollection(TARGET_COLLECTION);
- this.waitForCollectionToDisappear(TARGET_COLLECTION);
- this.createCollection(TARGET_COLLECTION);
- this.waitForRecoveriesToFinish(TARGET_COLLECTION, true);
- this.updateMappingsFromZk(TARGET_COLLECTION);
- }
-
- /**
- * Create a new collection through the Collection API. It enforces the use of one max shard per node.
- * It will define the nodes to spread the new collection across by using the mapping {@link #collectionToNodeNames},
- * to ensure that a node will not host more than one core (which will create problem when trying to restart servers).
- */
- private void createCollection(String name) throws Exception {
- CloudSolrClient client = createCloudClient(null);
- try {
- // Create the target collection
- Map<String, List<Integer>> collectionInfos = new HashMap<>();
- int maxShardsPerNode = 1;
-
- StringBuilder sb = new StringBuilder();
- for (String nodeName : collectionToNodeNames.get(name)) {
- sb.append(nodeName);
- sb.append(',');
- }
- sb.deleteCharAt(sb.length() - 1);
-
- createCollection(collectionInfos, name, shardCount, replicationFactor, maxShardsPerNode, client, sb.toString());
- } finally {
- client.close();
- }
- }
-
- private CollectionAdminResponse createCollection(Map<String, List<Integer>> collectionInfos,
- String collectionName, int numShards, int replicationFactor,
- int maxShardsPerNode, SolrClient client, String createNodeSetStr)
- throws SolrServerException, IOException {
- return createCollection(collectionInfos, collectionName,
- Utils.makeMap(
- NUM_SLICES, numShards,
- REPLICATION_FACTOR, replicationFactor,
- CREATE_NODE_SET, createNodeSetStr,
- MAX_SHARDS_PER_NODE, maxShardsPerNode),
- client, "conf1");
- }
-
- private CollectionAdminResponse createCollection(Map<String, List<Integer>> collectionInfos, String collectionName,
- Map<String, Object> collectionProps, SolrClient client,
- String confSetName)
- throws SolrServerException, IOException {
- ModifiableSolrParams params = new ModifiableSolrParams();
- params.set("action", CollectionParams.CollectionAction.CREATE.toString());
- for (Map.Entry<String, Object> entry : collectionProps.entrySet()) {
- if (entry.getValue() != null) params.set(entry.getKey(), String.valueOf(entry.getValue()));
- }
- Integer numShards = (Integer) collectionProps.get(NUM_SLICES);
- if (numShards == null) {
- String shardNames = (String) collectionProps.get(SHARDS_PROP);
- numShards = StrUtils.splitSmart(shardNames, ',').size();
- }
- Integer replicationFactor = (Integer) collectionProps.get(REPLICATION_FACTOR);
- if (replicationFactor == null) {
- replicationFactor = (Integer) OverseerCollectionMessageHandler.COLL_PROPS.get(REPLICATION_FACTOR);
- }
-
- if (confSetName != null) {
- params.set("collection.configName", confSetName);
- }
-
- List<Integer> list = new ArrayList<>();
- list.add(numShards);
- list.add(replicationFactor);
- if (collectionInfos != null) {
- collectionInfos.put(collectionName, list);
- }
- params.set("name", collectionName);
- SolrRequest request = new QueryRequest(params);
- request.setPath("/admin/collections");
-
- CollectionAdminResponse res = new CollectionAdminResponse();
- res.setResponse(client.request(request));
- return res;
- }
-
- /**
- * Delete a collection through the Collection API.
- */
- protected CollectionAdminResponse deleteCollection(String collectionName) throws Exception {
- SolrClient client = createCloudClient(null);
- CollectionAdminResponse res;
-
- try {
- ModifiableSolrParams params = new ModifiableSolrParams();
- params.set("action", CollectionParams.CollectionAction.DELETE.toString());
- params.set("name", collectionName);
- QueryRequest request = new QueryRequest(params);
- request.setPath("/admin/collections");
-
- res = new CollectionAdminResponse();
- res.setResponse(client.request(request));
- } catch (Exception e) {
- log.warn("Error while deleting the collection " + collectionName, e);
- return new CollectionAdminResponse();
- } finally {
- client.close();
- }
-
- return res;
- }
-
- private void waitForCollectionToDisappear(String collection) throws Exception {
- CloudSolrClient client = this.createCloudClient(null);
- try {
- client.connect();
- ZkStateReader zkStateReader = client.getZkStateReader();
- AbstractDistribZkTestBase.waitForCollectionToDisappear(collection, zkStateReader, false, true, 15);
- } finally {
- client.close();
- }
- }
-
- private void waitForRecoveriesToFinish(String collection, boolean verbose) throws Exception {
- CloudSolrClient client = this.createCloudClient(null);
- try {
- client.connect();
- ZkStateReader zkStateReader = client.getZkStateReader();
- super.waitForRecoveriesToFinish(collection, zkStateReader, verbose);
- } finally {
- client.close();
- }
- }
-
- /**
- * Asserts that the collection has the correct number of shards and replicas
- */
- protected void assertCollectionExpectations(String collectionName) throws Exception {
- CloudSolrClient client = this.createCloudClient(null);
- try {
- client.connect();
- ClusterState clusterState = client.getZkStateReader().getClusterState();
-
- assertTrue("Could not find new collection " + collectionName, clusterState.hasCollection(collectionName));
- Map<String, Slice> shards = clusterState.getCollection(collectionName).getSlicesMap();
- // did we find expectedSlices shards/shards?
- assertEquals("Found new collection " + collectionName + ", but mismatch on number of shards.", shardCount, shards.size());
- int totalShards = 0;
- for (String shardName : shards.keySet()) {
- totalShards += shards.get(shardName).getReplicas().size();
- }
- int expectedTotalShards = shardCount * replicationFactor;
- assertEquals("Found new collection " + collectionName + " with correct number of shards, but mismatch on number " +
- "of shards.", expectedTotalShards, totalShards);
- } finally {
- client.close();
- }
- }
-
- /**
- * Restart a server.
- */
- protected void restartServer(CloudJettyRunner server) throws Exception {
- // it seems we need to set the collection property to have the jetty properly restarted
- System.setProperty("collection", server.collection);
- JettySolrRunner jetty = server.jetty;
- ChaosMonkey.stop(jetty);
- ChaosMonkey.start(jetty);
- System.clearProperty("collection");
- waitForRecoveriesToFinish(server.collection, true);
- updateMappingsFromZk(server.collection); // must update the mapping as the core node name might have changed
- }
-
- /**
- * Restarts a list of servers.
- */
- protected void restartServers(List<CloudJettyRunner> servers) throws Exception {
- for (CloudJettyRunner server : servers) {
- this.restartServer(server);
- }
- }
-
- private List<JettySolrRunner> jettys = new ArrayList<>();
-
- /**
- * Creates and starts a given number of servers.
- */
- protected List<String> startServers(int nServer) throws Exception {
- String temporaryCollection = "tmp_collection";
-
- for (int i = 1; i <= nServer; i++) {
- // give everyone there own solrhome
- File jettyDir = createTempDir("jetty").toFile();
- jettyDir.mkdirs();
- setupJettySolrHome(jettyDir);
- JettySolrRunner jetty = createJetty(jettyDir, null, "shard" + i);
- jettys.add(jetty);
- }
-
- try (SolrClient client = createCloudClient(temporaryCollection)) {
- assertEquals(0, CollectionAdminRequest
- .createCollection(temporaryCollection, "conf1", shardCount, 1)
- .setCreateNodeSet("")
- .process(client).getStatus());
- for (int i = 0; i < jettys.size(); i++) {
- assertTrue(CollectionAdminRequest
- .addReplicaToShard(temporaryCollection, "shard"+((i % shardCount) + 1))
- .setNode(jettys.get(i).getNodeName())
- .process(client).isSuccess());
- }
- }
-
- ZkStateReader zkStateReader = jettys.get(0).getCoreContainer().getZkController().getZkStateReader();
-
- // now wait till we see the leader for each shard
- for (int i = 1; i <= shardCount; i++) {
- this.printLayout();
- zkStateReader.getLeaderRetry(temporaryCollection, "shard" + i, 15000);
- }
-
- // store the node names
- List<String> nodeNames = new ArrayList<>();
- for (Slice shard : zkStateReader.getClusterState().getCollection(temporaryCollection).getSlices()) {
- for (Replica replica : shard.getReplicas()) {
- nodeNames.add(replica.getNodeName());
- }
- }
-
- this.waitForRecoveriesToFinish(temporaryCollection,zkStateReader, true);
- // delete the temporary collection - we will create our own collections later
- this.deleteCollection(temporaryCollection);
- this.waitForCollectionToDisappear(temporaryCollection);
- System.clearProperty("collection");
-
- return nodeNames;
- }
-
- @Override
- protected void destroyServers() throws Exception {
- for (JettySolrRunner runner : jettys) {
- try {
- ChaosMonkey.stop(runner);
- } catch (Exception e) {
- log.error("", e);
- }
- }
-
- jettys.clear();
- }
-
- /**
- * Mapping from collection to jettys
- */
- protected Map<String, List<CloudJettyRunner>> cloudJettys = new HashMap<>();
-
- /**
- * Mapping from collection/shard to jettys
- */
- protected Map<String, Map<String, List<CloudJettyRunner>>> shardToJetty = new HashMap<>();
-
- /**
- * Mapping from collection/shard leader to jettys
- */
- protected Map<String, Map<String, CloudJettyRunner>> shardToLeaderJetty = new HashMap<>();
-
- /**
- * Updates the mappings between the jetty's instances and the zookeeper cluster state.
- */
- protected void updateMappingsFromZk(String collection) throws Exception {
- List<CloudJettyRunner> cloudJettys = new ArrayList<>();
- Map<String, List<CloudJettyRunner>> shardToJetty = new HashMap<>();
- Map<String, CloudJettyRunner> shardToLeaderJetty = new HashMap<>();
-
- CloudSolrClient cloudClient = this.createCloudClient(null);
- try {
- cloudClient.connect();
- ZkStateReader zkStateReader = cloudClient.getZkStateReader();
- ClusterState clusterState = zkStateReader.getClusterState();
- DocCollection coll = clusterState.getCollection(collection);
-
- for (JettySolrRunner jetty : jettys) {
- int port = jetty.getLocalPort();
- if (port == -1) {
- throw new RuntimeException("Cannot find the port for jetty");
- }
-
- nextJetty:
- for (Slice shard : coll.getSlices()) {
- Set<Map.Entry<String, Replica>> entries = shard.getReplicasMap().entrySet();
- for (Map.Entry<String, Replica> entry : entries) {
- Replica replica = entry.getValue();
- if (replica.getStr(ZkStateReader.BASE_URL_PROP).contains(":" + port)) {
- if (!shardToJetty.containsKey(shard.getName())) {
- shardToJetty.put(shard.getName(), new ArrayList<CloudJettyRunner>());
- }
- boolean isLeader = shard.getLeader() == replica;
- CloudJettyRunner cjr = new CloudJettyRunner(jetty, replica, collection, shard.getName(), entry.getKey());
- shardToJetty.get(shard.getName()).add(cjr);
- if (isLeader) {
- shardToLeaderJetty.put(shard.getName(), cjr);
- }
- cloudJettys.add(cjr);
- break nextJetty;
- }
- }
- }
- }
-
- List<CloudJettyRunner> oldRunners = this.cloudJettys.putIfAbsent(collection, cloudJettys);
- if (oldRunners != null) {
- // must close resources for the old entries
- for (CloudJettyRunner oldRunner : oldRunners) {
- IOUtils.closeQuietly(oldRunner.client);
- }
- }
-
- this.cloudJettys.put(collection, cloudJettys);
- this.shardToJetty.put(collection, shardToJetty);
- this.shardToLeaderJetty.put(collection, shardToLeaderJetty);
- } finally {
- cloudClient.close();
- }
- }
-
- /**
- * Wrapper around a {@link org.apache.solr.client.solrj.embedded.JettySolrRunner} to map the jetty
- * instance to various information of the cloud cluster, such as the collection and shard
- * that is served by the jetty instance, the node name, core node name, url, etc.
- */
- public static class CloudJettyRunner {
-
- public JettySolrRunner jetty;
- public String nodeName;
- public String coreNodeName;
- public String url;
- public SolrClient client;
- public Replica info;
- public String shard;
- public String collection;
-
- public CloudJettyRunner(JettySolrRunner jetty, Replica replica,
- String collection, String shard, String coreNodeName) {
- this.jetty = jetty;
- this.info = replica;
- this.collection = collection;
-
- Properties nodeProperties = jetty.getNodeProperties();
-
- // we need to update the jetty's shard so that it registers itself to the right shard when restarted
- this.shard = shard;
- nodeProperties.setProperty(CoreDescriptor.CORE_SHARD, this.shard);
-
- // we need to update the jetty's shard so that it registers itself under the right core name when restarted
- this.coreNodeName = coreNodeName;
- nodeProperties.setProperty(CoreDescriptor.CORE_NODE_NAME, this.coreNodeName);
-
- this.nodeName = replica.getNodeName();
-
- ZkCoreNodeProps coreNodeProps = new ZkCoreNodeProps(info);
- this.url = coreNodeProps.getCoreUrl();
-
- // strip the trailing slash as this can cause issues when executing requests
- this.client = createNewSolrServer(this.url.substring(0, this.url.length() - 1));
- }
-
- @Override
- public int hashCode() {
- final int prime = 31;
- int result = 1;
- result = prime * result + ((url == null) ? 0 : url.hashCode());
- return result;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj) return true;
- if (obj == null) return false;
- if (getClass() != obj.getClass()) return false;
- CloudJettyRunner other = (CloudJettyRunner) obj;
- if (url == null) {
- if (other.url != null) return false;
- } else if (!url.equals(other.url)) return false;
- return true;
- }
-
- @Override
- public String toString() {
- return "CloudJettyRunner [url=" + url + "]";
- }
-
- }
-
- protected static SolrClient createNewSolrServer(String baseUrl) {
- try {
- // setup the server...
- HttpSolrClient s = getHttpSolrClient(baseUrl, DEFAULT_CONNECTION_TIMEOUT);
- return s;
- } catch (Exception ex) {
- throw new RuntimeException(ex);
- }
- }
-
- protected void waitForBootstrapToComplete(String collectionName, String shardId) throws Exception {
- NamedList rsp;// we need to wait until bootstrap is complete otherwise the replicator thread will never start
- TimeOut timeOut = new TimeOut(60, TimeUnit.SECONDS);
- while (!timeOut.hasTimedOut()) {
- rsp = invokeCdcrAction(shardToLeaderJetty.get(collectionName).get(shardId), CdcrParams.CdcrAction.BOOTSTRAP_STATUS);
- if (rsp.get(RESPONSE_STATUS).toString().equals(COMPLETED)) {
- break;
- }
- Thread.sleep(1000);
- }
- }
-
- protected void waitForReplicationToComplete(String collectionName, String shardId) throws Exception {
- int cnt = 15;
- while (cnt > 0) {
- log.info("Checking queue size @ {}:{}", collectionName, shardId);
- long size = this.getQueueSize(collectionName, shardId);
- if (size == 0) { // if we received -1, it means that the log reader is not yet initialised, we should wait
- return;
- }
- log.info("Waiting for replication to complete. Queue size: {} @ {}:{}", size, collectionName, shardId);
- cnt--;
- Thread.sleep(1000); // wait a bit for the replication to complete
- }
- throw new RuntimeException("Timeout waiting for CDCR replication to complete @" + collectionName + ":" + shardId);
- }
-
- protected long getQueueSize(String collectionName, String shardId) throws Exception {
- NamedList rsp = this.invokeCdcrAction(shardToLeaderJetty.get(collectionName).get(shardId), CdcrParams.CdcrAction.QUEUES);
- NamedList host = (NamedList) ((NamedList) rsp.get(CdcrParams.QUEUES)).getVal(0);
- NamedList status = (NamedList) host.get(TARGET_COLLECTION);
- return (Long) status.get(CdcrParams.QUEUE_SIZE);
- }
-
- protected CollectionInfo collectInfo(String collection) throws Exception {
- CollectionInfo info = new CollectionInfo(collection);
- for (String shard : shardToJetty.get(collection).keySet()) {
- List<CloudJettyRunner> jettyRunners = shardToJetty.get(collection).get(shard);
- for (CloudJettyRunner jettyRunner : jettyRunners) {
- for (SolrCore core : jettyRunner.jetty.getCoreContainer().getCores()) {
- info.addCore(core, shard, shardToLeaderJetty.get(collection).containsValue(jettyRunner));
- }
- }
- }
-
- return info;
- }
-
- protected static class CollectionInfo {
-
- List<CoreInfo> coreInfos = new ArrayList<>();
-
- String collection;
-
- CollectionInfo(String collection) {
- this.collection = collection;
- }
-
- /**
- * @return Returns a map shard -> list of cores
- */
- Map<String, List<CoreInfo>> getShardToCoresMap() {
- Map<String, List<CoreInfo>> map = new HashMap<>();
- for (CoreInfo info : coreInfos) {
- List<CoreInfo> list = map.get(info.shard);
- if (list == null) {
- list = new ArrayList<>();
- map.put(info.shard, list);
- }
- list.add(info);
- }
- return map;
- }
-
- CoreInfo getLeader(String shard) {
- List<CoreInfo> coreInfos = getShardToCoresMap().get(shard);
- for (CoreInfo info : coreInfos) {
- if (info.isLeader) {
- return info;
- }
- }
- assertTrue(String.format(Locale.ENGLISH, "There is no leader for collection %s shard %s", collection, shard), false);
- return null;
- }
-
- List<CoreInfo> getReplicas(String shard) {
- List<CoreInfo> coreInfos = getShardToCoresMap().get(shard);
- coreInfos.remove(getLeader(shard));
- return coreInfos;
- }
-
- void addCore(SolrCore core, String shard, boolean isLeader) throws Exception {
- CoreInfo info = new CoreInfo();
- info.collectionName = core.getName();
- info.shard = shard;
- info.isLeader = isLeader;
- info.ulogDir = core.getUpdateHandler().getUpdateLog().getLogDir();
-
- this.coreInfos.add(info);
- }
-
- public static class CoreInfo {
- String collectionName;
- String shard;
- boolean isLeader;
- String ulogDir;
- }
-
- }
-
-}
-
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a30c92a7/solr/core/src/test/org/apache/solr/cloud/CdcrBootstrapTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/CdcrBootstrapTest.java b/solr/core/src/test/org/apache/solr/cloud/CdcrBootstrapTest.java
deleted file mode 100644
index 580c047..0000000
--- a/solr/core/src/test/org/apache/solr/cloud/CdcrBootstrapTest.java
+++ /dev/null
@@ -1,356 +0,0 @@
-/*
- * 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.List;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.lucene.store.FSDirectory;
-import org.apache.solr.SolrTestCaseJ4;
-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.request.AbstractUpdateRequest;
-import org.apache.solr.client.solrj.request.CollectionAdminRequest;
-import org.apache.solr.client.solrj.request.UpdateRequest;
-import org.apache.solr.client.solrj.response.QueryResponse;
-import org.apache.solr.common.SolrInputDocument;
-import org.apache.solr.common.params.CommonParams;
-import org.apache.solr.common.params.ModifiableSolrParams;
-import org.apache.solr.common.util.NamedList;
-import org.apache.solr.handler.CdcrParams;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class CdcrBootstrapTest extends SolrTestCaseJ4 {
-
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- /**
- * Starts a source cluster with no CDCR configuration, indexes enough documents such that
- * the at least one old tlog is closed and thrown away so that the source cluster does not have
- * all updates available in tlogs only.
- * <p>
- * Then we start a target cluster with CDCR configuration and we change the source cluster configuration
- * to use CDCR (i.e. CdcrUpdateLog, CdcrRequestHandler and CdcrUpdateProcessor) and restart it.
- * <p>
- * We test that all updates eventually make it to the target cluster and that the collectioncheckpoint
- * call returns the same version as the last update indexed on the source.
- */
- @Test
- public void testConvertClusterToCdcrAndBootstrap() throws Exception {
- // start the target first so that we know its zkhost
- MiniSolrCloudCluster target = new MiniSolrCloudCluster(1, createTempDir("cdcr-target"), buildJettyConfig("/solr"));
- try {
- target.waitForAllNodes(30);
- log.info("Target zkHost = " + target.getZkServer().getZkAddress());
- System.setProperty("cdcr.target.zkHost", target.getZkServer().getZkAddress());
-
- // start a cluster with no cdcr
- MiniSolrCloudCluster source = new MiniSolrCloudCluster(1, createTempDir("cdcr-source"), buildJettyConfig("/solr"));
- try {
- source.waitForAllNodes(30);
- source.uploadConfigSet(configset("cdcr-source-disabled"), "cdcr-source");
-
- // create a collection with the cdcr-source-disabled configset
- CollectionAdminRequest.createCollection("cdcr-source", "cdcr-source", 1, 1)
- // todo investigate why this is necessary??? because by default it selects a ram directory which deletes the tlogs on reloads?
- .withProperty("solr.directoryFactory", "solr.StandardDirectoryFactory")
- .process(source.getSolrClient());
-
- CloudSolrClient sourceSolrClient = source.getSolrClient();
- int docs = (TEST_NIGHTLY ? 100 : 10);
- int numDocs = indexDocs(sourceSolrClient, "cdcr-source", docs);
-
- QueryResponse response = sourceSolrClient.query(new SolrQuery("*:*"));
- assertEquals("", numDocs, response.getResults().getNumFound());
-
- // lets find and keep the maximum version assigned by source cluster across all our updates
- long maxVersion = Long.MIN_VALUE;
- ModifiableSolrParams params = new ModifiableSolrParams();
- params.set(CommonParams.QT, "/get");
- params.set("getVersions", numDocs);
- response = sourceSolrClient.query(params);
- List<Long> versions = (List<Long>) response.getResponse().get("versions");
- for (Long version : versions) {
- maxVersion = Math.max(maxVersion, version);
- }
-
-// upload the cdcr-enabled config and restart source cluster
- source.uploadConfigSet(configset("cdcr-source"), "cdcr-source");
- JettySolrRunner runner = source.stopJettySolrRunner(0);
- source.startJettySolrRunner(runner);
- assertTrue(runner.isRunning());
- AbstractDistribZkTestBase.waitForRecoveriesToFinish("cdcr-source", source.getSolrClient().getZkStateReader(), true, true, 330);
-
- response = sourceSolrClient.query(new SolrQuery("*:*"));
- assertEquals("Document mismatch on source after restart", numDocs, response.getResults().getNumFound());
-
- // setup the target cluster
- target.uploadConfigSet(configset("cdcr-target"), "cdcr-target");
- CollectionAdminRequest.createCollection("cdcr-target", "cdcr-target", 1, 1)
- .process(target.getSolrClient());
- CloudSolrClient targetSolrClient = target.getSolrClient();
- targetSolrClient.setDefaultCollection("cdcr-target");
- Thread.sleep(1000);
-
- cdcrStart(targetSolrClient);
- cdcrStart(sourceSolrClient);
-
- response = getCdcrQueue(sourceSolrClient);
- log.info("Cdcr queue response: " + response.getResponse());
- long foundDocs = waitForTargetToSync(numDocs, targetSolrClient);
- assertEquals("Document mismatch on target after sync", numDocs, foundDocs);
-
- params = new ModifiableSolrParams();
- params.set(CommonParams.ACTION, CdcrParams.CdcrAction.COLLECTIONCHECKPOINT.toString());
- params.set(CommonParams.QT, "/cdcr");
- response = targetSolrClient.query(params);
- Long checkpoint = (Long) response.getResponse().get(CdcrParams.CHECKPOINT);
- assertNotNull(checkpoint);
- assertEquals("COLLECTIONCHECKPOINT from target cluster should have returned the maximum " +
- "version across all updates made to source", maxVersion, checkpoint.longValue());
- } finally {
- source.shutdown();
- }
- } finally {
- target.shutdown();
- }
- }
-
- private int indexDocs(CloudSolrClient sourceSolrClient, String collection, int batches) throws IOException, SolrServerException {
- sourceSolrClient.setDefaultCollection(collection);
- int numDocs = 0;
- for (int k = 0; k < batches; k++) {
- UpdateRequest req = new UpdateRequest();
- for (; numDocs < (k + 1) * 100; numDocs++) {
- SolrInputDocument doc = new SolrInputDocument();
- doc.addField("id", "source_" + numDocs);
- doc.addField("xyz", numDocs);
- req.add(doc);
- }
- req.setAction(AbstractUpdateRequest.ACTION.COMMIT, true, true);
- req.process(sourceSolrClient);
- }
- log.info("Adding numDocs=" + numDocs);
- return numDocs;
- }
- /**
- * This test start cdcr source, adds data,starts target cluster, verifies replication,
- * stops cdcr replication and buffering, adds more data, re-enables cdcr and verify replication
- */
- public void testBootstrapWithSourceCluster() throws Exception {
- // start the target first so that we know its zkhost
- MiniSolrCloudCluster target = new MiniSolrCloudCluster(1, createTempDir("cdcr-target"), buildJettyConfig("/solr"));
- try {
- target.waitForAllNodes(30);
- System.out.println("Target zkHost = " + target.getZkServer().getZkAddress());
- System.setProperty("cdcr.target.zkHost", target.getZkServer().getZkAddress());
-
- MiniSolrCloudCluster source = new MiniSolrCloudCluster(1, createTempDir("cdcr-source"), buildJettyConfig("/solr"));
- try {
- source.waitForAllNodes(30);
- source.uploadConfigSet(configset("cdcr-source"), "cdcr-source");
-
- CollectionAdminRequest.createCollection("cdcr-source", "cdcr-source", 1, 1)
- .withProperty("solr.directoryFactory", "solr.StandardDirectoryFactory")
- .process(source.getSolrClient());
-
- CloudSolrClient sourceSolrClient = source.getSolrClient();
- int docs = (TEST_NIGHTLY ? 100 : 10);
- int numDocs = indexDocs(sourceSolrClient, "cdcr-source", docs);
-
- QueryResponse response = sourceSolrClient.query(new SolrQuery("*:*"));
- assertEquals("", numDocs, response.getResults().getNumFound());
-
- // setup the target cluster
- target.uploadConfigSet(configset("cdcr-target"), "cdcr-target");
- CollectionAdminRequest.createCollection("cdcr-target", "cdcr-target", 1, 1)
- .process(target.getSolrClient());
- CloudSolrClient targetSolrClient = target.getSolrClient();
- targetSolrClient.setDefaultCollection("cdcr-target");
-
- cdcrStart(targetSolrClient);
- cdcrStart(sourceSolrClient);
-
- response = getCdcrQueue(sourceSolrClient);
- log.info("Cdcr queue response: " + response.getResponse());
- long foundDocs = waitForTargetToSync(numDocs, targetSolrClient);
- assertEquals("Document mismatch on target after sync", numDocs, foundDocs);
-
- int total_tlogs_in_index = FSDirectory.open(target.getBaseDir().resolve("node1").
- resolve("cdcr-target_shard1_replica_n1").resolve("data").
- resolve("tlog")).listAll().length;
-
- assertEquals("tlogs count should be ZERO",0, total_tlogs_in_index);
-
- cdcrStop(sourceSolrClient);
- cdcrDisableBuffer(sourceSolrClient);
-
- int c = 0;
- for (int k = 0; k < 10; k++) {
- UpdateRequest req = new UpdateRequest();
- for (; c < (k + 1) * 100; c++, numDocs++) {
- SolrInputDocument doc = new SolrInputDocument();
- doc.addField("id", "source_" + numDocs);
- doc.addField("xyz", numDocs);
- req.add(doc);
- }
- req.setAction(AbstractUpdateRequest.ACTION.COMMIT, true, true);
- log.info("Adding 100 docs with commit=true, numDocs=" + numDocs);
- req.process(sourceSolrClient);
- }
-
- response = sourceSolrClient.query(new SolrQuery("*:*"));
- assertEquals("", numDocs, response.getResults().getNumFound());
-
- cdcrStart(sourceSolrClient);
- cdcrEnableBuffer(sourceSolrClient);
-
- foundDocs = waitForTargetToSync(numDocs, targetSolrClient);
- assertEquals("Document mismatch on target after sync", numDocs, foundDocs);
-
- } finally {
- source.shutdown();
- }
- } finally {
- target.shutdown();
- }
- }
-
- public void testBootstrapWithContinousIndexingOnSourceCluster() throws Exception {
- // start the target first so that we know its zkhost
- MiniSolrCloudCluster target = new MiniSolrCloudCluster(1, createTempDir("cdcr-target"), buildJettyConfig("/solr"));
- target.waitForAllNodes(30);
- try {
- log.info("Target zkHost = " + target.getZkServer().getZkAddress());
- System.setProperty("cdcr.target.zkHost", target.getZkServer().getZkAddress());
-
- MiniSolrCloudCluster source = new MiniSolrCloudCluster(1, createTempDir("cdcr-source"), buildJettyConfig("/solr"));
- try {
- source.waitForAllNodes(30);
- source.uploadConfigSet(configset("cdcr-source"), "cdcr-source");
-
- CollectionAdminRequest.createCollection("cdcr-source", "cdcr-source", 1, 1)
- .withProperty("solr.directoryFactory", "solr.StandardDirectoryFactory")
- .process(source.getSolrClient());
-
- CloudSolrClient sourceSolrClient = source.getSolrClient();
- int docs = (TEST_NIGHTLY ? 100 : 10);
- int numDocs = indexDocs(sourceSolrClient, "cdcr-source", docs);
-
- QueryResponse response = sourceSolrClient.query(new SolrQuery("*:*"));
- assertEquals("", numDocs, response.getResults().getNumFound());
-
- // setup the target cluster
- target.uploadConfigSet(configset("cdcr-target"), "cdcr-target");
- CollectionAdminRequest.createCollection("cdcr-target", "cdcr-target", 1, 1)
- .process(target.getSolrClient());
- CloudSolrClient targetSolrClient = target.getSolrClient();
- targetSolrClient.setDefaultCollection("cdcr-target");
- Thread.sleep(1000);
-
- cdcrStart(targetSolrClient);
- cdcrStart(sourceSolrClient);
- int c = 0;
- for (int k = 0; k < docs; k++) {
- UpdateRequest req = new UpdateRequest();
- for (; c < (k + 1) * 100; c++, numDocs++) {
- SolrInputDocument doc = new SolrInputDocument();
- doc.addField("id", "source_" + numDocs);
- doc.addField("xyz", numDocs);
- req.add(doc);
- }
- req.setAction(AbstractUpdateRequest.ACTION.COMMIT, true, true);
- log.info("Adding " + docs + " docs with commit=true, numDocs=" + numDocs);
- req.process(sourceSolrClient);
- }
-
- response = sourceSolrClient.query(new SolrQuery("*:*"));
- assertEquals("", numDocs, response.getResults().getNumFound());
-
- response = getCdcrQueue(sourceSolrClient);
- log.info("Cdcr queue response: " + response.getResponse());
- long foundDocs = waitForTargetToSync(numDocs, targetSolrClient);
- assertEquals("Document mismatch on target after sync", numDocs, foundDocs);
-
- } finally {
- source.shutdown();
- }
- } finally {
- target.shutdown();
- }
- }
-
- private long waitForTargetToSync(int numDocs, CloudSolrClient targetSolrClient) throws SolrServerException, IOException, InterruptedException {
- long start = System.nanoTime();
- QueryResponse response = null;
- while (System.nanoTime() - start <= TimeUnit.NANOSECONDS.convert(120, TimeUnit.SECONDS)) {
- try {
- targetSolrClient.commit();
- response = targetSolrClient.query(new SolrQuery("*:*"));
- if (response.getResults().getNumFound() == numDocs) {
- break;
- }
- } catch (Exception e) {
- log.warn("Exception trying to commit on target. This is expected and safe to ignore.", e);
- }
- Thread.sleep(1000);
- }
- return response != null ? response.getResults().getNumFound() : 0;
- }
-
-
- private void cdcrStart(CloudSolrClient client) throws SolrServerException, IOException {
- QueryResponse response = invokeCdcrAction(client, CdcrParams.CdcrAction.START);
- assertEquals("started", ((NamedList) response.getResponse().get("status")).get("process"));
- }
-
- private void cdcrStop(CloudSolrClient client) throws SolrServerException, IOException {
- QueryResponse response = invokeCdcrAction(client, CdcrParams.CdcrAction.STOP);
- assertEquals("stopped", ((NamedList) response.getResponse().get("status")).get("process"));
- }
-
- private void cdcrEnableBuffer(CloudSolrClient client) throws IOException, SolrServerException {
- QueryResponse response = invokeCdcrAction(client, CdcrParams.CdcrAction.ENABLEBUFFER);
- assertEquals("enabled", ((NamedList) response.getResponse().get("status")).get("buffer"));
- }
-
- private void cdcrDisableBuffer(CloudSolrClient client) throws IOException, SolrServerException {
- QueryResponse response = invokeCdcrAction(client, CdcrParams.CdcrAction.DISABLEBUFFER);
- assertEquals("disabled", ((NamedList) response.getResponse().get("status")).get("buffer"));
- }
-
- private QueryResponse invokeCdcrAction(CloudSolrClient client, CdcrParams.CdcrAction action) throws IOException, SolrServerException {
- ModifiableSolrParams params = new ModifiableSolrParams();
- params.set(CommonParams.QT, "/cdcr");
- params.set(CommonParams.ACTION, action.toLower());
- return client.query(params);
- }
-
- private QueryResponse getCdcrQueue(CloudSolrClient client) throws SolrServerException, IOException {
- ModifiableSolrParams params = new ModifiableSolrParams();
- params.set(CommonParams.QT, "/cdcr");
- params.set(CommonParams.ACTION, CdcrParams.QUEUES);
- return client.query(params);
- }
-}