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/10/31 18:43:29 UTC
[5/5] lucene-solr:master: SOLR-11003
SOLR-11003
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/bc46de3b
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/bc46de3b
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/bc46de3b
Branch: refs/heads/master
Commit: bc46de3b2a07b77d40e85e1609d0ee04eacbcf76
Parents: d58ec87
Author: Varun Thacker <va...@apache.org>
Authored: Mon Oct 30 09:50:50 2017 -0500
Committer: Varun Thacker <va...@apache.org>
Committed: Tue Oct 31 13:43:19 2017 -0500
----------------------------------------------------------------------
solr/CHANGES.txt | 3 +
.../org/apache/solr/handler/CdcrReplicator.java | 28 +
.../apache/solr/update/CdcrTransactionLog.java | 144 +++
.../org/apache/solr/update/TransactionLog.java | 2 +-
.../configsets/cdcr-cluster1/conf/schema.xml | 32 +
.../cdcr-cluster1/conf/solrconfig.xml | 80 ++
.../configsets/cdcr-cluster2/conf/schema.xml | 32 +
.../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 | 238 +++++
.../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 | 94 ++
.../cloud/cdcr/CdcrVersionReplicationTest.java | 307 +++++++
22 files changed, 3404 insertions(+), 2722 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/bc46de3b/solr/CHANGES.txt
----------------------------------------------------------------------
diff --git a/solr/CHANGES.txt b/solr/CHANGES.txt
index b4d7bf2..c1b8f98 100644
--- a/solr/CHANGES.txt
+++ b/solr/CHANGES.txt
@@ -63,6 +63,9 @@ New Features
* SOLR-11520: Implement autoscaling suggestions for cores count violations (noble)
+* 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/bc46de3b/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/bc46de3b/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/bc46de3b/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/bc46de3b/solr/core/src/test-files/solr/configsets/cdcr-cluster1/conf/schema.xml
----------------------------------------------------------------------
diff --git a/solr/core/src/test-files/solr/configsets/cdcr-cluster1/conf/schema.xml b/solr/core/src/test-files/solr/configsets/cdcr-cluster1/conf/schema.xml
new file mode 100644
index 0000000..19b506f
--- /dev/null
+++ b/solr/core/src/test-files/solr/configsets/cdcr-cluster1/conf/schema.xml
@@ -0,0 +1,32 @@
+<?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="text" class="solr.TextField"/>
+ <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="_text_" type="text" indexed="true" stored="true" multiValued="true" />
+ <field name="_version_" type="long" indexed="true" stored="true"/>
+ <dynamicField name="*" type="text" indexed="true" stored="true"/>
+ </fields>
+ <copyField source="*" dest="_text_"/>
+ <uniqueKey>id</uniqueKey>
+</schema>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/bc46de3b/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/bc46de3b/solr/core/src/test-files/solr/configsets/cdcr-cluster2/conf/schema.xml
----------------------------------------------------------------------
diff --git a/solr/core/src/test-files/solr/configsets/cdcr-cluster2/conf/schema.xml b/solr/core/src/test-files/solr/configsets/cdcr-cluster2/conf/schema.xml
new file mode 100644
index 0000000..648c0f9
--- /dev/null
+++ b/solr/core/src/test-files/solr/configsets/cdcr-cluster2/conf/schema.xml
@@ -0,0 +1,32 @@
+<?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="text" class="solr.TextField"/>
+ <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="_text_" type="text" indexed="true" stored="true" multiValued="true"/>
+ <field name="_version_" type="long" indexed="true" stored="true"/>
+ <dynamicField name="*" type="text" indexed="true" stored="true"/>
+ </fields>
+ <copyField source="*" dest="_text_"/>
+ <uniqueKey>id</uniqueKey>
+</schema>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/bc46de3b/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/bc46de3b/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/bc46de3b/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);
- }
-}