You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by is...@apache.org on 2020/12/16 10:18:14 UTC
[lucene-solr] branch jira/solr-15052-8x created (now 3c02c91)
This is an automated email from the ASF dual-hosted git repository.
ishan pushed a change to branch jira/solr-15052-8x
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git.
at 3c02c91 SOLR-15052: Per-replica states for reducing overseer bottlenecks
This branch includes the following new commits:
new 3c02c91 SOLR-15052: Per-replica states for reducing overseer bottlenecks
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[lucene-solr] 01/01: SOLR-15052: Per-replica states for reducing
overseer bottlenecks
Posted by is...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ishan pushed a commit to branch jira/solr-15052-8x
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit 3c02c91973763973138db7c874f9fca11d88f492
Author: Ishan Chattopadhyaya <is...@apache.org>
AuthorDate: Wed Dec 16 15:45:41 2020 +0530
SOLR-15052: Per-replica states for reducing overseer bottlenecks
Co-authored-by: Noble Paul <no...@apache.org>
---
.../solr/cloud/ShardLeaderElectionContextBase.java | 15 +-
.../java/org/apache/solr/cloud/ZkController.java | 38 +-
.../cloud/api/collections/CreateCollectionCmd.java | 1 +
.../OverseerCollectionMessageHandler.java | 1 +
.../solr/cloud/overseer/CollectionMutator.java | 28 +-
.../apache/solr/cloud/overseer/NodeMutator.java | 11 +-
.../apache/solr/cloud/overseer/ReplicaMutator.java | 21 +-
.../apache/solr/cloud/overseer/SliceMutator.java | 47 +-
.../apache/solr/cloud/overseer/ZkStateWriter.java | 79 ++-
.../apache/solr/cloud/overseer/ZkWriteCommand.java | 20 +
.../solr/handler/admin/CollectionsHandler.java | 7 +-
.../apache/solr/cloud/CollectionsAPISolrJTest.java | 11 +-
.../test/org/apache/solr/cloud/SplitShardTest.java | 3 +
.../solr/cloud/api/collections/ShardSplitTest.java | 12 +-
.../solr/handler/PingRequestHandlerTest.java | 2 +
.../apache/solr/handler/TestContainerPlugin.java | 29 +
.../org/apache/solr/handler/TestSQLHandler.java | 4 +-
.../solr/handler/TestStressThreadBackup.java | 5 +-
.../solr/handler/admin/HealthCheckHandlerTest.java | 1 +
.../solr/handler/admin/IndexSizeEstimatorTest.java | 1 +
.../handler/admin/MetricsHistoryHandlerTest.java | 3 +-
.../component/CustomHighlightComponentTest.java | 3 +-
.../DistributedQueryComponentOptimizationTest.java | 1 +
.../solr/handler/component/SearchHandlerTest.java | 4 +
.../reporters/SolrJmxReporterCloudTest.java | 1 +
.../reporters/solr/SolrCloudReportersTest.java | 2 +
.../src/test/org/apache/solr/pkg/TestPackages.java | 1 +
.../transform/TestSubQueryTransformerDistrib.java | 2 +
.../schema/ManagedSchemaRoundRobinCloudTest.java | 1 +
.../PreAnalyzedFieldManagedSchemaCloudTest.java | 1 +
.../apache/solr/schema/TestManagedSchemaAPI.java | 2 +
.../solr/search/CurrencyRangeFacetCloudTest.java | 1 +
.../solr/search/facet/RangeFacetCloudTest.java | 1 +
.../search/facet/TestCloudJSONFacetJoinDomain.java | 1 +
.../solr/search/facet/TestCloudJSONFacetSKG.java | 1 +
.../search/facet/TestCloudJSONFacetSKGEquiv.java | 1 +
.../search/join/BlockJoinFacetDistribTest.java | 1 +
.../search/join/CrossCollectionJoinQueryTest.java | 2 +
.../apache/solr/search/stats/TestDistribIDF.java | 3 +-
.../solr/servlet/HttpSolrCallGetCoreTest.java | 1 +
.../update/TestInPlaceUpdateWithRouteField.java | 1 +
.../update/processor/AtomicUpdateJavabinTest.java | 370 -------------
.../DimensionalRoutedAliasUpdateProcessorTest.java | 3 +
.../processor/TemplateUpdateProcessorTest.java | 2 +-
.../test/org/apache/solr/util/TestExportTool.java | 2 +
.../solr/util/tracing/TestDistributedTracing.java | 1 +
.../client/solrj/cloud/DistribStateManager.java | 7 +
.../client/solrj/impl/SolrClientCloudManager.java | 3 +
.../client/solrj/impl/ZkDistribStateManager.java | 6 +
.../solrj/request/CollectionAdminRequest.java | 8 +
.../org/apache/solr/common/cloud/ClusterState.java | 71 +++
.../apache/solr/common/cloud/DocCollection.java | 74 ++-
.../apache/solr/common/cloud/PerReplicaStates.java | 587 +++++++++++++++++++++
.../java/org/apache/solr/common/cloud/Replica.java | 84 ++-
.../java/org/apache/solr/common/cloud/Slice.java | 26 +-
.../org/apache/solr/common/cloud/SolrZkClient.java | 12 +
.../apache/solr/common/cloud/ZkStateReader.java | 132 ++++-
.../IndexingNestedDocuments.java | 4 +-
.../JsonRequestApiHeatmapFacetingTest.java | 4 +-
.../ref_guide_examples/JsonRequestApiTest.java | 3 +-
.../UsingSolrJRefGuideExamplesTest.java | 1 +
.../client/solrj/impl/CloudSolrClientTest.java | 221 ++++----
.../solrj/io/stream/CloudAuthStreamTest.java | 3 +-
.../client/solrj/io/stream/JDBCStreamTest.java | 4 +-
.../client/solrj/io/stream/MathExpressionTest.java | 5 +-
.../solrj/io/stream/SelectWithEvaluatorsTest.java | 4 +-
.../solrj/io/stream/StreamDecoratorTest.java | 51 +-
...ectJsonQueryRequestFacetingIntegrationTest.java | 4 +-
.../JsonQueryRequestFacetingIntegrationTest.java | 4 +-
.../json/JsonQueryRequestHeatmapFacetingTest.java | 4 +-
.../cloud/TestCloudCollectionsListeners.java | 4 +
.../common/cloud/TestCollectionStateWatchers.java | 22 +-
.../solr/common/cloud/TestPerReplicaStates.java | 137 +++++
.../org/apache/solr/cloud/SolrCloudTestCase.java | 1 +
74 files changed, 1676 insertions(+), 558 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContextBase.java b/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContextBase.java
index b010de2..5f74855 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContextBase.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ShardLeaderElectionContextBase.java
@@ -25,6 +25,8 @@ import java.util.ArrayList;
import org.apache.solr.cloud.overseer.OverseerAction;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrException.ErrorCode;
+import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.PerReplicaStates;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkCmdExecutor;
@@ -161,9 +163,11 @@ class ShardLeaderElectionContextBase extends ElectionContext {
assert shardId != null;
boolean isAlreadyLeader = false;
+ String currentLeader = null;
if (zkStateReader.getClusterState() != null &&
zkStateReader.getClusterState().getCollection(collection).getSlice(shardId).getReplicas().size() < 2) {
Replica leader = zkStateReader.getLeader(collection, shardId);
+ if(leader != null) currentLeader = leader.getName();
if (leader != null
&& leader.getNodeName().equals(leaderProps.get(ZkStateReader.NODE_NAME_PROP))
&& leader.getCoreName().equals(leaderProps.get(ZkStateReader.CORE_NAME_PROP))) {
@@ -179,7 +183,16 @@ class ShardLeaderElectionContextBase extends ElectionContext {
ZkStateReader.STATE_PROP, Replica.State.ACTIVE.toString());
assert zkController != null;
assert zkController.getOverseer() != null;
- zkController.getOverseer().offerStateUpdate(Utils.toJSON(m));
+ DocCollection coll = zkStateReader.getCollection(this.collection);
+ if (coll == null || coll.getStateFormat() < 2 || ZkController.sendToOverseer(coll, id)) {
+ zkController.getOverseer().offerStateUpdate(Utils.toJSON(m));
+ } else {
+ PerReplicaStates prs = PerReplicaStates.fetch(coll.getZNode(), zkClient, coll.getPerReplicaStates());
+ PerReplicaStates.WriteOps writeOps = PerReplicaStates.WriteOps.flipLeader(zkStateReader.getClusterState().getCollection(collection).getSlice(shardId).getReplicaNames(), id, prs);
+ //nocommit make this debug
+ log.info("bypassed Zookeeper for leader election for {}/{}, old:{},new {} ", this.collection, shardId, currentLeader, id);
+ PerReplicaStates.persist(writeOps, coll.getZNode(), zkStateReader.getZkClient());
+ }
}
}
diff --git a/solr/core/src/java/org/apache/solr/cloud/ZkController.java b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
index 6cc2afe..994c2dc 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkController.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
@@ -374,7 +374,7 @@ public class ZkController implements Closeable {
}
cc.cancelCoreRecoveries();
-
+
try {
registerAllCoresAsDown(registerOnReconnect, false);
} catch (SessionExpiredException e) {
@@ -384,7 +384,7 @@ public class ZkController implements Closeable {
// this is really best effort - in case of races or failure cases where we now need to be the leader, if anything fails,
// just continue
log.warn("Exception while trying to register all cores as DOWN", e);
- }
+ }
// we have to register as live first to pick up docs in the buffer
createEphemeralLiveNode();
@@ -474,6 +474,7 @@ public class ZkController implements Closeable {
zkStateReader = new ZkStateReader(zkClient, () -> {
if (cc != null) cc.securityNodeChanged();
});
+ zkStateReader.nodeName = nodeName;
init(registerOnReconnect);
@@ -1609,12 +1610,41 @@ public class ZkController implements Closeable {
if (updateLastState) {
cd.getCloudDescriptor().setLastPublished(state);
}
- overseerJobQueue.offer(Utils.toJSON(m));
+ DocCollection coll = zkStateReader.getCollection(collection);
+ if(forcePublish || sendToOverseer(coll, coreNodeName)) {
+ overseerJobQueue.offer(Utils.toJSON(m));
+ } else {
+// if(log.isInfoEnabled()) {
+ //nocommit make this debug
+ log.info("bypassed overseer for message : {}", Utils.toJSONString(m));
+// }
+ PerReplicaStates perReplicaStates = PerReplicaStates.fetch(coll.getZNode(), zkClient, coll.getPerReplicaStates());
+ PerReplicaStates.WriteOps ops = PerReplicaStates.WriteOps.flipState(coreNodeName, state, perReplicaStates);
+ PerReplicaStates.persist(ops, coll.getZNode(), zkClient);
+ }
} finally {
MDCLoggingContext.clear();
}
}
+ /**
+ * Whether a message needs to be sent to overseer or not
+ */
+ static boolean sendToOverseer(DocCollection coll, String replicaName) {
+ if (coll == null) return true;
+ if (coll.getStateFormat() < 2 || !coll.isPerReplicaState()) return true;
+ Replica r = coll.getReplica(replicaName);
+ if (r == null) return true;
+ Slice shard = coll.getSlice(r.slice);
+ if (shard == null) return true;//very unlikely
+ if (shard.getState() == Slice.State.RECOVERY) return true;
+ if (shard.getParent() != null) return true;
+ for (Slice slice : coll.getSlices()) {
+ if (Objects.equals(shard.getName(), slice.getParent())) return true;
+ }
+ return false;
+ }
+
public ZkShardTerms getShardTerms(String collection, String shardId) {
return getCollectionTerms(collection).getShard(shardId);
}
@@ -2198,7 +2228,7 @@ public class ZkController implements Closeable {
try {
if (electionNode != null) {
// Check whether we came to this node by mistake
- if ( overseerElector.getContext() != null && overseerElector.getContext().leaderSeqPath == null
+ if ( overseerElector.getContext() != null && overseerElector.getContext().leaderSeqPath == null
&& !overseerElector.getContext().leaderSeqPath.endsWith(electionNode)) {
log.warn("Asked to rejoin with wrong election node : {}, current node is {}", electionNode, overseerElector.getContext().leaderSeqPath);
//however delete it . This is possible when the last attempt at deleting the election node failed.
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateCollectionCmd.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateCollectionCmd.java
index 89f94a9..dfe17a4 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateCollectionCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/CreateCollectionCmd.java
@@ -182,6 +182,7 @@ public class CreateCollectionCmd implements OverseerCollectionMessageHandler.Cmd
if(created) break;
}
if (!created) {
+ ocmh.zkStateReader.debugCollectionState(collectionName);
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Could not fully create collection: " + collectionName);
}
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerCollectionMessageHandler.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerCollectionMessageHandler.java
index f4e349b..3c474e5 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerCollectionMessageHandler.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerCollectionMessageHandler.java
@@ -150,6 +150,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler,
ZkStateReader.PULL_REPLICAS, "0",
ZkStateReader.MAX_SHARDS_PER_NODE, "1",
ZkStateReader.AUTO_ADD_REPLICAS, "false",
+ DocCollection.PER_REPLICA_STATE, null,
DocCollection.RULE, null,
POLICY, null,
SNITCH, null,
diff --git a/solr/core/src/java/org/apache/solr/cloud/overseer/CollectionMutator.java b/solr/core/src/java/org/apache/solr/cloud/overseer/CollectionMutator.java
index 80e0e9e..c4201c1 100644
--- a/solr/core/src/java/org/apache/solr/cloud/overseer/CollectionMutator.java
+++ b/solr/core/src/java/org/apache/solr/cloud/overseer/CollectionMutator.java
@@ -24,12 +24,15 @@ import java.util.Map;
import org.apache.solr.client.solrj.cloud.DistribStateManager;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
+import org.apache.solr.client.solrj.impl.SolrClientCloudManager;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.ImplicitDocRouter;
+import org.apache.solr.common.cloud.PerReplicaStates;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.Slice;
+import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.util.Utils;
@@ -46,10 +49,12 @@ public class CollectionMutator {
protected final SolrCloudManager cloudManager;
protected final DistribStateManager stateManager;
+ protected final SolrZkClient zkClient;
public CollectionMutator(SolrCloudManager cloudManager) {
this.cloudManager = cloudManager;
this.stateManager = cloudManager.getDistribStateManager();
+ this.zkClient = SliceMutator.getZkClient(cloudManager);
}
public ZkWriteCommand createShard(final ClusterState clusterState, ZkNodeProps message) {
@@ -107,7 +112,21 @@ public class CollectionMutator {
DocCollection coll = clusterState.getCollection(message.getStr(COLLECTION_PROP));
Map<String, Object> m = coll.shallowCopy();
boolean hasAnyOps = false;
+ PerReplicaStates.WriteOps replicaOps = null;
for (String prop : CollectionAdminRequest.MODIFIABLE_COLLECTION_PROPERTIES) {
+ if(prop.equals(DocCollection.PER_REPLICA_STATE)) {
+ String val = message.getStr(DocCollection.PER_REPLICA_STATE);
+ if(val == null) continue;
+ boolean enable = Boolean.parseBoolean(val);
+ if(enable == coll.isPerReplicaState()) {
+ //already enabled
+ log.error("trying to set perReplicaState to {} from {}", val, coll.isPerReplicaState());
+ continue;
+ }
+ replicaOps = PerReplicaStates.WriteOps.modifyCollection(coll, enable, PerReplicaStates.fetch(coll.getZNode(), zkClient, null));
+ }
+
+
if (message.containsKey(prop)) {
hasAnyOps = true;
if (message.get(prop) == null) {
@@ -136,8 +155,13 @@ public class CollectionMutator {
return ZkStateWriter.NO_OP;
}
- return new ZkWriteCommand(coll.getName(),
- new DocCollection(coll.getName(), coll.getSlicesMap(), m, coll.getRouter(), coll.getZNodeVersion(), coll.getZNode()));
+ DocCollection collection = new DocCollection(coll.getName(), coll.getSlicesMap(), m, coll.getRouter(), coll.getZNodeVersion(), coll.getZNode());
+ if(replicaOps == null){
+ return new ZkWriteCommand(coll.getName(), collection);
+ } else {
+ return new ZkWriteCommand(coll.getName(), collection, replicaOps, true);
+ }
+
}
public static DocCollection updateSlice(String collectionName, DocCollection collection, Slice slice) {
diff --git a/solr/core/src/java/org/apache/solr/cloud/overseer/NodeMutator.java b/solr/core/src/java/org/apache/solr/cloud/overseer/NodeMutator.java
index 56bcfd5..77f0550 100644
--- a/solr/core/src/java/org/apache/solr/cloud/overseer/NodeMutator.java
+++ b/solr/core/src/java/org/apache/solr/cloud/overseer/NodeMutator.java
@@ -26,6 +26,7 @@ import java.util.Map.Entry;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.PerReplicaStates;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.cloud.ZkNodeProps;
@@ -45,6 +46,8 @@ public class NodeMutator {
Map<String, DocCollection> collections = clusterState.getCollectionsMap();
for (Map.Entry<String, DocCollection> entry : collections.entrySet()) {
+ List<String> downedReplicas = new ArrayList<>();
+
String collection = entry.getKey();
DocCollection docCollection = entry.getValue();
@@ -68,6 +71,7 @@ public class NodeMutator {
Replica newReplica = new Replica(replica.getName(), props, collection, slice.getName());
newReplicas.put(replica.getName(), newReplica);
needToUpdateCollection = true;
+ downedReplicas.add(replica.getName());
}
}
@@ -76,7 +80,12 @@ public class NodeMutator {
}
if (needToUpdateCollection) {
- zkWriteCommands.add(new ZkWriteCommand(collection, docCollection.copyWithSlices(slicesCopy)));
+ if(docCollection.isPerReplicaState()) {
+ zkWriteCommands.add(new ZkWriteCommand(collection, docCollection.copyWithSlices(slicesCopy),
+ PerReplicaStates.WriteOps.downReplicas(downedReplicas, docCollection.getPerReplicaStates()), false));
+ } else {
+ zkWriteCommands.add(new ZkWriteCommand(collection, docCollection.copyWithSlices(slicesCopy)));
+ }
}
}
diff --git a/solr/core/src/java/org/apache/solr/cloud/overseer/ReplicaMutator.java b/solr/core/src/java/org/apache/solr/cloud/overseer/ReplicaMutator.java
index 7891cc1..97b3eaa 100644
--- a/solr/core/src/java/org/apache/solr/cloud/overseer/ReplicaMutator.java
+++ b/solr/core/src/java/org/apache/solr/cloud/overseer/ReplicaMutator.java
@@ -31,6 +31,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.solr.client.solrj.cloud.DistribStateManager;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
+import org.apache.solr.client.solrj.impl.SolrClientCloudManager;
import org.apache.solr.cloud.CloudUtil;
import org.apache.solr.cloud.Overseer;
import org.apache.solr.cloud.api.collections.Assign;
@@ -40,7 +41,9 @@ import org.apache.solr.common.SolrException;
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.PerReplicaStates;
import org.apache.solr.common.cloud.Slice;
+import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.util.Utils;
@@ -50,6 +53,7 @@ import org.slf4j.LoggerFactory;
import static org.apache.solr.cloud.overseer.CollectionMutator.checkCollectionKeyExistence;
import static org.apache.solr.cloud.overseer.CollectionMutator.checkKeyExistence;
+import static org.apache.solr.cloud.overseer.SliceMutator.getZkClient;
import static org.apache.solr.common.params.CommonParams.NAME;
public class ReplicaMutator {
@@ -57,10 +61,12 @@ public class ReplicaMutator {
protected final SolrCloudManager cloudManager;
protected final DistribStateManager stateManager;
+ protected SolrZkClient zkClient;
public ReplicaMutator(SolrCloudManager cloudManager) {
this.cloudManager = cloudManager;
this.stateManager = cloudManager.getDistribStateManager();
+ this.zkClient = getZkClient(cloudManager);
}
protected Replica setProperty(Replica replica, String key, String value) {
@@ -242,10 +248,12 @@ public class ReplicaMutator {
boolean forceSetState = message.getBool(ZkStateReader.FORCE_SET_STATE_PROP, true);
DocCollection collection = prevState.getCollectionOrNull(collectionName);
+
if (!forceSetState && !CloudUtil.replicaExists(prevState, collectionName, sliceName, coreNodeName)) {
log.info("Failed to update state because the replica does not exist, {}", message);
return ZkStateWriter.NO_OP;
}
+ boolean persistCollectionState = collection != null && collection.isPerReplicaState();
if (coreNodeName == null) {
coreNodeName = ClusterStateMutator.getAssignedCoreNodeName(collection,
@@ -257,6 +265,7 @@ public class ReplicaMutator {
log.info("Failed to update state because the replica does not exist, {}", message);
return ZkStateWriter.NO_OP;
}
+ persistCollectionState = true;
// if coreNodeName is null, auto assign one
coreNodeName = Assign.assignCoreNodeName(stateManager, collection);
}
@@ -271,6 +280,7 @@ public class ReplicaMutator {
if (sliceName != null) {
log.debug("shard={} is already registered", sliceName);
}
+ persistCollectionState = true;
}
if (sliceName == null) {
//request new shardId
@@ -281,13 +291,15 @@ public class ReplicaMutator {
}
sliceName = Assign.assignShard(collection, numShards);
log.info("Assigning new node to shard shard={}", sliceName);
+ persistCollectionState = true;
}
Slice slice = collection != null ? collection.getSlice(sliceName) : null;
Map<String, Object> replicaProps = new LinkedHashMap<>(message.getProperties());
+ Replica oldReplica = null;
if (slice != null) {
- Replica oldReplica = slice.getReplica(coreNodeName);
+ oldReplica = slice.getReplica(coreNodeName);
if (oldReplica != null) {
if (oldReplica.containsKey(ZkStateReader.LEADER_PROP)) {
replicaProps.put(ZkStateReader.LEADER_PROP, oldReplica.get(ZkStateReader.LEADER_PROP));
@@ -352,7 +364,12 @@ public class ReplicaMutator {
DocCollection newCollection = CollectionMutator.updateSlice(collectionName, collection, slice);
log.debug("Collection is now: {}", newCollection);
- return new ZkWriteCommand(collectionName, newCollection);
+ if(collection != null && collection.isPerReplicaState()) {
+ PerReplicaStates prs = PerReplicaStates.fetch(collection.getZNode(), zkClient, collection.getPerReplicaStates());
+ return new ZkWriteCommand(collectionName, newCollection, PerReplicaStates.WriteOps.flipState(replica.getName(), replica.getState(), prs), persistCollectionState);
+ } else{
+ return new ZkWriteCommand(collectionName, newCollection);
+ }
}
/**
diff --git a/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java b/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
index 40ab1a3..774aa5b 100644
--- a/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
+++ b/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
@@ -25,14 +25,17 @@ import java.util.Set;
import com.google.common.collect.ImmutableSet;
import org.apache.solr.client.solrj.cloud.DistribStateManager;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
+import org.apache.solr.client.solrj.impl.SolrClientCloudManager;
import org.apache.solr.cloud.Overseer;
import org.apache.solr.cloud.api.collections.Assign;
import org.apache.solr.cloud.api.collections.OverseerCollectionMessageHandler;
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.PerReplicaStates;
import org.apache.solr.common.cloud.RoutingRule;
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;
@@ -51,10 +54,21 @@ public class SliceMutator {
protected final SolrCloudManager cloudManager;
protected final DistribStateManager stateManager;
+ protected final SolrZkClient zkClient;
public SliceMutator(SolrCloudManager cloudManager) {
this.cloudManager = cloudManager;
this.stateManager = cloudManager.getDistribStateManager();
+ this.zkClient = getZkClient(cloudManager);
+ }
+
+ static SolrZkClient getZkClient(SolrCloudManager cloudManager) {
+ if (cloudManager instanceof SolrClientCloudManager) {
+ SolrClientCloudManager manager = (SolrClientCloudManager) cloudManager;
+ return manager.getZkClient();
+ } else {
+ return null;
+ }
}
public ZkWriteCommand addReplica(ClusterState clusterState, ZkNodeProps message) {
@@ -80,7 +94,15 @@ public class SliceMutator {
ZkStateReader.STATE_PROP, message.getStr(ZkStateReader.STATE_PROP),
ZkStateReader.NODE_NAME_PROP, message.getStr(ZkStateReader.NODE_NAME_PROP),
ZkStateReader.REPLICA_TYPE, message.get(ZkStateReader.REPLICA_TYPE)), coll, slice);
- return new ZkWriteCommand(coll, updateReplica(collection, sl, replica.getName(), replica));
+
+ if(collection.isPerReplicaState()) {
+ PerReplicaStates prs = PerReplicaStates.fetch(collection.getZNode(), zkClient, collection.getPerReplicaStates());
+ return new ZkWriteCommand(coll, updateReplica(collection, sl, replica.getName(), replica),
+ PerReplicaStates.WriteOps.addReplica(replica.getName(), replica.getState(), replica.isLeader(), prs), true);
+ } else {
+ return new ZkWriteCommand(coll, updateReplica(collection, sl, replica.getName(), replica));
+ }
+
}
public ZkWriteCommand removeReplica(ClusterState clusterState, ZkNodeProps message) {
@@ -106,7 +128,12 @@ public class SliceMutator {
newSlices.put(slice.getName(), slice);
}
- return new ZkWriteCommand(collection, coll.copyWithSlices(newSlices));
+
+ if(coll.isPerReplicaState()) {
+ return new ZkWriteCommand(collection, coll.copyWithSlices(newSlices), PerReplicaStates.WriteOps.deleteReplica(cnn, coll.getPerReplicaStates()) , true);
+ } else {
+ return new ZkWriteCommand(collection, coll.copyWithSlices(newSlices));
+ }
}
public ZkWriteCommand setShardLeader(ClusterState clusterState, ZkNodeProps message) {
@@ -124,6 +151,7 @@ public class SliceMutator {
Slice slice = slices.get(sliceName);
Replica oldLeader = slice.getLeader();
+ Replica newLeader = null;
final Map<String, Replica> newReplicas = new LinkedHashMap<>();
for (Replica replica : slice.getReplicas()) {
// TODO: this should only be calculated once and cached somewhere?
@@ -132,7 +160,7 @@ public class SliceMutator {
if (replica == oldLeader && !coreURL.equals(leaderUrl)) {
replica = new ReplicaMutator(cloudManager).unsetLeader(replica);
} else if (coreURL.equals(leaderUrl)) {
- replica = new ReplicaMutator(cloudManager).setLeader(replica);
+ newLeader= replica = new ReplicaMutator(cloudManager).setLeader(replica);
}
newReplicas.put(replica.getName(), replica);
@@ -141,8 +169,17 @@ public class SliceMutator {
Map<String, Object> newSliceProps = slice.shallowCopy();
newSliceProps.put(Slice.REPLICAS, newReplicas);
slice = new Slice(slice.getName(), newReplicas, slice.getProperties(), collectionName);
- return new ZkWriteCommand(collectionName, CollectionMutator.updateSlice(collectionName, coll, slice));
- }
+ if(coll.isPerReplicaState()) {
+ PerReplicaStates prs = PerReplicaStates.fetch(coll.getZNode(), zkClient, coll.getPerReplicaStates());
+ return new ZkWriteCommand(collectionName, CollectionMutator.updateSlice(collectionName, coll, slice),
+ PerReplicaStates.WriteOps.flipLeader(
+ slice.getReplicaNames(),
+ newLeader == null ? null : newLeader.getName(),
+ prs), false);
+ } else {
+ return new ZkWriteCommand(collectionName, CollectionMutator.updateSlice(collectionName, coll, slice));
+ }
+ }
public ZkWriteCommand updateShardState(ClusterState clusterState, ZkNodeProps message) {
String collectionName = message.getStr(ZkStateReader.COLLECTION_PROP);
diff --git a/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java b/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java
index cb89371..5979ddb 100644
--- a/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java
+++ b/solr/core/src/java/org/apache/solr/cloud/overseer/ZkStateWriter.java
@@ -27,6 +27,7 @@ import org.apache.solr.cloud.Overseer;
import org.apache.solr.cloud.Stats;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.PerReplicaStates;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.util.Utils;
import org.apache.zookeeper.CreateMode;
@@ -64,7 +65,7 @@ public class ZkStateWriter {
protected final ZkStateReader reader;
protected final Stats stats;
- protected Map<String, DocCollection> updates = new HashMap<>();
+ protected Map<String, ZkWriteCommand> updates = new HashMap<>();
private int numUpdates = 0;
protected ClusterState clusterState = null;
protected boolean isClusterStateModified = false;
@@ -113,6 +114,38 @@ public class ZkStateWriter {
if (cmds.isEmpty()) return prevState;
if (isNoOps(cmds)) return prevState;
+ boolean forceFlush = false;
+ if(cmds.size() == 1) {
+ //most messages result in only one command. let's deal with it right away
+ ZkWriteCommand cmd = cmds.get(0);
+ if (cmd.collection != null && cmd.collection.isPerReplicaState()) {
+ //we do not wish to batch any updates for collections with per-replica state because
+ // these changes go to individual ZK nodes and there is zero advantage to batching
+ //now check if there are any updates for the same collection already present
+ if (updates.containsKey(cmd.name)) {
+ //this should not happen
+ // but let's get those updates out anyway
+ writeUpdate(updates.remove(cmd.name));
+ }
+ //now let's write the current message
+ try {
+ return writeUpdate(cmd);
+ } finally {
+ if(callback !=null) callback.onWrite();
+ }
+ }
+ } else {
+ //there are more than one commands created as a result of this message
+ for (ZkWriteCommand cmd : cmds) {
+ if(cmd.collection != null && cmd.collection.isPerReplicaState()) {
+ // we don't try to optimize for this case. let's flush out all after this
+ forceFlush = true;
+ break;
+ }
+ }
+ }
+
+
for (ZkWriteCommand cmd : cmds) {
if (cmd == NO_OP) continue;
if (!isClusterStateModified && clusterStateGetModifiedWith(cmd, prevState)) {
@@ -120,13 +153,13 @@ public class ZkStateWriter {
}
prevState = prevState.copyWith(cmd.name, cmd.collection);
if (cmd.collection == null || cmd.collection.getStateFormat() != 1) {
- updates.put(cmd.name, cmd.collection);
+ updates.put(cmd.name, cmd);
numUpdates++;
}
}
clusterState = prevState;
- if (maybeFlushAfter()) {
+ if (forceFlush || maybeFlushAfter()) {
ClusterState state = writePendingUpdates();
if (callback != null) {
callback.onWrite();
@@ -165,7 +198,15 @@ public class ZkStateWriter {
public boolean hasPendingUpdates() {
return numUpdates != 0 || isClusterStateModified;
}
+ public ClusterState writeUpdate(ZkWriteCommand command) throws IllegalStateException, KeeperException, InterruptedException {
+ Map<String, ZkWriteCommand> commands = new HashMap<>();
+ commands.put(command.name, command);
+ return writePendingUpdates(commands);
+ }
+ public ClusterState writePendingUpdates() throws KeeperException, InterruptedException {
+ return writePendingUpdates(updates);
+ }
/**
* Writes all pending updates to ZooKeeper and returns the modified cluster state
*
@@ -174,20 +215,36 @@ public class ZkStateWriter {
* @throws KeeperException if any ZooKeeper operation results in an error
* @throws InterruptedException if the current thread is interrupted
*/
- public ClusterState writePendingUpdates() throws IllegalStateException, KeeperException, InterruptedException {
+ public ClusterState writePendingUpdates(Map<String, ZkWriteCommand> updates) throws IllegalStateException, KeeperException, InterruptedException {
if (invalidState) {
throw new IllegalStateException("ZkStateWriter has seen a tragic error, this instance can no longer be used");
}
- if (!hasPendingUpdates()) return clusterState;
+ if ((updates == this.updates)
+ && !hasPendingUpdates()) {
+ return clusterState;
+ }
Timer.Context timerContext = stats.time("update_state");
boolean success = false;
try {
if (!updates.isEmpty()) {
- for (Map.Entry<String, DocCollection> entry : updates.entrySet()) {
+ for (Map.Entry<String, ZkWriteCommand> entry : updates.entrySet()) {
String name = entry.getKey();
String path = ZkStateReader.getCollectionPath(name);
- DocCollection c = entry.getValue();
+ ZkWriteCommand cmd = entry.getValue();
+ DocCollection c = cmd.collection;
+
+ if(cmd.ops != null && cmd.ops.isPreOp()) {
+ PerReplicaStates.persist(cmd.ops, path, reader.getZkClient());
+ //nocommit
+ /*PerReplicaStates prs = PerReplicaStates.fetch(ZkStateReader.getCollectionPath(cmd.collection.getName()), reader.getZkClient());
+ log.debug("node {} ,per-replica states persisted {}->{}",reader.nodeName,
+ cmd.ops.getPerReplicaStates(),
+ prs);*/
+ clusterState = clusterState.copyWith(name,
+ cmd.collection.copyWith(PerReplicaStates.fetch(cmd.collection.getZNode(), reader.getZkClient(), null)));
+ }
+ if(!cmd.persistCollState) continue;
if (c == null) {
// let's clean up the state.json of this collection only, the rest should be clean by delete collection cmd
log.debug("going to delete state.json {}", path);
@@ -210,6 +267,14 @@ public class ZkStateWriter {
} else if (c.getStateFormat() == 1) {
isClusterStateModified = true;
}
+ if(cmd.ops != null && !cmd.ops.isPreOp()) {
+ PerReplicaStates.persist(cmd.ops, path, reader.getZkClient());
+ DocCollection currentCollState = clusterState.getCollection(cmd.name);
+ if ( currentCollState != null) {
+ clusterState = clusterState.copyWith(name,
+ currentCollState.copyWith(PerReplicaStates.fetch(currentCollState.getZNode(), reader.getZkClient(), null)));
+ }
+ }
}
updates.clear();
diff --git a/solr/core/src/java/org/apache/solr/cloud/overseer/ZkWriteCommand.java b/solr/core/src/java/org/apache/solr/cloud/overseer/ZkWriteCommand.java
index d464863..2f71674 100644
--- a/solr/core/src/java/org/apache/solr/cloud/overseer/ZkWriteCommand.java
+++ b/solr/core/src/java/org/apache/solr/cloud/overseer/ZkWriteCommand.java
@@ -17,16 +17,34 @@
package org.apache.solr.cloud.overseer;
import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.PerReplicaStates;
public class ZkWriteCommand {
+
public final String name;
public final DocCollection collection;
+
public final boolean noop;
+ // persist the collection state. If this is false, it means the collection state is not modified
+ public final boolean persistCollState;
+ public final PerReplicaStates.WriteOps ops;
+ public ZkWriteCommand(String name, DocCollection collection, PerReplicaStates.WriteOps replicaOps, boolean persistCollState) {
+ boolean isPerReplicaState = collection.isPerReplicaState();
+ this.name = name;
+ this.collection = collection;
+ this.noop = false;
+ this.ops = isPerReplicaState ? replicaOps : null;
+ this.persistCollState = isPerReplicaState ? persistCollState : true;
+ }
public ZkWriteCommand(String name, DocCollection collection) {
this.name = name;
this.collection = collection;
this.noop = false;
+ persistCollState = true;
+ this.ops = collection != null && collection.isPerReplicaState() ?
+ PerReplicaStates.WriteOps.touchChildren():
+ null;
}
/**
@@ -36,6 +54,8 @@ public class ZkWriteCommand {
this.noop = true;
this.name = null;
this.collection = null;
+ this.ops = null;
+ persistCollState = true;
}
public static ZkWriteCommand noop() {
diff --git a/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java b/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
index 0832126..917d386 100644
--- a/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
+++ b/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
@@ -123,6 +123,7 @@ import static org.apache.solr.cloud.api.collections.OverseerCollectionMessageHan
import static org.apache.solr.cloud.api.collections.RoutedAlias.CREATE_COLLECTION_PREFIX;
import static org.apache.solr.common.SolrException.ErrorCode.BAD_REQUEST;
import static org.apache.solr.common.cloud.DocCollection.DOC_ROUTER;
+import static org.apache.solr.common.cloud.DocCollection.PER_REPLICA_STATE;
import static org.apache.solr.common.cloud.DocCollection.RULE;
import static org.apache.solr.common.cloud.DocCollection.SNITCH;
import static org.apache.solr.common.cloud.DocCollection.STATE_FORMAT;
@@ -495,6 +496,7 @@ public class CollectionsHandler extends RequestHandlerBase implements Permission
POLICY,
WAIT_FOR_FINAL_STATE,
WITH_COLLECTION,
+ PER_REPLICA_STATE,
ALIAS);
props.putIfAbsent(STATE_FORMAT, "2");
@@ -1453,6 +1455,9 @@ public class CollectionsHandler extends RequestHandlerBase implements Permission
}
if (!n.contains(replica.getNodeName())
|| !state.equals(Replica.State.ACTIVE.toString())) {
+ if(log.isDebugEnabled()) {
+ log.debug("inactive replica {} , state {}", replica.getName(), replica.getReplicaState());
+ }
replicaNotAliveCnt++;
return false;
}
@@ -1464,7 +1469,7 @@ public class CollectionsHandler extends RequestHandlerBase implements Permission
return false;
});
} catch (TimeoutException | InterruptedException e) {
-
+ cc.getZkController().getZkStateReader().debugCollectionState(collectionName);
String error = "Timeout waiting for active collection " + collectionName + " with timeout=" + seconds;
throw new NotInClusterStateException(ErrorCode.SERVER_ERROR, error);
}
diff --git a/solr/core/src/test/org/apache/solr/cloud/CollectionsAPISolrJTest.java b/solr/core/src/test/org/apache/solr/cloud/CollectionsAPISolrJTest.java
index e1f502e..f2b6e59 100644
--- a/solr/core/src/test/org/apache/solr/cloud/CollectionsAPISolrJTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/CollectionsAPISolrJTest.java
@@ -112,6 +112,7 @@ public class CollectionsAPISolrJTest extends SolrCloudTestCase {
public void testCreateWithDefaultConfigSet() throws Exception {
String collectionName = "solrj_default_configset";
CollectionAdminResponse response = CollectionAdminRequest.createCollection(collectionName, 2, 2)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.process(cluster.getSolrClient());
cluster.waitForActiveCollection(collectionName, 2, 4);
@@ -244,6 +245,7 @@ public class CollectionsAPISolrJTest extends SolrCloudTestCase {
assertEquals("2", String.valueOf(clusterProperty));
CollectionAdminResponse response = CollectionAdminRequest
.createCollection(COLL_NAME, "conf", null, null, null, null)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.process(cluster.getSolrClient());
assertEquals(0, response.getStatus());
assertTrue(response.isSuccess());
@@ -419,7 +421,9 @@ public class CollectionsAPISolrJTest extends SolrCloudTestCase {
public void testCreateAndDeleteAlias() throws IOException, SolrServerException {
final String collection = "aliasedCollection";
- CollectionAdminRequest.createCollection(collection, "conf", 1, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(collection, "conf", 1, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
CollectionAdminResponse response
= CollectionAdminRequest.createAlias("solrj_alias", collection).process(cluster.getSolrClient());
@@ -434,6 +438,7 @@ public class CollectionsAPISolrJTest extends SolrCloudTestCase {
final String collectionName = "solrj_test_splitshard";
CollectionAdminRequest.createCollection(collectionName, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.process(cluster.getSolrClient());
cluster.waitForActiveCollection(collectionName, 2, 2);
@@ -488,6 +493,7 @@ public class CollectionsAPISolrJTest extends SolrCloudTestCase {
cluster.getJettySolrRunners().forEach(j -> j.getCoreContainer().getAllowPaths().add(tmpDir));
CollectionAdminResponse response = CollectionAdminRequest.createCollection(collectionName, "conf", 1, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.withProperty(CoreAdminParams.DATA_DIR, dataDir.toString())
.withProperty(CoreAdminParams.ULOG_DIR, ulogDir.toString())
.process(cluster.getSolrClient());
@@ -514,6 +520,7 @@ public class CollectionsAPISolrJTest extends SolrCloudTestCase {
final String collectionName = "solrj_replicatests";
CollectionAdminRequest.createCollection(collectionName, "conf", 1, 2)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.process(cluster.getSolrClient());
cluster.waitForActiveCollection(collectionName, 1, 2);
@@ -584,6 +591,7 @@ public class CollectionsAPISolrJTest extends SolrCloudTestCase {
final String propName = "testProperty";
CollectionAdminRequest.createCollection(collectionName, "conf", 2, 2)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.process(cluster.getSolrClient());
cluster.waitForActiveCollection(collectionName, 2, 4);
@@ -615,6 +623,7 @@ public class CollectionsAPISolrJTest extends SolrCloudTestCase {
public void testColStatus() throws Exception {
final String collectionName = "collectionStatusTest";
CollectionAdminRequest.createCollection(collectionName, "conf2", 2, 2)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.process(cluster.getSolrClient());
cluster.waitForActiveCollection(collectionName, 2, 4);
diff --git a/solr/core/src/test/org/apache/solr/cloud/SplitShardTest.java b/solr/core/src/test/org/apache/solr/cloud/SplitShardTest.java
index 0e36b57..365c396 100644
--- a/solr/core/src/test/org/apache/solr/cloud/SplitShardTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/SplitShardTest.java
@@ -77,6 +77,7 @@ public class SplitShardTest extends SolrCloudTestCase {
public void doTest() throws IOException, SolrServerException {
CollectionAdminRequest
.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(100)
.process(cluster.getSolrClient());
@@ -128,6 +129,7 @@ public class SplitShardTest extends SolrCloudTestCase {
String collectionName = "splitFuzzCollection";
CollectionAdminRequest
.createCollection(collectionName, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(100)
.process(cluster.getSolrClient());
@@ -157,6 +159,7 @@ public class SplitShardTest extends SolrCloudTestCase {
CollectionAdminRequest
.createCollection(collectionName, "conf", 1, repFactor)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(100)
.process(cluster.getSolrClient());
diff --git a/solr/core/src/test/org/apache/solr/cloud/api/collections/ShardSplitTest.java b/solr/core/src/test/org/apache/solr/cloud/api/collections/ShardSplitTest.java
index e3b2634..87e2ef4 100644
--- a/solr/core/src/test/org/apache/solr/cloud/api/collections/ShardSplitTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/api/collections/ShardSplitTest.java
@@ -77,7 +77,7 @@ import static org.apache.solr.common.cloud.ZkStateReader.MAX_SHARDS_PER_NODE;
import static org.apache.solr.common.cloud.ZkStateReader.REPLICATION_FACTOR;
@Slow
-@LogLevel("org.apache.solr.cloud.Overseer=DEBUG;org.apache.solr.cloud.overseer=DEBUG;org.apache.solr.cloud.api.collections=DEBUG;org.apache.solr.cloud.OverseerTaskProcessor=DEBUG;org.apache.solr.util.TestInjection=DEBUG")
+@LogLevel("org.apache.solr.common.cloud.PerReplicaStates=DEBUG;org.apache.solr.common.cloud=DEBUG;org.apache.solr.cloud.Overseer=DEBUG;org.apache.solr.cloud.overseer=DEBUG;org.apache.solr.cloud.api.collections=DEBUG;org.apache.solr.cloud.OverseerTaskProcessor=DEBUG;org.apache.solr.util.TestInjection=DEBUG")
public class ShardSplitTest extends BasicDistributedZkTest {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -140,6 +140,7 @@ public class ShardSplitTest extends BasicDistributedZkTest {
String collectionName = "testSplitStaticIndexReplication_" + splitMethod.toLower();
CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection(collectionName, "conf1", 1, 1);
+ create.setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE);
create.setMaxShardsPerNode(5); // some high number so we can create replicas without hindrance
create.setCreateNodeSet(nodeName); // we want to create the leader on a fixed node so that we know which one to restart later
create.process(cloudClient);
@@ -359,6 +360,7 @@ public class ShardSplitTest extends BasicDistributedZkTest {
waitForThingsToLevelOut(15);
String collectionName = "testSplitMixedReplicaTypes_" + splitMethod.toLower();
CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection(collectionName, "conf1", 1, 2, 0, 2); // TODO tlog replicas disabled right now.
+ create.setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE);
create.setMaxShardsPerNode(5); // some high number so we can create replicas without hindrance
create.process(cloudClient);
@@ -652,6 +654,7 @@ public class ShardSplitTest extends BasicDistributedZkTest {
log.info("Starting testSplitShardWithRule");
String collectionName = "shardSplitWithRule_" + splitMethod.toLower();
CollectionAdminRequest.Create createRequest = CollectionAdminRequest.createCollection(collectionName, "conf1", 1, 2)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setRule("shard:*,replica:<2,node:*");
CollectionAdminResponse response = createRequest.process(cloudClient);
@@ -820,6 +823,9 @@ public class ShardSplitTest extends BasicDistributedZkTest {
MAX_SHARDS_PER_NODE, maxShardsPerNode,
OverseerCollectionMessageHandler.NUM_SLICES, numShards,
"router.field", shard_fld);
+ if(SolrCloudTestCase.USE_PER_REPLICA_STATE) {
+ props.put(DocCollection.PER_REPLICA_STATE, Boolean.TRUE);
+ }
createCollection(collectionInfos, collectionName, props, client);
}
@@ -881,7 +887,9 @@ public class ShardSplitTest extends BasicDistributedZkTest {
REPLICATION_FACTOR, replicationFactor,
MAX_SHARDS_PER_NODE, maxShardsPerNode,
OverseerCollectionMessageHandler.NUM_SLICES, numShards);
-
+ if(SolrCloudTestCase.USE_PER_REPLICA_STATE) {
+ props.put(DocCollection.PER_REPLICA_STATE, Boolean.TRUE);
+ }
createCollection(collectionInfos, collectionName,props,client);
}
diff --git a/solr/core/src/test/org/apache/solr/handler/PingRequestHandlerTest.java b/solr/core/src/test/org/apache/solr/handler/PingRequestHandlerTest.java
index dd8aead..9d196cf 100644
--- a/solr/core/src/test/org/apache/solr/handler/PingRequestHandlerTest.java
+++ b/solr/core/src/test/org/apache/solr/handler/PingRequestHandlerTest.java
@@ -28,6 +28,7 @@ import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.client.solrj.request.SolrPing;
import org.apache.solr.client.solrj.response.SolrPingResponse;
import org.apache.solr.cloud.MiniSolrCloudCluster;
+import org.apache.solr.cloud.SolrCloudTestCase;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.request.SolrQueryRequest;
@@ -188,6 +189,7 @@ public class PingRequestHandlerTest extends SolrTestCaseJ4 {
String configName = "solrCloudCollectionConfig";
miniCluster.uploadConfigSet(SolrTestCaseJ4.TEST_PATH().resolve("collection1").resolve("conf"), configName);
CollectionAdminRequest.createCollection(collectionName, configName, NUM_SHARDS, REPLICATION_FACTOR)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.process(miniCluster.getSolrClient());
// Send distributed and non-distributed ping query
diff --git a/solr/core/src/test/org/apache/solr/handler/TestContainerPlugin.java b/solr/core/src/test/org/apache/solr/handler/TestContainerPlugin.java
index 5ce53d6..3d952f3 100644
--- a/solr/core/src/test/org/apache/solr/handler/TestContainerPlugin.java
+++ b/solr/core/src/test/org/apache/solr/handler/TestContainerPlugin.java
@@ -21,6 +21,8 @@ import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
//import java.util.Collections;
+import java.util.Arrays;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
@@ -41,6 +43,7 @@ import org.apache.solr.client.solrj.response.V2Response;
import org.apache.solr.cloud.MiniSolrCloudCluster;
import org.apache.solr.cloud.SolrCloudTestCase;
import org.apache.solr.common.NavigableObject;
+import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.util.Utils;
import org.apache.solr.core.SolrResourceLoader;
import org.apache.solr.filestore.PackageStoreAPI;
@@ -50,6 +53,9 @@ import org.apache.solr.pkg.TestPackages;
import org.apache.solr.request.SolrQueryRequest;
import org.apache.solr.response.SolrQueryResponse;
import org.apache.solr.security.PermissionNameProvider;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.Op;
+import org.apache.zookeeper.data.Stat;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -73,6 +79,29 @@ public class TestContainerPlugin extends SolrCloudTestCase {
System.clearProperty("enable.packages");
}
+ public void testZkBehavior() throws Exception {
+ MiniSolrCloudCluster cluster =
+ configureCluster(4)
+ .withJettyConfig(jetty -> jetty.enableV2(true))
+ .configure();
+ try{
+ SolrZkClient zkClient = cluster.getZkClient();
+ zkClient.create("/test-node", null, CreateMode.PERSISTENT, true);
+
+ Stat stat = zkClient.exists("/test-node", null, true);
+ System.out.println(stat.getCversion());
+ List<Op> ops = Arrays.asList(
+ Op.create("/test-node/abc", null, zkClient.getZkACLProvider().getACLsToAdd("/test-node/abc"), CreateMode.PERSISTENT),
+ Op.delete("/test-node/abc", -1));
+ zkClient.multi(ops, true);
+ stat = zkClient.exists("/test-node", null, true);
+ System.out.println(stat.getCversion());
+ } finally {
+ cluster.shutdown();
+ }
+
+ }
+
@Test
public void testApi() throws Exception {
MiniSolrCloudCluster cluster =
diff --git a/solr/core/src/test/org/apache/solr/handler/TestSQLHandler.java b/solr/core/src/test/org/apache/solr/handler/TestSQLHandler.java
index 96555ca..5bc68532 100644
--- a/solr/core/src/test/org/apache/solr/handler/TestSQLHandler.java
+++ b/solr/core/src/test/org/apache/solr/handler/TestSQLHandler.java
@@ -63,7 +63,9 @@ public class TestSQLHandler extends SolrCloudTestCase {
collection = COLLECTIONORALIAS;
}
- CollectionAdminRequest.createCollection(collection, "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(collection, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection(collection, 2, 2);
if (useAlias) {
CollectionAdminRequest.createAlias(COLLECTIONORALIAS, collection).process(cluster.getSolrClient());
diff --git a/solr/core/src/test/org/apache/solr/handler/TestStressThreadBackup.java b/solr/core/src/test/org/apache/solr/handler/TestStressThreadBackup.java
index 3622948..5ef2793 100644
--- a/solr/core/src/test/org/apache/solr/handler/TestStressThreadBackup.java
+++ b/solr/core/src/test/org/apache/solr/handler/TestStressThreadBackup.java
@@ -61,7 +61,7 @@ import org.junit.BeforeClass;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-@Nightly
+//@Nightly
@SuppressCodecs({"SimpleText"})
@LogLevel("org.apache.solr.handler.SnapShooter=DEBUG;org.apache.solr.core.IndexDeletionPolicyWrapper=DEBUG")
public class TestStressThreadBackup extends SolrCloudTestCase {
@@ -96,7 +96,8 @@ public class TestStressThreadBackup extends SolrCloudTestCase {
.configure();
assertEquals(0, (CollectionAdminRequest.createCollection(DEFAULT_TEST_COLLECTION_NAME, "conf1", 1, 1)
- .process(cluster.getSolrClient()).getStatus()));
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient()).getStatus()));
adminClient = getHttpSolrClient(cluster.getJettySolrRunners().get(0).getBaseUrl().toString());
initCoreNameAndSolrCoreClient();
}
diff --git a/solr/core/src/test/org/apache/solr/handler/admin/HealthCheckHandlerTest.java b/solr/core/src/test/org/apache/solr/handler/admin/HealthCheckHandlerTest.java
index 7a4dcb1..e30d7e0 100644
--- a/solr/core/src/test/org/apache/solr/handler/admin/HealthCheckHandlerTest.java
+++ b/solr/core/src/test/org/apache/solr/handler/admin/HealthCheckHandlerTest.java
@@ -79,6 +79,7 @@ public class HealthCheckHandlerTest extends SolrCloudTestCase {
try (HttpSolrClient httpSolrClient = getHttpSolrClient(cluster.getJettySolrRunner(0).getBaseUrl().toString())) {
CollectionAdminResponse collectionAdminResponse = CollectionAdminRequest.createCollection("test", "_default", 1, 1)
.withProperty("solr.directoryFactory", "solr.StandardDirectoryFactory")
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.process(httpSolrClient);
assertEquals(0, collectionAdminResponse.getStatus());
SolrResponse response = req.process(httpSolrClient);
diff --git a/solr/core/src/test/org/apache/solr/handler/admin/IndexSizeEstimatorTest.java b/solr/core/src/test/org/apache/solr/handler/admin/IndexSizeEstimatorTest.java
index b092be1..5dfb925 100644
--- a/solr/core/src/test/org/apache/solr/handler/admin/IndexSizeEstimatorTest.java
+++ b/solr/core/src/test/org/apache/solr/handler/admin/IndexSizeEstimatorTest.java
@@ -75,6 +75,7 @@ public class IndexSizeEstimatorTest extends SolrCloudTestCase {
.configure();
solrClient = cluster.getSolrClient();
CollectionAdminRequest.createCollection(collection, "conf", 2, 2)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(2).process(solrClient);
cluster.waitForActiveCollection(collection, 2, 4);
SolrInputDocument lastDoc = addDocs(collection, NUM_DOCS);
diff --git a/solr/core/src/test/org/apache/solr/handler/admin/MetricsHistoryHandlerTest.java b/solr/core/src/test/org/apache/solr/handler/admin/MetricsHistoryHandlerTest.java
index 27987bd..3c184e5 100644
--- a/solr/core/src/test/org/apache/solr/handler/admin/MetricsHistoryHandlerTest.java
+++ b/solr/core/src/test/org/apache/solr/handler/admin/MetricsHistoryHandlerTest.java
@@ -95,7 +95,8 @@ public class MetricsHistoryHandlerTest extends SolrCloudTestCase {
// create .system collection
CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection(CollectionAdminParams.SYSTEM_COLL,
- "conf", 1, 1);
+ "conf", 1, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE);
create.process(solrClient);
CloudUtil.waitForState(cloudManager, "failed to create " + CollectionAdminParams.SYSTEM_COLL,
CollectionAdminParams.SYSTEM_COLL, CloudUtil.clusterShape(1, 1));
diff --git a/solr/core/src/test/org/apache/solr/handler/component/CustomHighlightComponentTest.java b/solr/core/src/test/org/apache/solr/handler/component/CustomHighlightComponentTest.java
index 84dd45a..d026ff4 100644
--- a/solr/core/src/test/org/apache/solr/handler/component/CustomHighlightComponentTest.java
+++ b/solr/core/src/test/org/apache/solr/handler/component/CustomHighlightComponentTest.java
@@ -128,7 +128,8 @@ public class CustomHighlightComponentTest extends SolrCloudTestCase {
// create an empty collection
CollectionAdminRequest
.createCollection(COLLECTION, "conf", numShards, numReplicas)
- .setMaxShardsPerNode(maxShardsPerNode)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .setMaxShardsPerNode(maxShardsPerNode)
.processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
AbstractDistribZkTestBase.waitForRecoveriesToFinish(COLLECTION, cluster.getSolrClient().getZkStateReader(), false, true, DEFAULT_TIMEOUT);
}
diff --git a/solr/core/src/test/org/apache/solr/handler/component/DistributedQueryComponentOptimizationTest.java b/solr/core/src/test/org/apache/solr/handler/component/DistributedQueryComponentOptimizationTest.java
index 8572ae4..865f9be 100644
--- a/solr/core/src/test/org/apache/solr/handler/component/DistributedQueryComponentOptimizationTest.java
+++ b/solr/core/src/test/org/apache/solr/handler/component/DistributedQueryComponentOptimizationTest.java
@@ -62,6 +62,7 @@ public class DistributedQueryComponentOptimizationTest extends SolrCloudTestCase
.configure();
CollectionAdminRequest.createCollection(COLLECTION, "conf", 3, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(1)
.processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
cluster.getSolrClient().waitForState(COLLECTION, DEFAULT_TIMEOUT, TimeUnit.SECONDS,
diff --git a/solr/core/src/test/org/apache/solr/handler/component/SearchHandlerTest.java b/solr/core/src/test/org/apache/solr/handler/component/SearchHandlerTest.java
index f0b2973..aa17b55 100644
--- a/solr/core/src/test/org/apache/solr/handler/component/SearchHandlerTest.java
+++ b/solr/core/src/test/org/apache/solr/handler/component/SearchHandlerTest.java
@@ -30,6 +30,7 @@ import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.client.solrj.request.QueryRequest;
import org.apache.solr.client.solrj.response.QueryResponse;
import org.apache.solr.cloud.MiniSolrCloudCluster;
+import org.apache.solr.cloud.SolrCloudTestCase;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.params.ModifiableSolrParams;
@@ -140,6 +141,7 @@ public class SearchHandlerTest extends SolrTestCaseJ4
miniCluster.uploadConfigSet(SolrTestCaseJ4.TEST_PATH().resolve("collection1/conf"), configName);
CollectionAdminRequest.createCollection(collectionName, configName, 2, 2)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.process(miniCluster.getSolrClient());
QueryRequest req = new QueryRequest();
@@ -182,6 +184,7 @@ public class SearchHandlerTest extends SolrTestCaseJ4
miniCluster.uploadConfigSet(SolrTestCaseJ4.TEST_PATH().resolve("collection1/conf"), configName);
CollectionAdminRequest.createCollection(collectionName, configName, 2, 2)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.process(miniCluster.getSolrClient());
ModifiableSolrParams params = new ModifiableSolrParams();
@@ -229,6 +232,7 @@ public class SearchHandlerTest extends SolrTestCaseJ4
miniCluster.uploadConfigSet(SolrTestCaseJ4.TEST_PATH().resolve("collection1/conf"), configName);
CollectionAdminRequest.createCollection(collectionName, configName, 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.process(miniCluster.getSolrClient());
ModifiableSolrParams params = new ModifiableSolrParams();
diff --git a/solr/core/src/test/org/apache/solr/metrics/reporters/SolrJmxReporterCloudTest.java b/solr/core/src/test/org/apache/solr/metrics/reporters/SolrJmxReporterCloudTest.java
index 42e7159..43949ad 100644
--- a/solr/core/src/test/org/apache/solr/metrics/reporters/SolrJmxReporterCloudTest.java
+++ b/solr/core/src/test/org/apache/solr/metrics/reporters/SolrJmxReporterCloudTest.java
@@ -60,6 +60,7 @@ public class SolrJmxReporterCloudTest extends SolrCloudTestCase {
.configure();
CollectionAdminRequest.createCollection(COLLECTION, "conf", 2, 1)
.setMaxShardsPerNode(2)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.process(cluster.getSolrClient());
}
@AfterClass
diff --git a/solr/core/src/test/org/apache/solr/metrics/reporters/solr/SolrCloudReportersTest.java b/solr/core/src/test/org/apache/solr/metrics/reporters/solr/SolrCloudReportersTest.java
index cbd6092..181c38c 100644
--- a/solr/core/src/test/org/apache/solr/metrics/reporters/solr/SolrCloudReportersTest.java
+++ b/solr/core/src/test/org/apache/solr/metrics/reporters/solr/SolrCloudReportersTest.java
@@ -68,6 +68,7 @@ public class SolrCloudReportersTest extends SolrCloudTestCase {
cluster.uploadConfigSet(Paths.get(TEST_PATH().toString(), "configsets", "minimal", "conf"), "test");
CollectionAdminRequest.createCollection("test_collection", "test", 2, 2)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(4)
.process(cluster.getSolrClient());
cluster.waitForActiveCollection("test_collection", 2, 4);
@@ -170,6 +171,7 @@ public class SolrCloudReportersTest extends SolrCloudTestCase {
cluster.uploadConfigSet(Paths.get(TEST_PATH().toString(), "configsets", "minimal", "conf"), "test");
CollectionAdminRequest.createCollection("test_collection", "test", 2, 2)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(4)
.process(cluster.getSolrClient());
cluster.waitForActiveCollection("test_collection", 2, 4);
diff --git a/solr/core/src/test/org/apache/solr/pkg/TestPackages.java b/solr/core/src/test/org/apache/solr/pkg/TestPackages.java
index 87134a3..9b30753 100644
--- a/solr/core/src/test/org/apache/solr/pkg/TestPackages.java
+++ b/solr/core/src/test/org/apache/solr/pkg/TestPackages.java
@@ -144,6 +144,7 @@ public class TestPackages extends SolrCloudTestCase {
CollectionAdminRequest
.createCollection(COLLECTION_NAME, "conf", 2, 2)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(100)
.process(cluster.getSolrClient());
cluster.waitForActiveCollection(COLLECTION_NAME, 2, 4);
diff --git a/solr/core/src/test/org/apache/solr/response/transform/TestSubQueryTransformerDistrib.java b/solr/core/src/test/org/apache/solr/response/transform/TestSubQueryTransformerDistrib.java
index 0f9221c..524dff9 100644
--- a/solr/core/src/test/org/apache/solr/response/transform/TestSubQueryTransformerDistrib.java
+++ b/solr/core/src/test/org/apache/solr/response/transform/TestSubQueryTransformerDistrib.java
@@ -78,10 +78,12 @@ public class TestSubQueryTransformerDistrib extends SolrCloudTestCase {
int replicas = 2 ;
CollectionAdminRequest.createCollection(people, configName, shards, replicas)
.withProperty("config", "solrconfig-doctransformers.xml")
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.withProperty("schema", "schema-docValuesJoin.xml")
.process(cluster.getSolrClient());
CollectionAdminRequest.createCollection(depts, configName, shards, replicas)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.withProperty("config", "solrconfig-doctransformers.xml")
.withProperty("schema",
differentUniqueId ? "schema-minimal-with-another-uniqkey.xml":
diff --git a/solr/core/src/test/org/apache/solr/schema/ManagedSchemaRoundRobinCloudTest.java b/solr/core/src/test/org/apache/solr/schema/ManagedSchemaRoundRobinCloudTest.java
index 883ebfd..44995bd 100644
--- a/solr/core/src/test/org/apache/solr/schema/ManagedSchemaRoundRobinCloudTest.java
+++ b/solr/core/src/test/org/apache/solr/schema/ManagedSchemaRoundRobinCloudTest.java
@@ -46,6 +46,7 @@ public class ManagedSchemaRoundRobinCloudTest extends SolrCloudTestCase {
System.setProperty("managed.schema.mutable", "true");
configureCluster(NUM_SHARDS).addConfig(CONFIG, configset(CONFIG)).configure();
CollectionAdminRequest.createCollection(COLLECTION, CONFIG, NUM_SHARDS, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(1)
.process(cluster.getSolrClient());
cluster.getSolrClient().waitForState(COLLECTION, DEFAULT_TIMEOUT, TimeUnit.SECONDS,
diff --git a/solr/core/src/test/org/apache/solr/schema/PreAnalyzedFieldManagedSchemaCloudTest.java b/solr/core/src/test/org/apache/solr/schema/PreAnalyzedFieldManagedSchemaCloudTest.java
index 04e1be0..d56939d 100644
--- a/solr/core/src/test/org/apache/solr/schema/PreAnalyzedFieldManagedSchemaCloudTest.java
+++ b/solr/core/src/test/org/apache/solr/schema/PreAnalyzedFieldManagedSchemaCloudTest.java
@@ -40,6 +40,7 @@ public class PreAnalyzedFieldManagedSchemaCloudTest extends SolrCloudTestCase {
public static void setupCluster() throws Exception {
configureCluster(2).addConfig(CONFIG, configset(CONFIG)).configure();
CollectionAdminRequest.createCollection(COLLECTION, CONFIG, 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(1)
.process(cluster.getSolrClient());
cluster.getSolrClient().waitForState(COLLECTION, DEFAULT_TIMEOUT, TimeUnit.SECONDS,
diff --git a/solr/core/src/test/org/apache/solr/schema/TestManagedSchemaAPI.java b/solr/core/src/test/org/apache/solr/schema/TestManagedSchemaAPI.java
index a2375ba..f022ac3 100644
--- a/solr/core/src/test/org/apache/solr/schema/TestManagedSchemaAPI.java
+++ b/solr/core/src/test/org/apache/solr/schema/TestManagedSchemaAPI.java
@@ -51,6 +51,8 @@ public class TestManagedSchemaAPI extends SolrCloudTestCase {
public void test() throws Exception {
String collection = "testschemaapi";
CollectionAdminRequest.createCollection(collection, "conf1", 1, 2)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+
.process(cluster.getSolrClient());
testModifyField(collection);
testReloadAndAddSimple(collection);
diff --git a/solr/core/src/test/org/apache/solr/search/CurrencyRangeFacetCloudTest.java b/solr/core/src/test/org/apache/solr/search/CurrencyRangeFacetCloudTest.java
index 61ad8c2..86b3175 100644
--- a/solr/core/src/test/org/apache/solr/search/CurrencyRangeFacetCloudTest.java
+++ b/solr/core/src/test/org/apache/solr/search/CurrencyRangeFacetCloudTest.java
@@ -74,6 +74,7 @@ public class CurrencyRangeFacetCloudTest extends SolrCloudTestCase {
.configure();
assertEquals(0, (CollectionAdminRequest.createCollection(COLLECTION, CONF, numShards, numReplicas)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(maxShardsPerNode)
.setProperties(Collections.singletonMap(CoreAdminParams.CONFIG, "solrconfig-minimal.xml"))
.process(cluster.getSolrClient())).getStatus());
diff --git a/solr/core/src/test/org/apache/solr/search/facet/RangeFacetCloudTest.java b/solr/core/src/test/org/apache/solr/search/facet/RangeFacetCloudTest.java
index 8e4cd11..7bdf34f 100644
--- a/solr/core/src/test/org/apache/solr/search/facet/RangeFacetCloudTest.java
+++ b/solr/core/src/test/org/apache/solr/search/facet/RangeFacetCloudTest.java
@@ -93,6 +93,7 @@ public class RangeFacetCloudTest extends SolrCloudTestCase {
.configure();
assertEquals(0, (CollectionAdminRequest.createCollection(COLLECTION, CONF, numShards, numReplicas)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(maxShardsPerNode)
.setProperties(Collections.singletonMap(CoreAdminParams.CONFIG, "solrconfig-minimal.xml"))
.process(cluster.getSolrClient())).getStatus());
diff --git a/solr/core/src/test/org/apache/solr/search/facet/TestCloudJSONFacetJoinDomain.java b/solr/core/src/test/org/apache/solr/search/facet/TestCloudJSONFacetJoinDomain.java
index a88ed95..c1dc549 100644
--- a/solr/core/src/test/org/apache/solr/search/facet/TestCloudJSONFacetJoinDomain.java
+++ b/solr/core/src/test/org/apache/solr/search/facet/TestCloudJSONFacetJoinDomain.java
@@ -109,6 +109,7 @@ public class TestCloudJSONFacetJoinDomain extends SolrCloudTestCase {
collectionProperties.put("config", "solrconfig-tlog.xml");
collectionProperties.put("schema", "schema_latest.xml");
CollectionAdminRequest.createCollection(COLLECTION_NAME, configName, numShards, repFactor)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setProperties(collectionProperties)
.process(cluster.getSolrClient());
diff --git a/solr/core/src/test/org/apache/solr/search/facet/TestCloudJSONFacetSKG.java b/solr/core/src/test/org/apache/solr/search/facet/TestCloudJSONFacetSKG.java
index 0992af8..2c0754a 100644
--- a/solr/core/src/test/org/apache/solr/search/facet/TestCloudJSONFacetSKG.java
+++ b/solr/core/src/test/org/apache/solr/search/facet/TestCloudJSONFacetSKG.java
@@ -139,6 +139,7 @@ public class TestCloudJSONFacetSKG extends SolrCloudTestCase {
collectionProperties.put("config", "solrconfig-tlog.xml");
collectionProperties.put("schema", "schema_latest.xml");
CollectionAdminRequest.createCollection(COLLECTION_NAME, configName, numShards, repFactor)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setProperties(collectionProperties)
.process(cluster.getSolrClient());
diff --git a/solr/core/src/test/org/apache/solr/search/facet/TestCloudJSONFacetSKGEquiv.java b/solr/core/src/test/org/apache/solr/search/facet/TestCloudJSONFacetSKGEquiv.java
index eb68662..aae946e 100644
--- a/solr/core/src/test/org/apache/solr/search/facet/TestCloudJSONFacetSKGEquiv.java
+++ b/solr/core/src/test/org/apache/solr/search/facet/TestCloudJSONFacetSKGEquiv.java
@@ -133,6 +133,7 @@ public class TestCloudJSONFacetSKGEquiv extends SolrCloudTestCase {
collectionProperties.put("config", "solrconfig-tlog.xml");
collectionProperties.put("schema", "schema_latest.xml");
CollectionAdminRequest.createCollection(COLLECTION_NAME, configName, numShards, repFactor)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setProperties(collectionProperties)
.process(cluster.getSolrClient());
diff --git a/solr/core/src/test/org/apache/solr/search/join/BlockJoinFacetDistribTest.java b/solr/core/src/test/org/apache/solr/search/join/BlockJoinFacetDistribTest.java
index c4f0896..2c04d25 100644
--- a/solr/core/src/test/org/apache/solr/search/join/BlockJoinFacetDistribTest.java
+++ b/solr/core/src/test/org/apache/solr/search/join/BlockJoinFacetDistribTest.java
@@ -66,6 +66,7 @@ public class BlockJoinFacetDistribTest extends SolrCloudTestCase{
int shards = 3;
int replicas = 2 ;
CollectionAdminRequest.createCollection(collection, configName, shards, replicas)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setProperties(collectionProperties)
.process(cluster.getSolrClient());
diff --git a/solr/core/src/test/org/apache/solr/search/join/CrossCollectionJoinQueryTest.java b/solr/core/src/test/org/apache/solr/search/join/CrossCollectionJoinQueryTest.java
index ebdb960..a42ebaa 100644
--- a/solr/core/src/test/org/apache/solr/search/join/CrossCollectionJoinQueryTest.java
+++ b/solr/core/src/test/org/apache/solr/search/join/CrossCollectionJoinQueryTest.java
@@ -54,9 +54,11 @@ public class CrossCollectionJoinQueryTest extends SolrCloudTestCase {
CollectionAdminRequest.createCollection("products", "ccjoin", NUM_SHARDS, NUM_REPLICAS)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.process(cluster.getSolrClient());
CollectionAdminRequest.createCollection("parts", "ccjoin", NUM_SHARDS, NUM_REPLICAS)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.process(cluster.getSolrClient());
}
diff --git a/solr/core/src/test/org/apache/solr/search/stats/TestDistribIDF.java b/solr/core/src/test/org/apache/solr/search/stats/TestDistribIDF.java
index 205b30b..c7f4564 100644
--- a/solr/core/src/test/org/apache/solr/search/stats/TestDistribIDF.java
+++ b/solr/core/src/test/org/apache/solr/search/stats/TestDistribIDF.java
@@ -29,6 +29,7 @@ import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.client.solrj.response.CollectionAdminResponse;
import org.apache.solr.client.solrj.response.QueryResponse;
import org.apache.solr.cloud.MiniSolrCloudCluster;
+import org.apache.solr.cloud.SolrCloudTestCase;
import org.apache.solr.common.SolrInputDocument;
import org.apache.solr.common.cloud.CompositeIdRouter;
import org.apache.solr.common.cloud.ImplicitDocRouter;
@@ -201,7 +202,7 @@ public class TestDistribIDF extends SolrTestCaseJ4 {
response = create.process(solrCluster.getSolrClient());
solrCluster.waitForActiveCollection(name, 3, 3);
} else {
- CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection(name,config,2,1);
+ CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection(name,config,2,1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE);
create.setMaxShardsPerNode(1);
response = create.process(solrCluster.getSolrClient());
solrCluster.waitForActiveCollection(name, 2, 2);
diff --git a/solr/core/src/test/org/apache/solr/servlet/HttpSolrCallGetCoreTest.java b/solr/core/src/test/org/apache/solr/servlet/HttpSolrCallGetCoreTest.java
index 4f94388..3889f32 100644
--- a/solr/core/src/test/org/apache/solr/servlet/HttpSolrCallGetCoreTest.java
+++ b/solr/core/src/test/org/apache/solr/servlet/HttpSolrCallGetCoreTest.java
@@ -48,6 +48,7 @@ public class HttpSolrCallGetCoreTest extends SolrCloudTestCase {
CollectionAdminRequest
.createCollection(COLLECTION, "config", NUM_SHARD, REPLICA_FACTOR)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(NUM_SHARD * REPLICA_FACTOR)
.process(cluster.getSolrClient());
AbstractDistribZkTestBase.waitForRecoveriesToFinish(COLLECTION, cluster.getSolrClient().getZkStateReader(),
diff --git a/solr/core/src/test/org/apache/solr/update/TestInPlaceUpdateWithRouteField.java b/solr/core/src/test/org/apache/solr/update/TestInPlaceUpdateWithRouteField.java
index fc85fad..1e0b6f8 100644
--- a/solr/core/src/test/org/apache/solr/update/TestInPlaceUpdateWithRouteField.java
+++ b/solr/core/src/test/org/apache/solr/update/TestInPlaceUpdateWithRouteField.java
@@ -73,6 +73,7 @@ public class TestInPlaceUpdateWithRouteField extends SolrCloudTestCase {
boolean implicit = random().nextBoolean();
String routerName = implicit ? "implicit":"compositeId";
Create createCmd = CollectionAdminRequest.createCollection(COLLECTION, configName, shards.length, replicas)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(shards.length * replicas)
.setProperties(collectionProperties)
.setRouterName(routerName)
diff --git a/solr/core/src/test/org/apache/solr/update/processor/AtomicUpdateJavabinTest.java b/solr/core/src/test/org/apache/solr/update/processor/AtomicUpdateJavabinTest.java
deleted file mode 100644
index 758de5f..0000000
--- a/solr/core/src/test/org/apache/solr/update/processor/AtomicUpdateJavabinTest.java
+++ /dev/null
@@ -1,370 +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.update.processor;
-
-import org.apache.solr.client.solrj.request.CollectionAdminRequest;
-import org.apache.solr.client.solrj.request.QueryRequest;
-import org.apache.solr.client.solrj.request.UpdateRequest;
-import org.apache.solr.client.solrj.response.QueryResponse;
-import org.apache.solr.cloud.SolrCloudTestCase;
-import org.apache.solr.common.SolrDocument;
-import org.apache.solr.common.SolrInputDocument;
-import org.apache.solr.common.params.ModifiableSolrParams;
-import org.apache.solr.common.util.NamedList;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Test;
-
-import java.time.Instant;
-import java.util.Collection;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * Tests Solr's atomic-update functionality using requests sent through SolrJ using wt=javabin
- *
- * {@link AtomicUpdatesTest} covers some of the same functionality, but does so by making xml-based requests. Recent
- * changes to Solr have made it possible for the same data sent with different formats to result in different NamedLists
- * after unmarshalling, so the test duplication is now necessary. See SOLR-13331 for an example.
- */
-public class AtomicUpdateJavabinTest extends SolrCloudTestCase {
- private static final String COMMITTED_DOC_ID = "1";
- private static final String COMMITTED_DOC_STR_VALUES_ID = "1s";
- private static final String UNCOMMITTED_DOC_ID = "2";
- private static final String UNCOMMITTED_DOC_STR_VALUES_ID = "2s";
- private static final String COLLECTION = "collection1";
- private static final int NUM_SHARDS = 1;
- private static final int NUM_REPLICAS = 1;
- private static final Date DATE_1 = Date.from(Instant.ofEpochSecond(1554243309));
- private static final String DATE_1_STR = "2019-04-02T22:15:09Z";
- private static final Date DATE_2 = Date.from(Instant.ofEpochSecond(1554243609));
- private static final String DATE_2_STR = "2019-04-02T22:20:09Z";
- private static final Date DATE_3 = Date.from(Instant.ofEpochSecond(1554243909));
- private static final String DATE_3_STR = "2019-04-02T22:25:09Z";
-
-
- @BeforeClass
- public static void setupCluster() throws Exception {
- configureCluster(1)
- .addConfig("conf", configset("cloud-dynamic"))
- .configure();
-
- CollectionAdminRequest.createCollection(COLLECTION, "conf", NUM_SHARDS, NUM_REPLICAS)
- .process(cluster.getSolrClient());
-
- cluster.waitForActiveCollection(COLLECTION, 1, 1);
- }
-
- @Before
- public void setUp() throws Exception {
- super.setUp();
-
- final SolrInputDocument committedDoc = sdoc(
- "id", COMMITTED_DOC_ID,
- "title_s", "title_1", "title_s", "title_2",
- "tv_mv_text", "text_1", "tv_mv_text", "text_2",
- "count_is", 1, "count_is", 2,
- "count_md", 1.0, "count_md", 2.0,
- "timestamps_mdt", DATE_1, "timestamps_mdt", DATE_2);
- final SolrInputDocument committedStrDoc = sdoc(
- "id", COMMITTED_DOC_STR_VALUES_ID,
- "title_s", "title_1", "title_s", "title_2",
- "tv_mv_text", "text_1", "tv_mv_text", "text_2",
- "count_is", "1", "count_is", "2",
- "count_md", "1.0", "count_md", "2.0",
- "timestamps_mdt", DATE_1_STR, "timestamps_mdt", DATE_2_STR);
- final UpdateRequest committedRequest = new UpdateRequest()
- .add(committedDoc)
- .add(committedStrDoc);
- committedRequest.commit(cluster.getSolrClient(), COLLECTION);
-
- // Upload a copy of id:1 that's uncommitted to test how atomic-updates modify values in the tlog
- // See SOLR-14971 for an example of why this case needs tested separately
- final SolrInputDocument uncommittedDoc = sdoc(
- "id", UNCOMMITTED_DOC_ID,
- "title_s", "title_1", "title_s", "title_2",
- "tv_mv_text", "text_1", "tv_mv_text", "text_2",
- "count_is", 1, "count_is", 2,
- "count_md", 1.0, "count_md", 2.0,
- "timestamps_mdt", DATE_1, "timestamps_mdt", DATE_2);
- final SolrInputDocument uncommittedStrDoc = sdoc(
- "id", UNCOMMITTED_DOC_STR_VALUES_ID,
- "title_s", "title_1", "title_s", "title_2",
- "tv_mv_text", "text_1", "tv_mv_text", "text_2",
- "count_is", "1", "count_is", "2",
- "count_md", "1.0", "count_md", "2.0",
- "timestamps_mdt", DATE_1_STR, "timestamps_mdt", DATE_2_STR);
- final UpdateRequest uncommittedRequest = new UpdateRequest()
- .add(uncommittedDoc)
- .add(uncommittedStrDoc);
- uncommittedRequest.process(cluster.getSolrClient(), COLLECTION);
- }
-
- @Test
- public void testAtomicUpdateRemovalOfStrField() throws Exception {
- ensureFieldHasValues(COMMITTED_DOC_ID, "title_s", "title_1", "title_2");
- atomicRemoveValueFromField(COMMITTED_DOC_ID, "title_s", "title_1");
- ensureFieldHasValues(COMMITTED_DOC_ID, "title_s", "title_2");
-
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "title_s", "title_1", "title_2");
- atomicRemoveValueFromField(UNCOMMITTED_DOC_ID, "title_s", "title_1");
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "title_s", "title_2");
- }
-
- @Test
- public void testAtomicUpdateRemovalOfTextField() throws Exception {
- ensureFieldHasValues(COMMITTED_DOC_ID, "tv_mv_text", "text_1", "text_2");
- atomicRemoveValueFromField(COMMITTED_DOC_ID, "tv_mv_text", "text_1");
- ensureFieldHasValues(COMMITTED_DOC_ID, "tv_mv_text", "text_2");
-
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "tv_mv_text", "text_1", "text_2");
- atomicRemoveValueFromField(UNCOMMITTED_DOC_ID, "tv_mv_text", "text_1");
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "tv_mv_text", "text_2");
- }
-
- @Test
- public void testAtomicUpdateRemovalOfIntField() throws Exception {
- ensureFieldHasValues(COMMITTED_DOC_ID, "count_is", 1, 2);
- atomicRemoveValueFromField(COMMITTED_DOC_ID, "count_is", 1);
- ensureFieldHasValues(COMMITTED_DOC_ID, "count_is", 2);
-
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "count_is", 1, 2);
- atomicRemoveValueFromField(UNCOMMITTED_DOC_ID, "count_is", 1);
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "count_is", 2);
- }
-
- @Test
- public void testAtomicUpdateRemovalOfDoubleField() throws Exception {
- ensureFieldHasValues(COMMITTED_DOC_ID, "count_md", 1.0, 2.0);
- atomicRemoveValueFromField(COMMITTED_DOC_ID, "count_md", 1.0);
- ensureFieldHasValues(COMMITTED_DOC_ID, "count_md", 2.0);
-
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "count_md", 1.0, 2.0);
- atomicRemoveValueFromField(UNCOMMITTED_DOC_ID, "count_md", 1.0);
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "count_md", 2.0);
- }
-
- @Test
- public void testAtomicUpdateRemovalOfDateField() throws Exception {
- ensureFieldHasValues(COMMITTED_DOC_ID, "timestamps_mdt", DATE_1, DATE_2);
- atomicRemoveValueFromField(COMMITTED_DOC_ID, "timestamps_mdt", DATE_1);
- ensureFieldHasValues(COMMITTED_DOC_ID, "timestamps_mdt", DATE_2);
-
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "timestamps_mdt", DATE_1, DATE_2);
- atomicRemoveValueFromField(UNCOMMITTED_DOC_ID, "timestamps_mdt", DATE_1);
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "timestamps_mdt", DATE_2);
- }
-
- @Test
- public void testAtomicUpdateAddDistinctOfDistinctValueOnStrField() throws Exception {
- ensureFieldHasValues(COMMITTED_DOC_ID, "title_s", "title_1", "title_2");
- atomicAddDistinctValueToField(COMMITTED_DOC_ID, "title_s", "title_3");
- ensureFieldHasValues(COMMITTED_DOC_ID, "title_s", "title_1", "title_2", "title_3");
-
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "title_s", "title_1", "title_2");
- atomicAddDistinctValueToField(UNCOMMITTED_DOC_ID, "title_s", "title_3");
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "title_s", "title_1", "title_2", "title_3");
- }
-
- @Test
- public void testAtomicUpdateAddDistinctOfDuplicateValueOnStrField() throws Exception {
- ensureFieldHasValues(COMMITTED_DOC_ID, "title_s", "title_1", "title_2");
- atomicAddDistinctValueToField(COMMITTED_DOC_ID, "title_s", "title_2");
- ensureFieldHasValues(COMMITTED_DOC_ID, "title_s", "title_1", "title_2");
-
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "title_s", "title_1", "title_2");
- atomicAddDistinctValueToField(UNCOMMITTED_DOC_ID, "title_s", "title_2");
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "title_s", "title_1", "title_2");
- }
-
- @Test
- public void testAtomicUpdateAddDistinctOfDistinctValueOnTextField() throws Exception {
- ensureFieldHasValues(COMMITTED_DOC_ID, "tv_mv_text", "text_1", "text_2");
- atomicAddDistinctValueToField(COMMITTED_DOC_ID, "tv_mv_text", "text_3");
- ensureFieldHasValues(COMMITTED_DOC_ID, "tv_mv_text", "text_1", "text_2", "text_3");
-
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "tv_mv_text", "text_1", "text_2");
- atomicAddDistinctValueToField(UNCOMMITTED_DOC_ID, "tv_mv_text", "text_3");
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "tv_mv_text", "text_1", "text_2", "text_3");
- }
-
- @Test
- public void testAtomicUpdateAddDistinctOfDuplicateValueOnTextField() throws Exception {
- ensureFieldHasValues(COMMITTED_DOC_ID, "tv_mv_text", "text_1", "text_2");
- atomicAddDistinctValueToField(COMMITTED_DOC_ID, "tv_mv_text", "text_2");
- ensureFieldHasValues(COMMITTED_DOC_ID, "tv_mv_text", "text_1", "text_2");
-
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "tv_mv_text", "text_1", "text_2");
- atomicAddDistinctValueToField(UNCOMMITTED_DOC_ID, "tv_mv_text", "text_2");
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "tv_mv_text", "text_1", "text_2");
- }
-
- @Test
- public void testAtomicUpdateAddDistinctOfDistinctValueOnIntField() throws Exception {
- ensureFieldHasValues(COMMITTED_DOC_ID, "count_is", 1, 2);
- atomicAddDistinctValueToField(COMMITTED_DOC_ID, "count_is", 3);
- ensureFieldHasValues(COMMITTED_DOC_ID, "count_is", 1, 2, 3);
-
- ensureFieldHasValues(COMMITTED_DOC_STR_VALUES_ID, "count_is", 1, 2);
- atomicAddDistinctValueToField(COMMITTED_DOC_STR_VALUES_ID, "count_is", 3);
- ensureFieldHasValues(COMMITTED_DOC_STR_VALUES_ID, "count_is", 1, 2, 3);
-
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "count_is", 1, 2);
- atomicAddDistinctValueToField(UNCOMMITTED_DOC_ID, "count_is", 3);
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "count_is", 1, 2, 3);
-
- ensureFieldHasValues(UNCOMMITTED_DOC_STR_VALUES_ID, "count_is", 1, 2);
- atomicAddDistinctValueToField(UNCOMMITTED_DOC_STR_VALUES_ID, "count_is", 3);
- ensureFieldHasValues(UNCOMMITTED_DOC_STR_VALUES_ID, "count_is", 1, 2, 3);
- }
-
- @Test
- public void testAtomicUpdateAddDistinctOfDuplicateValueOnIntField() throws Exception {
- ensureFieldHasValues(COMMITTED_DOC_ID, "count_is", 1, 2);
- atomicAddDistinctValueToField(COMMITTED_DOC_ID, "count_is", 2);
- ensureFieldHasValues(COMMITTED_DOC_ID, "count_is", 1, 2);
-
- ensureFieldHasValues(COMMITTED_DOC_STR_VALUES_ID, "count_is", 1, 2);
- atomicAddDistinctValueToField(COMMITTED_DOC_STR_VALUES_ID, "count_is", 2);
- ensureFieldHasValues(COMMITTED_DOC_STR_VALUES_ID, "count_is", 1, 2);
-
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "count_is", 1, 2);
- atomicAddDistinctValueToField(UNCOMMITTED_DOC_ID, "count_is", 2);
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "count_is", 1, 2);
-
- ensureFieldHasValues(UNCOMMITTED_DOC_STR_VALUES_ID, "count_is", 1, 2);
- atomicAddDistinctValueToField(UNCOMMITTED_DOC_STR_VALUES_ID, "count_is", 2);
- ensureFieldHasValues(UNCOMMITTED_DOC_STR_VALUES_ID, "count_is", 1, 2);
- }
-
- @Test
- public void testAtomicUpdateAddDistinctOfDistinctValueOnDoubleField() throws Exception {
- ensureFieldHasValues(COMMITTED_DOC_ID, "count_md", 1.0, 2.0);
- atomicAddDistinctValueToField(COMMITTED_DOC_ID, "count_md", 3.0);
- ensureFieldHasValues(COMMITTED_DOC_ID, "count_md", 1.0, 2.0, 3.0);
-
- ensureFieldHasValues(COMMITTED_DOC_STR_VALUES_ID, "count_md", 1.0, 2.0);
- atomicAddDistinctValueToField(COMMITTED_DOC_STR_VALUES_ID, "count_md", 3.0);
- ensureFieldHasValues(COMMITTED_DOC_STR_VALUES_ID, "count_md", 1.0, 2.0, 3.0);
-
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "count_md", 1.0, 2.0);
- atomicAddDistinctValueToField(UNCOMMITTED_DOC_ID, "count_md", 3.0);
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "count_md", 1.0, 2.0, 3.0);
-
- ensureFieldHasValues(UNCOMMITTED_DOC_STR_VALUES_ID, "count_md", 1.0, 2.0);
- atomicAddDistinctValueToField(UNCOMMITTED_DOC_STR_VALUES_ID, "count_md", 3.0);
- ensureFieldHasValues(UNCOMMITTED_DOC_STR_VALUES_ID, "count_md", 1.0, 2.0, 3.0);
- }
-
- @Test
- public void testAtomicUpdateAddDistinctOfDuplicateValueOnDoubleField() throws Exception {
- ensureFieldHasValues(COMMITTED_DOC_ID, "count_md", 1.0, 2.0);
- atomicAddDistinctValueToField(COMMITTED_DOC_ID, "count_md", 2.0);
- ensureFieldHasValues(COMMITTED_DOC_ID, "count_md", 1.0, 2.0);
-
- ensureFieldHasValues(COMMITTED_DOC_STR_VALUES_ID, "count_md", 1.0, 2.0);
- atomicAddDistinctValueToField(COMMITTED_DOC_STR_VALUES_ID, "count_md", 2.0);
- ensureFieldHasValues(COMMITTED_DOC_STR_VALUES_ID, "count_md", 1.0, 2.0);
-
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "count_md", 1.0, 2.0);
- atomicAddDistinctValueToField(UNCOMMITTED_DOC_ID, "count_md", 2.0);
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "count_md", 1.0, 2.0);
-
- ensureFieldHasValues(UNCOMMITTED_DOC_STR_VALUES_ID, "count_md", 1.0, 2.0);
- atomicAddDistinctValueToField(UNCOMMITTED_DOC_STR_VALUES_ID, "count_md", 2.0);
- ensureFieldHasValues(UNCOMMITTED_DOC_STR_VALUES_ID, "count_md", 1.0, 2.0);
- }
-
- @Test
- public void testAtomicUpdateAddDistinctOfDistinctValueOnDateField() throws Exception {
- ensureFieldHasValues(COMMITTED_DOC_ID, "timestamps_mdt", DATE_1, DATE_2);
- atomicAddDistinctValueToField(COMMITTED_DOC_ID, "timestamps_mdt", DATE_3);
- ensureFieldHasValues(COMMITTED_DOC_ID, "timestamps_mdt", DATE_1, DATE_2, DATE_3);
-
- ensureFieldHasValues(COMMITTED_DOC_STR_VALUES_ID, "timestamps_mdt", DATE_1, DATE_2);
- atomicAddDistinctValueToField(COMMITTED_DOC_STR_VALUES_ID, "timestamps_mdt", DATE_3);
- ensureFieldHasValues(COMMITTED_DOC_STR_VALUES_ID, "timestamps_mdt", DATE_1, DATE_2, DATE_3);
-
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "timestamps_mdt", DATE_1, DATE_2);
- atomicAddDistinctValueToField(UNCOMMITTED_DOC_ID, "timestamps_mdt", DATE_3);
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "timestamps_mdt", DATE_1, DATE_2, DATE_3);
-
- ensureFieldHasValues(UNCOMMITTED_DOC_STR_VALUES_ID, "timestamps_mdt", DATE_1, DATE_2);
- atomicAddDistinctValueToField(UNCOMMITTED_DOC_STR_VALUES_ID, "timestamps_mdt", DATE_3);
- ensureFieldHasValues(UNCOMMITTED_DOC_STR_VALUES_ID, "timestamps_mdt", DATE_1, DATE_2, DATE_3);
- }
-
- @Test
- public void testAtomicUpdateAddDistinctOfDuplicateValueOnDateField() throws Exception {
- ensureFieldHasValues(COMMITTED_DOC_ID, "timestamps_mdt", DATE_1, DATE_2);
- atomicAddDistinctValueToField(COMMITTED_DOC_ID, "timestamps_mdt", DATE_2);
- ensureFieldHasValues(COMMITTED_DOC_ID, "timestamps_mdt", DATE_1, DATE_2);
-
- ensureFieldHasValues(COMMITTED_DOC_STR_VALUES_ID, "timestamps_mdt", DATE_1, DATE_2);
- atomicAddDistinctValueToField(COMMITTED_DOC_STR_VALUES_ID, "timestamps_mdt", DATE_2);
- ensureFieldHasValues(COMMITTED_DOC_STR_VALUES_ID, "timestamps_mdt", DATE_1, DATE_2);
-
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "timestamps_mdt", DATE_1, DATE_2);
- atomicAddDistinctValueToField(UNCOMMITTED_DOC_ID, "timestamps_mdt", DATE_2);
- ensureFieldHasValues(UNCOMMITTED_DOC_ID, "timestamps_mdt", DATE_1, DATE_2);
-
- ensureFieldHasValues(UNCOMMITTED_DOC_STR_VALUES_ID, "timestamps_mdt", DATE_1, DATE_2);
- atomicAddDistinctValueToField(UNCOMMITTED_DOC_STR_VALUES_ID, "timestamps_mdt", DATE_2);
- ensureFieldHasValues(UNCOMMITTED_DOC_STR_VALUES_ID, "timestamps_mdt", DATE_1, DATE_2);
- }
-
- private void atomicRemoveValueFromField(String docId, String fieldName, Object value) throws Exception {
- final SolrInputDocument doc = new SolrInputDocument();
- doc.setField("id", docId);
- Map<String, Object> atomicUpdateRemoval = new HashMap<>(1);
- atomicUpdateRemoval.put("remove", value);
- doc.setField(fieldName, atomicUpdateRemoval);
-
- cluster.getSolrClient().add(COLLECTION, doc);
- }
-
- private void atomicAddDistinctValueToField(String docId, String fieldName, Object value) throws Exception {
- final SolrInputDocument doc = new SolrInputDocument();
- doc.setField("id", docId);
- Map<String, Object> atomicUpdateRemoval = new HashMap<>(1);
- atomicUpdateRemoval.put("add-distinct", value);
- doc.setField(fieldName, atomicUpdateRemoval);
-
- cluster.getSolrClient().add(COLLECTION, doc);
- }
-
- private void ensureFieldHasValues(String identifyingDocId, String fieldName, Object... expectedValues) throws Exception {
- final ModifiableSolrParams solrParams = new ModifiableSolrParams();
- solrParams.set("id", identifyingDocId);
- QueryRequest request = new QueryRequest(solrParams);
- request.setPath("/get");
- final QueryResponse response = request.process(cluster.getSolrClient(), COLLECTION);
-
- final NamedList<Object> rawResponse = response.getResponse();
- assertTrue(rawResponse.get("doc") != null);
- assertTrue(rawResponse.get("doc") instanceof SolrDocument);
- final SolrDocument doc = (SolrDocument) rawResponse.get("doc");
- final Collection<Object> valuesAfterUpdate = doc.getFieldValues(fieldName);
- assertEquals("Expected field to have " + expectedValues.length + " values, but found " + valuesAfterUpdate.size(),
- expectedValues.length, valuesAfterUpdate.size());
- for (Object expectedValue: expectedValues) {
- assertTrue("Expected value [" + expectedValue + "] was not found in field", valuesAfterUpdate.contains(expectedValue));
- }
- }
-}
diff --git a/solr/core/src/test/org/apache/solr/update/processor/DimensionalRoutedAliasUpdateProcessorTest.java b/solr/core/src/test/org/apache/solr/update/processor/DimensionalRoutedAliasUpdateProcessorTest.java
index 30faefa..4261fd7 100644
--- a/solr/core/src/test/org/apache/solr/update/processor/DimensionalRoutedAliasUpdateProcessorTest.java
+++ b/solr/core/src/test/org/apache/solr/update/processor/DimensionalRoutedAliasUpdateProcessorTest.java
@@ -35,6 +35,7 @@ import org.apache.solr.client.solrj.request.CollectionAdminRequest.CreateTimeRou
import org.apache.solr.client.solrj.response.FieldStatsInfo;
import org.apache.solr.client.solrj.response.QueryResponse;
import org.apache.solr.client.solrj.response.UpdateResponse;
+import org.apache.solr.cloud.SolrCloudTestCase;
import org.apache.solr.cloud.api.collections.CategoryRoutedAlias;
import org.apache.solr.cloud.api.collections.TimeRoutedAlias;
import org.apache.solr.common.SolrDocument;
@@ -100,6 +101,7 @@ public class DimensionalRoutedAliasUpdateProcessorTest extends RoutedAliasUpdate
CollectionAdminRequest.DimensionalRoutedAlias dra = CollectionAdminRequest.createDimensionalRoutedAlias(getAlias(),
CollectionAdminRequest.createCollection("_unused_", configName, 2, 2)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(2), TRA_Dim, CRA_Dim);
SolrParams params = dra.getParams();
@@ -361,6 +363,7 @@ public class DimensionalRoutedAliasUpdateProcessorTest extends RoutedAliasUpdate
CollectionAdminRequest.DimensionalRoutedAlias dra = CollectionAdminRequest.createDimensionalRoutedAlias(getAlias(),
CollectionAdminRequest.createCollection("_unused_", configName, 2, 2)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(2), CRA_Dim, TRA_Dim);
SolrParams params = dra.getParams();
diff --git a/solr/core/src/test/org/apache/solr/update/processor/TemplateUpdateProcessorTest.java b/solr/core/src/test/org/apache/solr/update/processor/TemplateUpdateProcessorTest.java
index 6eb122a..c8ae8b2 100644
--- a/solr/core/src/test/org/apache/solr/update/processor/TemplateUpdateProcessorTest.java
+++ b/solr/core/src/test/org/apache/solr/update/processor/TemplateUpdateProcessorTest.java
@@ -88,7 +88,7 @@ public class TemplateUpdateProcessorTest extends SolrCloudTestCase {
params.add("commit", "true");
UpdateRequest add = new UpdateRequest().add(solrDoc);
add.setParams(params);
- NamedList<Object> result = cluster.getSolrClient().request(CollectionAdminRequest.createCollection("c", "conf1", 1, 1));
+ NamedList<Object> result = cluster.getSolrClient().request(CollectionAdminRequest.createCollection("c", "conf1", 1, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE));
Utils.toJSONString(result.asMap(4));
AbstractFullDistribZkTestBase.waitForCollection(cluster.getSolrClient().getZkStateReader(), "c",1);
cluster.getSolrClient().request(add, "c");
diff --git a/solr/core/src/test/org/apache/solr/util/TestExportTool.java b/solr/core/src/test/org/apache/solr/util/TestExportTool.java
index 7b34958..6427b48 100644
--- a/solr/core/src/test/org/apache/solr/util/TestExportTool.java
+++ b/solr/core/src/test/org/apache/solr/util/TestExportTool.java
@@ -56,6 +56,7 @@ public class TestExportTool extends SolrCloudTestCase {
try {
CollectionAdminRequest
.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(100)
.process(cluster.getSolrClient());
cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
@@ -131,6 +132,7 @@ public class TestExportTool extends SolrCloudTestCase {
try {
CollectionAdminRequest
.createCollection(COLLECTION_NAME, "conf", 8, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(10)
.process(cluster.getSolrClient());
cluster.waitForActiveCollection(COLLECTION_NAME, 8, 8);
diff --git a/solr/core/src/test/org/apache/solr/util/tracing/TestDistributedTracing.java b/solr/core/src/test/org/apache/solr/util/tracing/TestDistributedTracing.java
index 23f75ec..fc5dcb2 100644
--- a/solr/core/src/test/org/apache/solr/util/tracing/TestDistributedTracing.java
+++ b/solr/core/src/test/org/apache/solr/util/tracing/TestDistributedTracing.java
@@ -56,6 +56,7 @@ public class TestDistributedTracing extends SolrCloudTestCase {
waitForSampleRateUpdated(1.0);
CollectionAdminRequest
.createCollection(COLLECTION, "config", 2, 2)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.process(cluster.getSolrClient());
cluster.waitForActiveCollection(COLLECTION, 2, 4);
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/DistribStateManager.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/DistribStateManager.java
index f811c5a..f459a65 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/DistribStateManager.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/DistribStateManager.java
@@ -30,6 +30,7 @@ import org.apache.solr.client.solrj.cloud.autoscaling.BadVersionException;
import org.apache.solr.client.solrj.cloud.autoscaling.NotEmptyException;
import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
import org.apache.solr.common.SolrCloseable;
+import org.apache.solr.common.cloud.PerReplicaStates;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Op;
@@ -111,6 +112,12 @@ public interface DistribStateManager extends SolrCloseable {
return tree;
}
+ default PerReplicaStates getReplicaStates(String path) throws KeeperException, InterruptedException {
+ throw new UnsupportedOperationException("Not implemented");
+
+
+ }
+
/**
* Remove data recursively.
* @param root root path
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientCloudManager.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientCloudManager.java
index 5ad7ff4..7724b58 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientCloudManager.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientCloudManager.java
@@ -187,6 +187,9 @@ public class SolrClientCloudManager implements SolrCloudManager {
return EMPTY;
}
}
+ public SolrZkClient getZkClient() {
+ return zkClient;
+ }
@Override
public DistributedQueueFactory getDistributedQueueFactory() {
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkDistribStateManager.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkDistribStateManager.java
index b15445b..c1c30de 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkDistribStateManager.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkDistribStateManager.java
@@ -30,6 +30,7 @@ import org.apache.solr.client.solrj.cloud.DistribStateManager;
import org.apache.solr.client.solrj.cloud.autoscaling.NotEmptyException;
import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
import org.apache.solr.common.AlreadyClosedException;
+import org.apache.solr.common.cloud.PerReplicaStates;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.AutoScalingParams;
@@ -207,4 +208,9 @@ public class ZkDistribStateManager implements DistribStateManager {
public SolrZkClient getZkClient() {
return zkClient;
}
+
+ @Override
+ public PerReplicaStates getReplicaStates(String path) throws KeeperException, InterruptedException {
+ return PerReplicaStates.fetch(path, zkClient, null);
+ }
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/request/CollectionAdminRequest.java b/solr/solrj/src/java/org/apache/solr/client/solrj/request/CollectionAdminRequest.java
index 432c901..21ef417 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/request/CollectionAdminRequest.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/request/CollectionAdminRequest.java
@@ -56,6 +56,7 @@ import org.apache.solr.common.params.SolrParams;
import org.apache.solr.common.util.NamedList;
import static org.apache.solr.client.solrj.cloud.autoscaling.Policy.POLICY;
+import static org.apache.solr.common.cloud.DocCollection.PER_REPLICA_STATE;
import static org.apache.solr.common.cloud.DocCollection.RULE;
import static org.apache.solr.common.cloud.DocCollection.SNITCH;
import static org.apache.solr.common.cloud.ZkStateReader.AUTO_ADD_REPLICAS;
@@ -94,6 +95,7 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
COLL_CONF,
WITH_COLLECTION,
COLOCATED_WITH,
+ PER_REPLICA_STATE,
READ_ONLY);
protected final CollectionAction action;
@@ -443,6 +445,7 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
protected Integer nrtReplicas;
protected Integer pullReplicas;
protected Integer tlogReplicas;
+ protected Boolean perReplicaState;
protected Properties properties;
protected Boolean autoAddReplicas;
@@ -489,6 +492,7 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
public Create setStateFormat(Integer stateFormat) { this.stateFormat = stateFormat; return this; }
public Create setRule(String... s){ this.rule = s; return this; }
public Create setSnitch(String... s){ this.snitch = s; return this; }
+ public Create setPerReplicaState(Boolean b) {this.perReplicaState = b; return this; }
public Create setAlias(String alias) {
this.alias = alias;
@@ -507,6 +511,7 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
public Boolean getAutoAddReplicas() { return autoAddReplicas; }
public Integer getNumTlogReplicas() {return tlogReplicas;}
public Integer getNumPullReplicas() {return pullReplicas;}
+ public Boolean getPerReplicaState() {return perReplicaState;}
public Integer getStateFormat() { return stateFormat; }
@@ -588,6 +593,9 @@ public abstract class CollectionAdminRequest<T extends CollectionAdminResponse>
if (tlogReplicas != null) {
params.set(ZkStateReader.TLOG_REPLICAS, tlogReplicas);
}
+ if(Boolean.TRUE.equals(perReplicaState)) {
+ params.set(PER_REPLICA_STATE, perReplicaState);
+ }
if (rule != null) params.set(DocCollection.RULE, rule);
if (snitch != null) params.set(DocCollection.SNITCH, snitch);
params.setNonNull(POLICY, policy);
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/ClusterState.java b/solr/solrj/src/java/org/apache/solr/common/cloud/ClusterState.java
index 1e44912..65bcc97 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/ClusterState.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/ClusterState.java
@@ -16,6 +16,7 @@
*/
package org.apache.solr.common.cloud;
+import java.lang.invoke.MethodHandles;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -23,15 +24,19 @@ import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
+import java.util.function.Supplier;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.common.util.Utils;
import org.apache.zookeeper.KeeperException;
import org.noggit.JSONWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Immutable state of the cloud. Normally you can get the state by using
@@ -39,6 +44,8 @@ import org.noggit.JSONWriter;
* @lucene.experimental
*/
public class ClusterState implements JSONWriter.Writable {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
private final Integer znodeVersion;
@@ -256,6 +263,12 @@ public class ClusterState implements JSONWriter.Writable {
Map<String,Object> props;
Map<String,Slice> slices;
+ if("true".equals(String.valueOf(objs.get(DocCollection.PER_REPLICA_STATE)))) {
+ log.info("a collection {} has per-replica state" , name);
+ //this collection has replica states stored outside
+ ReplicaStatesProvider rsp = REPLICASTATES_PROVIDER.get();
+ if (rsp instanceof StatesProvider) ((StatesProvider) rsp).isPerReplicaState = true;
+ }
@SuppressWarnings({"unchecked"})
Map<String, Object> sliceObjs = (Map<String, Object>) objs.get(DocCollection.SHARDS);
if (sliceObjs == null) {
@@ -420,5 +433,63 @@ public class ClusterState implements JSONWriter.Writable {
public int size() {
return collectionStates.size();
}
+ interface ReplicaStatesProvider {
+
+ Optional<ReplicaStatesProvider> get();
+
+ PerReplicaStates getStates();
+
+ }
+
+ private static final ReplicaStatesProvider EMPTYSTATEPROVIDER = new ReplicaStatesProvider() {
+ @Override
+ public Optional<ReplicaStatesProvider> get() {
+ return Optional.empty();
+ }
+
+ @Override
+ public PerReplicaStates getStates() {
+ throw new RuntimeException("Invalid operation");
+ }
+
+ };
+
+ private static ThreadLocal<ReplicaStatesProvider> REPLICASTATES_PROVIDER = new ThreadLocal<>();
+
+
+ public static ReplicaStatesProvider getReplicaStatesProvider() {
+ return (REPLICASTATES_PROVIDER.get() == null)? EMPTYSTATEPROVIDER: REPLICASTATES_PROVIDER.get() ;
+ }
+ public static void initReplicaStateProvider(Supplier<PerReplicaStates> replicaStatesSupplier) {
+ REPLICASTATES_PROVIDER.set(new StatesProvider(replicaStatesSupplier));
+ }
+
+
+ public static void clearReplicaStateProvider(){
+ REPLICASTATES_PROVIDER.remove();
+ }
+
+ private static class StatesProvider implements ReplicaStatesProvider {
+ private final Supplier<PerReplicaStates> replicaStatesSupplier;
+ private PerReplicaStates perReplicaStates;
+ private boolean isPerReplicaState = false;
+
+ public StatesProvider(Supplier<PerReplicaStates> replicaStatesSupplier) {
+ this.replicaStatesSupplier = replicaStatesSupplier;
+ }
+
+ @Override
+ public Optional<ReplicaStatesProvider> get() {
+ return isPerReplicaState ? Optional.of(this) : Optional.empty();
+ }
+
+ @Override
+ public PerReplicaStates getStates() {
+ if(perReplicaStates == null) perReplicaStates = replicaStatesSupplier.get();
+ return perReplicaStates;
+ }
+
+ }
+
}
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/DocCollection.java b/solr/solrj/src/java/org/apache/solr/common/cloud/DocCollection.java
index 38bbeae..f52b1dd 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/DocCollection.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/DocCollection.java
@@ -16,6 +16,7 @@
*/
package org.apache.solr.common.cloud;
+import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Collection;
import java.util.EnumSet;
@@ -33,6 +34,8 @@ import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrException.ErrorCode;
import org.noggit.JSONWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import static org.apache.solr.common.cloud.ZkStateReader.AUTO_ADD_REPLICAS;
import static org.apache.solr.common.cloud.ZkStateReader.MAX_SHARDS_PER_NODE;
@@ -47,9 +50,12 @@ import static org.apache.solr.common.util.Utils.toJSONString;
* Models a Collection in zookeeper (but that Java name is obviously taken, hence "DocCollection")
*/
public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
public static final String DOC_ROUTER = "router";
public static final String SHARDS = "shards";
+ public static final String PER_REPLICA_STATE = "perReplicaState";
public static final String STATE_FORMAT = "stateFormat";
/** @deprecated to be removed in Solr 9.0 (see SOLR-14930)
*
@@ -80,6 +86,10 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
private final Boolean autoAddReplicas;
private final String policy;
private final Boolean readOnly;
+ private final Boolean perReplicaState;
+ private final Map<String, Replica> replicaMap = new HashMap<>();
+ private volatile PerReplicaStates perReplicaStates;
+
public DocCollection(String name, Map<String, Slice> slices, Map<String, Object> props, DocRouter router) {
this(name, slices, props, router, Integer.MAX_VALUE, ZkStateReader.CLUSTER_STATE);
@@ -105,8 +115,10 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
this.numTlogReplicas = (Integer) verifyProp(props, TLOG_REPLICAS, 0);
this.numPullReplicas = (Integer) verifyProp(props, PULL_REPLICAS, 0);
this.maxShardsPerNode = (Integer) verifyProp(props, MAX_SHARDS_PER_NODE);
+ this.perReplicaState = (Boolean) verifyProp(props, PER_REPLICA_STATE, Boolean.FALSE);
Boolean autoAddReplicas = (Boolean) verifyProp(props, AUTO_ADD_REPLICAS);
this.policy = (String) props.get(Policy.POLICY);
+ ClusterState.getReplicaStatesProvider().get().ifPresent(it -> perReplicaStates = it.getStates());
this.autoAddReplicas = autoAddReplicas == null ? Boolean.FALSE : autoAddReplicas;
Boolean readOnly = (Boolean) verifyProp(props, READ_ONLY);
this.readOnly = readOnly == null ? Boolean.FALSE : readOnly;
@@ -122,6 +134,9 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
}
for (Replica replica : slice.getValue()) {
addNodeNameReplica(replica);
+ if(perReplicaState) {
+ replicaMap.put(replica.getName(), replica);
+ }
}
}
this.activeSlicesArr = activeSlices.values().toArray(new Slice[activeSlices.size()]);
@@ -130,6 +145,31 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
assert name != null && slices != null;
}
+ /**Update our state with a state of a {@link Replica}
+ * Used to create a new Collection State when only a replica is updated
+ */
+ public DocCollection copyWith( PerReplicaStates newPerReplicaStates) {
+ log.debug("collection :{} going to be updated : per-replica state :{} -> {}",
+ name,
+ getChildNodesVersion(), newPerReplicaStates.cversion);
+ if(getChildNodesVersion() == newPerReplicaStates.cversion) return this;
+ Set<String> modifiedReplicas = PerReplicaStates.findModifiedReplicas(newPerReplicaStates, this.perReplicaStates);
+ if(modifiedReplicas.isEmpty()) return this; //nothing is modified
+ Map<String, Slice> modifiedShards = new HashMap<>(getSlicesMap());
+ for (String s : modifiedReplicas) {
+ Replica replica = getReplica(s);
+ if(replica != null) {
+ Replica newReplica = replica.copyWith(newPerReplicaStates.get(s));
+ Slice shard = modifiedShards.get(replica.slice);
+ modifiedShards.put(replica.slice, shard.copyWith(newReplica));
+ }
+ }
+ DocCollection result = new DocCollection(getName(), modifiedShards, propMap, router, znodeVersion, znode);
+ result.perReplicaStates = newPerReplicaStates;
+ return result;
+
+ }
+
private void addNodeNameReplica(Replica replica) {
List<Replica> replicas = nodeNameReplicas.get(replica.getNodeName());
if (replicas == null) {
@@ -165,6 +205,8 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
case AUTO_ADD_REPLICAS:
case READ_ONLY:
return Boolean.parseBoolean(o.toString());
+ case PER_REPLICA_STATE:
+ return Boolean.parseBoolean(o.toString());
case "snitch":
case "rule":
return (List) o;
@@ -178,8 +220,10 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
* @param slices the new set of Slices
* @return the resulting DocCollection
*/
- public DocCollection copyWithSlices(Map<String, Slice> slices){
- return new DocCollection(getName(), slices, propMap, router, znodeVersion,znode);
+ public DocCollection copyWithSlices(Map<String, Slice> slices) {
+ DocCollection result = new DocCollection(getName(), slices, propMap, router, znodeVersion, znode);
+ result.perReplicaStates = perReplicaStates;
+ return result;
}
/**
@@ -253,6 +297,16 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
public int getZNodeVersion(){
return znodeVersion;
}
+ public int getChildNodesVersion() {
+ return perReplicaStates == null ? -1 : perReplicaStates.cversion;
+ }
+
+ public boolean isModified(int dataVersion, int childVersion) {
+ if(dataVersion > znodeVersion) return true;
+ if(childVersion > getChildNodesVersion()) return true;
+ return false;
+
+ }
public int getStateFormat() {
return ZkStateReader.CLUSTER_STATE.equals(znode) ? 1 : 2;
@@ -292,7 +346,9 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
@Override
public String toString() {
- return "DocCollection("+name+"/" + znode + "/" + znodeVersion + ")=" + toJSONString(this);
+ return "DocCollection("+name+"/" + znode + "/" + znodeVersion
+ + (perReplicaStates == null ? "": perReplicaStates.toString())+")="
+ + toJSONString(this);
}
@Override
@@ -304,6 +360,9 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
}
public Replica getReplica(String coreNodeName) {
+ if(perReplicaState) {
+ return replicaMap.get(coreNodeName);
+ }
for (Slice slice : slices.values()) {
Replica replica = slice.getReplica(coreNodeName);
if (replica != null) return replica;
@@ -432,6 +491,15 @@ public class DocCollection extends ZkNodeProps implements Iterable<Slice> {
return policy;
}
+ public boolean isPerReplicaState() {
+ return Boolean.TRUE.equals(perReplicaState);
+ }
+
+ public PerReplicaStates getPerReplicaStates() {
+ return perReplicaStates;
+ }
+
+
public int getExpectedReplicaCount(Replica.Type type, int def) {
Integer result = null;
if (type == Replica.Type.NRT) result = numNrtReplicas;
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/PerReplicaStates.java b/solr/solrj/src/java/org/apache/solr/common/cloud/PerReplicaStates.java
new file mode 100644
index 0000000..9c3e70a
--- /dev/null
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/PerReplicaStates.java
@@ -0,0 +1,587 @@
+/*
+ * 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.common.cloud;
+
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+
+import org.apache.solr.cluster.api.SimpleMap;
+import org.apache.solr.common.MapWriter;
+import org.apache.solr.common.SolrException;
+import org.apache.solr.common.annotation.JsonProperty;
+import org.apache.solr.common.util.ReflectMapWriter;
+import org.apache.solr.common.util.StrUtils;
+import org.apache.solr.common.util.WrappedSimpleMap;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static java.util.Collections.singletonList;
+import static org.apache.solr.common.params.CommonParams.NAME;
+import static org.apache.solr.common.params.CommonParams.VERSION;
+
+/**
+ * This represents the individual replica states in a collection
+ * This is an immutable object. When states are modified, a new instance is constructed
+ */
+public class PerReplicaStates implements ReflectMapWriter {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ public static final char SEPARATOR = ':';
+
+
+ @JsonProperty
+ public final String path;
+
+ @JsonProperty
+ public final int cversion;
+
+ @JsonProperty
+ public final SimpleMap<State> states;
+
+ public PerReplicaStates(String path, int cversion, List<String> states) {
+ this.path = path;
+ this.cversion = cversion;
+ Map<String, State> tmp = new LinkedHashMap<>();
+
+ for (String state : states) {
+ State rs = State.parse(state);
+ if (rs == null) continue;
+ State existing = tmp.get(rs.replica);
+ if (existing == null) {
+ tmp.put(rs.replica, rs);
+ } else {
+ tmp.put(rs.replica, rs.insert(existing));
+ }
+ }
+ this.states = new WrappedSimpleMap<>(tmp);
+
+ }
+
+ /**Get the changed replicas
+ */
+ public static Set<String> findModifiedReplicas(PerReplicaStates old, PerReplicaStates fresh) {
+ Set<String> result = new HashSet<>();
+ if(fresh == null) {
+ old.states.forEachKey(result::add);
+ return result;
+ }
+ old.states.forEachEntry((s, state) -> {
+ //the state is modified or missing
+ if(!Objects.equals(fresh.get(s) , state)) result.add(s);
+ });
+ fresh.states.forEachEntry((s, state) -> { if(old.get(s) == null ) result.add(s);
+ });
+ return result;
+ }
+
+ /**
+ * This is a persist operation with retry if a write fails due to stale state
+ */
+ public static void persist(WriteOps ops, String znode, SolrZkClient zkClient) throws KeeperException, InterruptedException {
+ try {
+ persist(ops.get(), znode, zkClient);
+ } catch (KeeperException.NodeExistsException | KeeperException.NoNodeException e) {
+ //state is stale
+ log.info("stale state for {} . retrying...", znode);
+ List<Op> freshOps = ops.get(PerReplicaStates.fetch(znode, zkClient, null));
+ persist(freshOps, znode, zkClient);
+ log.info("retried for stale state {}, succeeded", znode);
+ }
+ }
+
+ /**
+ * Persist a set of operations to Zookeeper
+ */
+ public static void persist(List<Op> operations, String znode, SolrZkClient zkClient) throws KeeperException, InterruptedException {
+ if (operations == null || operations.isEmpty()) return;
+ log.debug("Per-replica state being persisted for :{}, ops: {}", znode, operations);
+
+ List<org.apache.zookeeper.Op> ops = new ArrayList<>(operations.size());
+ for (Op op : operations) {
+ //the state of the replica is being updated
+ String path = znode + "/" + op.state.asString;
+ List<ACL> acls = zkClient.getZkACLProvider().getACLsToAdd(path);
+ ops.add(op.typ == Op.Type.ADD ?
+ org.apache.zookeeper.Op.create(path, null, acls, CreateMode.PERSISTENT) :
+ org.apache.zookeeper.Op.delete(path, -1));
+ }
+ try {
+ zkClient.multi(ops, true);
+ if (log.isDebugEnabled()) {
+ //nocommit
+ try {
+ Stat stat = zkClient.exists(znode, null, true);
+ log.debug("After update, cversion : {}", stat.getCversion());
+ } catch (Exception e) {
+ }
+
+ }
+ } catch (KeeperException e) {
+ log.error("multi op exception : " + e.getMessage() + zkClient.getChildren(znode, null, true));
+ throw e;
+ }
+
+ }
+
+
+ /**
+ * Fetch the latest {@link PerReplicaStates} . It fetches data after checking the {@link Stat#getCversion()} of state.json.
+ * If this is not modified, the same object is returned
+ */
+ public static PerReplicaStates fetch(String path, SolrZkClient zkClient, PerReplicaStates current) {
+ try {
+ if (current != null) {
+ Stat stat = zkClient.exists(current.path, null, true);
+ if (stat == null) return new PerReplicaStates(path, -1, Collections.emptyList());
+ if (current.cversion == stat.getCversion()) return current;// not modifiedZkStateReaderTest
+ }
+ Stat stat = new Stat();
+ List<String> children = zkClient.getChildren(path, null, stat, true);
+ return new PerReplicaStates(path, stat.getCversion(), Collections.unmodifiableList(children));
+ } catch (KeeperException e) {
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Error fetching per-replica states", e);
+ } catch (InterruptedException e) {
+ SolrZkClient.checkInterrupted(e);
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Thread interrupted when loading per-replica states from " + path, e);
+ }
+ }
+
+
+ private static List<Op> addDeleteStaleNodes(List<Op> ops, State rs) {
+ while (rs != null) {
+ ops.add(new Op(Op.Type.DELETE, rs));
+ rs = rs.duplicate;
+ }
+ return ops;
+ }
+
+ public static String getReplicaName(String s) {
+ int idx = s.indexOf(SEPARATOR);
+ if (idx > 0) {
+ return s.substring(0, idx);
+ }
+ return null;
+ }
+
+ public State get(String replica) {
+ return states.get(replica);
+ }
+
+ public static class Op {
+ public final Type typ;
+ public final State state;
+
+ public Op(Type typ, State replicaState) {
+ this.typ = typ;
+ this.state = replicaState;
+ }
+
+
+ public enum Type {
+ //add a new node
+ ADD,
+ //delete an existing node
+ DELETE
+ }
+
+ @Override
+ public String toString() {
+ return typ.toString() + " : " + state;
+ }
+ }
+
+
+ /**
+ * The state of a replica as stored as a node under /collections/collection-name/state.json/replica-state
+ */
+ public static class State implements MapWriter {
+
+ public final String replica;
+
+ public final Replica.State state;
+
+ public final Boolean isLeader;
+
+ public final int version;
+
+ public final String asString;
+
+ /**
+ * if there are multiple entries for the same replica, e.g: core_node_1:12:A core_node_1:13:D
+ * <p>
+ * the entry with '13' is the latest and the one with '12' is considered a duplicate
+ * <p>
+ * These are unlikely, but possible
+ */
+ final State duplicate;
+
+ private State(String serialized, List<String> pieces) {
+ this.asString = serialized;
+ replica = pieces.get(0);
+ version = Integer.parseInt(pieces.get(1));
+ String encodedStatus = pieces.get(2);
+ this.state = Replica.getState(encodedStatus);
+ isLeader = pieces.size() > 3 && "L".equals(pieces.get(3));
+ duplicate = null;
+ }
+
+ public static State parse(String serialized) {
+ List<String> pieces = StrUtils.splitSmart(serialized, ':');
+ if (pieces.size() < 3) return null;
+ return new State(serialized, pieces);
+
+ }
+
+ public State(String replica, Replica.State state, Boolean isLeader, int version) {
+ this(replica, state, isLeader, version, null);
+ }
+
+ public State(String replica, Replica.State state, Boolean isLeader, int version, State duplicate) {
+ this.replica = replica;
+ this.state = state == null ? Replica.State.ACTIVE : state;
+ this.isLeader = isLeader == null ? Boolean.FALSE : isLeader;
+ this.version = version;
+ asString = serialize();
+ this.duplicate = duplicate;
+ }
+
+ @Override
+ public void writeMap(EntryWriter ew) throws IOException {
+ ew.put(NAME, replica);
+ ew.put(VERSION, version);
+ ew.put(ZkStateReader.STATE_PROP, state.toString());
+ if (isLeader) ew.put(Slice.LEADER, isLeader);
+ ew.putIfNotNull("duplicate", duplicate);
+ }
+
+ private State insert(State duplicate) {
+ assert this.replica.equals(duplicate.replica);
+ if (this.version >= duplicate.version) {
+ if (this.duplicate != null) {
+ duplicate = new State(duplicate.replica, duplicate.state, duplicate.isLeader, duplicate.version, this.duplicate);
+ }
+ return new State(this.replica, this.state, this.isLeader, this.version, duplicate);
+ } else {
+ return duplicate.insert(this);
+ }
+ }
+
+ /**
+ * fetch duplicates entries for this replica
+ */
+ List<State> getDuplicates() {
+ if (duplicate == null) return Collections.emptyList();
+ List<State> result = new ArrayList<>();
+ State current = duplicate;
+ while (current != null) {
+ result.add(current);
+ current = current.duplicate;
+ }
+ return result;
+ }
+
+ private String serialize() {
+ StringBuilder sb = new StringBuilder(replica)
+ .append(":")
+ .append(version)
+ .append(":")
+ .append(state.shortName);
+ if (isLeader) sb.append(":").append("L");
+ return sb.toString();
+ }
+
+
+ @Override
+ public String toString() {
+ return asString;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o instanceof State) {
+ State that = (State) o;
+ return Objects.equals(this.asString, that.asString);
+ }
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ return asString.hashCode();
+ }
+ }
+
+
+ public static abstract class WriteOps {
+ private PerReplicaStates rs;
+ List<Op> ops;
+ private boolean preOp = true;
+
+ /**
+ * state of a replica is changed
+ *
+ * @param newState the new state
+ */
+ public static WriteOps flipState(String replica, Replica.State newState, PerReplicaStates rs) {
+ return new WriteOps() {
+ @Override
+ protected List<Op> refresh(PerReplicaStates rs) {
+ List<Op> ops = new ArrayList<>(2);
+ State existing = rs.get(replica);
+ if (existing == null) {
+ ops.add(new Op(Op.Type.ADD, new State(replica, newState, Boolean.FALSE, 0)));
+ } else {
+ ops.add(new Op(Op.Type.ADD, new State(replica, newState, existing.isLeader, existing.version + 1)));
+ addDeleteStaleNodes(ops, existing);
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("flipState on {}, {} -> {}, ops :{}", rs.path, replica, newState, ops);
+ }
+ return ops;
+ }
+ }.init(rs);
+ }
+
+ public PerReplicaStates getPerReplicaStates() {
+ return rs;
+ }
+
+
+ /**Switch a collection from/to perReplicaState=true
+ */
+ public static WriteOps modifyCollection(DocCollection coll, boolean enable, PerReplicaStates prs) {
+ return new WriteOps() {
+ @Override
+ List<Op> refresh(PerReplicaStates prs) {
+ return enable ? enable(coll) : disable(prs);
+ }
+
+ List<Op> enable(DocCollection coll) {
+ List<Op> result = new ArrayList<>();
+ coll.forEachReplica((s, r) -> result.add(new Op(Op.Type.ADD, new State(r.getName(), r.getState(), r.isLeader(), 0))));
+ return result;
+ }
+
+ List<Op> disable(PerReplicaStates prs) {
+ List<Op> result = new ArrayList<>();
+ prs.states.forEachEntry((s, state) -> result.add(new Op(Op.Type.DELETE, state)));
+ return result;
+ }
+ }.init(prs);
+
+ }
+
+ /**
+ * Flip the leader replica to a new one
+ *
+ * @param allReplicas allReplicas of the shard
+ * @param next next leader
+ */
+ public static WriteOps flipLeader(Set<String> allReplicas, String next, PerReplicaStates rs) {
+ return new WriteOps() {
+
+ @Override
+ protected List<Op> refresh(PerReplicaStates rs) {
+ List<Op> ops = new ArrayList<>(4);
+ if(next != null) {
+ State st = rs.get(next);
+ if (st != null) {
+ if (!st.isLeader) {
+ ops.add(new Op(Op.Type.ADD, new State(st.replica, Replica.State.ACTIVE, Boolean.TRUE, st.version + 1)));
+ ops.add(new Op(Op.Type.DELETE, st));
+ }
+ //else do not do anything , that node is the leader
+ } else {
+ //there is no entry for the new leader.
+ //create one
+ ops.add(new Op(Op.Type.ADD, new State(next, Replica.State.ACTIVE, Boolean.TRUE, 0)));
+ }
+ }
+
+ //now go through all other replicas and unset previous leader
+ for (String r : allReplicas) {
+ State st = rs.get(r);
+ if (st == null) continue;//unlikely
+ if (!Objects.equals(r, next)) {
+ if(st.isLeader) {
+ //some other replica is the leader now. unset
+ ops.add(new Op(Op.Type.ADD, new State(st.replica, st.state, Boolean.FALSE, st.version + 1)));
+ ops.add(new Op(Op.Type.DELETE, st));
+ }
+ }
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("flipLeader on:{}, {} -> {}, ops: {}", rs.path, allReplicas, next, ops);
+ }
+ return ops;
+ }
+
+ }.init(rs);
+ }
+
+ /**
+ * Delete a replica entry from per-replica states
+ *
+ * @param replica name of the replica to be deleted
+ */
+ public static WriteOps deleteReplica(String replica, PerReplicaStates rs) {
+ return new WriteOps() {
+ @Override
+ protected List<Op> refresh(PerReplicaStates rs) {
+ List<Op> result;
+ if (rs == null) {
+ result = Collections.emptyList();
+ } else {
+ State state = rs.get(replica);
+ result = addDeleteStaleNodes(new ArrayList<>(), state);
+ }
+ return result;
+ }
+ }.init(rs);
+ }
+
+ public static WriteOps addReplica(String replica, Replica.State state, boolean isLeader, PerReplicaStates rs) {
+ return new WriteOps() {
+ @Override
+ protected List<Op> refresh(PerReplicaStates rs) {
+ return singletonList(new Op(Op.Type.ADD,
+ new State(replica, state, isLeader, 0)));
+ }
+ }.init(rs);
+ }
+
+ /**
+ * mark a bunch of replicas as DOWN
+ */
+ public static WriteOps downReplicas(List<String> replicas, PerReplicaStates rs) {
+ return new WriteOps() {
+ @Override
+ List<Op> refresh(PerReplicaStates rs) {
+ List<Op> ops = new ArrayList<>();
+ for (String replica : replicas) {
+ State r = rs.get(replica);
+ if (r != null) {
+ if (r.state == Replica.State.DOWN && !r.isLeader) continue;
+ ops.add(new Op(Op.Type.ADD, new State(replica, Replica.State.DOWN, Boolean.FALSE, r.version + 1)));
+ addDeleteStaleNodes(ops, r);
+ } else {
+ ops.add(new Op(Op.Type.ADD, new State(replica, Replica.State.DOWN, Boolean.FALSE, 0)));
+ }
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("for coll: {} down replicas {}, ops {}", rs, replicas, ops);
+ }
+ return ops;
+ }
+ }.init(rs);
+ }
+
+ /**
+ * Just creates and deletes a summy entry so that the {@link Stat#getCversion()} of states.json
+ * is updated
+ */
+ public static WriteOps touchChildren() {
+ WriteOps result = new WriteOps() {
+ @Override
+ List<Op> refresh(PerReplicaStates rs) {
+ List<Op> ops = new ArrayList<>();
+ State st = new State(".dummy." + System.nanoTime(), Replica.State.DOWN, Boolean.FALSE, 0);
+ ops.add(new Op(Op.Type.ADD, st));
+ ops.add(new Op(Op.Type.DELETE, st));
+ if (log.isDebugEnabled()) {
+ log.debug("touchChildren {}", ops);
+ }
+ return ops;
+ }
+ };
+ result.preOp = false;
+ result.ops = result.refresh(null);
+ return result;
+ }
+
+ WriteOps init(PerReplicaStates rs) {
+ if (rs == null) return null;
+ get(rs);
+ return this;
+ }
+
+ public List<Op> get() {
+ return ops;
+ }
+
+ public List<Op> get(PerReplicaStates rs) {
+ ops = refresh(rs);
+ if (ops == null) ops = Collections.emptyList();
+ this.rs = rs;
+ return ops;
+ }
+
+ /**
+ * To be executed before collection state.json is persisted
+ */
+ public boolean isPreOp() {
+ return preOp;
+ }
+
+ /**
+ * if a multi operation fails because the state got modified from behind,
+ * refresh the operation and try again
+ *
+ * @param prs The new state
+ */
+ abstract List<Op> refresh(PerReplicaStates prs);
+
+ @Override
+ public String toString() {
+ return ops.toString();
+ }
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("{").append(path).append("/[").append(cversion).append("]: [");
+ appendStates(sb);
+ return sb.append("]}").toString();
+ }
+
+ private StringBuilder appendStates(StringBuilder sb) {
+ states.forEachEntry(new BiConsumer<String, State>() {
+ int count = 0;
+ @Override
+ public void accept(String s, State state) {
+ if(count++ > 0) sb.append(", ");
+ sb.append(state.asString);
+ for (State d : state.getDuplicates()) sb.append(d.asString);
+ }
+ });
+ return sb;
+ }
+
+}
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/Replica.java b/solr/solrj/src/java/org/apache/solr/common/cloud/Replica.java
index eef3f89..15ecb12 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/Replica.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/Replica.java
@@ -16,6 +16,9 @@
*/
package org.apache.solr.common.cloud;
+import java.lang.invoke.MethodHandles;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
@@ -23,9 +26,11 @@ import java.util.Set;
import org.apache.solr.common.util.Utils;
import static org.apache.solr.common.cloud.ZkStateReader.BASE_URL_PROP;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class Replica extends ZkNodeProps {
-
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
/**
* The replica's state. In general, if the node the replica is hosted on is
* not under {@code /live_nodes} in ZK, the replica's state should be
@@ -43,7 +48,7 @@ public class Replica extends ZkNodeProps {
* {@link ClusterState#liveNodesContain(String)}).
* </p>
*/
- ACTIVE,
+ ACTIVE("A"),
/**
* The first state before {@link State#RECOVERING}. A node in this state
@@ -54,13 +59,13 @@ public class Replica extends ZkNodeProps {
* should not be relied on.
* </p>
*/
- DOWN,
+ DOWN("D"),
/**
* The node is recovering from the leader. This might involve peer-sync,
* full replication or finding out things are already in sync.
*/
- RECOVERING,
+ RECOVERING("R"),
/**
* Recovery attempts have not worked, something is not right.
@@ -70,8 +75,16 @@ public class Replica extends ZkNodeProps {
* cluster and it's state should be discarded.
* </p>
*/
- RECOVERY_FAILED;
-
+ RECOVERY_FAILED("F");
+
+ /**short name for a state. Used to encode this in the state node see {@link PerReplicaStates.State}
+ */
+ public final String shortName;
+
+ State(String c) {
+ this.shortName = c;
+ }
+
@Override
public String toString() {
return super.toString().toLowerCase(Locale.ROOT);
@@ -79,7 +92,7 @@ public class Replica extends ZkNodeProps {
/** Converts the state string to a State instance. */
public static State getState(String stateStr) {
- return stateStr == null ? null : State.valueOf(stateStr.toUpperCase(Locale.ROOT));
+ return stateStr == null ? null : Replica.State.valueOf(stateStr.toUpperCase(Locale.ROOT));
}
}
@@ -114,6 +127,7 @@ public class Replica extends ZkNodeProps {
private final State state;
private final Type type;
public final String slice, collection;
+ private PerReplicaStates.State replicaState;
public Replica(String name, Map<String,Object> propMap, String collection, String slice) {
super(propMap);
@@ -129,11 +143,23 @@ public class Replica extends ZkNodeProps {
Objects.requireNonNull(this.nodeName, "'node_name' must not be null");
Objects.requireNonNull(this.core, "'core' must not be null");
Objects.requireNonNull(this.type, "'type' must not be null");
- if (propMap.get(ZkStateReader.STATE_PROP) != null) {
- this.state = State.getState((String) propMap.get(ZkStateReader.STATE_PROP));
+ ClusterState.getReplicaStatesProvider().get().ifPresent(it -> {
+ log.debug("A replica {} state fetched from per-replica state", name);
+ replicaState = it.getStates().get(name);
+ if(replicaState!= null) {
+ propMap.put(ZkStateReader.STATE_PROP, replicaState.state.toString().toLowerCase(Locale.ROOT));
+ if (replicaState.isLeader) propMap.put(Slice.LEADER, "true");
+ }
+ }) ;
+ if (replicaState == null) {
+ if (propMap.get(ZkStateReader.STATE_PROP) != null) {
+ this.state = Replica.State.getState((String) propMap.get(ZkStateReader.STATE_PROP));
+ } else {
+ this.state = Replica.State.ACTIVE; //Default to ACTIVE
+ propMap.put(ZkStateReader.STATE_PROP, state.toString());
+ }
} else {
- this.state = State.ACTIVE; //Default to ACTIVE
- propMap.put(ZkStateReader.STATE_PROP, state.toString());
+ this.state = replicaState.state;
}
propMap.put(BASE_URL_PROP, UrlScheme.INSTANCE.getBaseUrlForNodeName(this.nodeName));
}
@@ -190,7 +216,7 @@ public class Replica extends ZkNodeProps {
}
public boolean isActive(Set<String> liveNodes) {
- return this.nodeName != null && liveNodes.contains(this.nodeName) && this.state == State.ACTIVE;
+ return this.nodeName != null && liveNodes.contains(this.nodeName) && this.state == Replica.State.ACTIVE;
}
public Type getType() {
@@ -208,6 +234,40 @@ public class Replica extends ZkNodeProps {
return propertyValue;
}
+ public Replica copyWith(PerReplicaStates.State state) {
+ log.debug("A replica is updated with new state : {}", state);
+ Map<String, Object> props = new LinkedHashMap<>(propMap);
+ if (state == null) {
+ props.put(ZkStateReader.STATE_PROP, State.DOWN.toString());
+ props.remove(Slice.LEADER);
+ } else {
+ props.put(ZkStateReader.STATE_PROP, state.state.toString());
+ if (state.isLeader) props.put(Slice.LEADER, "true");
+ }
+ Replica r = new Replica(name, props, collection, slice);
+ r.replicaState = state;
+ return r;
+ }
+
+ public PerReplicaStates.State getReplicaState() {
+ return replicaState;
+ }
+
+ private static final Map<String, State> STATES = new HashMap<>();
+ static {
+ STATES.put(Replica.State.ACTIVE.shortName, Replica.State.ACTIVE);
+ STATES.put(Replica.State.DOWN.shortName, Replica.State.DOWN);
+ STATES.put(Replica.State.RECOVERING.shortName, Replica.State.RECOVERING);
+ STATES.put(Replica.State.RECOVERY_FAILED.shortName, Replica.State.RECOVERY_FAILED);
+ }
+ public static State getState(String c) {
+ return STATES.get(c);
+ }
+
+ public boolean isLeader() {
+ if(replicaState != null) return replicaState.isLeader;
+ return getStr(Slice.LEADER) != null;
+ }
@Override
public String toString() {
return name + ':' + Utils.toJSONString(propMap); // small enough, keep it on one line (i.e. no indent)
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/Slice.java b/solr/solrj/src/java/org/apache/solr/common/cloud/Slice.java
index 4378ef7..802299a 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/Slice.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/Slice.java
@@ -16,6 +16,7 @@
*/
package org.apache.solr.common.cloud;
+import java.lang.invoke.MethodHandles;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
@@ -25,11 +26,14 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Set;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import org.apache.solr.common.cloud.Replica.Type;
import org.noggit.JSONWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import static org.apache.solr.common.util.Utils.toJSONString;
@@ -37,6 +41,8 @@ import static org.apache.solr.common.util.Utils.toJSONString;
* A Slice contains immutable information about a logical shard (all replicas that share the same shard id).
*/
public class Slice extends ZkNodeProps implements Iterable<Replica> {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
public final String collection;
/** Loads multiple slices into a Map from a generic Map that probably came from deserialized JSON. */
@@ -61,6 +67,14 @@ public class Slice extends ZkNodeProps implements Iterable<Replica> {
return replicas.values().iterator();
}
+ /**Make a copy with a modified replica
+ */
+ public Slice copyWith(Replica modified) {
+ log.debug("modified replica : {}", modified);
+ Map<String, Replica> replicasCopy = new LinkedHashMap<>(replicas);
+ replicasCopy.put(modified.getName(), modified);
+ return new Slice(name, replicasCopy, propMap, collection);
+ }
/** The slice's state. */
public enum State {
@@ -107,7 +121,7 @@ public class Slice extends ZkNodeProps implements Iterable<Replica> {
/** Converts the state string to a State instance. */
public static State getState(String stateStr) {
- return State.valueOf(stateStr.toUpperCase(Locale.ROOT));
+ return Slice.State.valueOf(stateStr.toUpperCase(Locale.ROOT));
}
}
@@ -138,9 +152,9 @@ public class Slice extends ZkNodeProps implements Iterable<Replica> {
Object rangeObj = propMap.get(RANGE);
if (propMap.get(ZkStateReader.STATE_PROP) != null) {
- this.state = State.getState((String) propMap.get(ZkStateReader.STATE_PROP));
+ this.state = Slice.State.getState((String) propMap.get(ZkStateReader.STATE_PROP));
} else {
- this.state = State.ACTIVE; //Default to ACTIVE
+ this.state = Slice.State.ACTIVE; //Default to ACTIVE
propMap.put(ZkStateReader.STATE_PROP, state.toString());
}
DocRouter.Range tmpRange = null;
@@ -210,7 +224,7 @@ public class Slice extends ZkNodeProps implements Iterable<Replica> {
private Replica findLeader() {
for (Replica replica : replicas.values()) {
- if (replica.getStr(LEADER) != null) {
+ if (replica.isLeader()) {
assert replica.getType() == Type.TLOG || replica.getType() == Type.NRT: "Pull replica should not become leader!";
return replica;
}
@@ -235,6 +249,10 @@ public class Slice extends ZkNodeProps implements Iterable<Replica> {
return replicas.values();
}
+ public Set<String> getReplicaNames() {
+ return Collections.unmodifiableSet(replicas.keySet());
+ }
+
/**
* Gets all replicas that match a predicate
*/
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java b/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java
index 9b1101e..6943fc5 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java
@@ -331,6 +331,18 @@ public class SolrZkClient implements Closeable {
}
/**
+ * Returns children of the node at the path
+ */
+ public List<String> getChildren(final String path, final Watcher watcher,Stat stat, boolean retryOnConnLoss)
+ throws KeeperException, InterruptedException {
+ if (retryOnConnLoss) {
+ return zkCmdExecutor.retryOperation(() -> keeper.getChildren(path, wrapWatcher(watcher) , stat));
+ } else {
+ return keeper.getChildren(path, wrapWatcher(watcher), stat);
+ }
+ }
+
+ /**
* Returns node's data
*/
public byte[] getData(final String path, final Watcher watcher, final Stat stat, boolean retryOnConnLoss)
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
index fef21b6..b1d27ac 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
@@ -142,6 +142,9 @@ public class ZkStateReader implements SolrCloseable {
public static final String LEGACY_CLOUD = "legacyCloud";
public static final String SAMPLE_PERCENTAGE = "samplePercentage";
+ //nocommit
+ public String nodeName;
+
/**
* @deprecated use {@link org.apache.solr.common.params.CollectionAdminParams#DEFAULTS} instead.
*/
@@ -264,6 +267,20 @@ public class ZkStateReader implements SolrCloseable {
return new AutoScalingConfig(map);
}
+ //nocommit
+ public void debugCollectionState(String collectionName) {
+ try {
+ String path = getCollectionPath(collectionName);
+ byte[] bytes = zkClient.getData(path, null, null, true);
+ log.info("{}/state.json: {}", collectionName, new String(bytes, StandardCharsets.UTF_8));
+ log.info("{}/state.json/: {}", collectionName, zkClient.getChildren(path, null, true));
+ } catch (Exception e) {
+
+ log.error("", e);
+ }
+
+ }
+
private static class CollectionWatch<T> {
int coreRefCount = 0;
@@ -396,7 +413,7 @@ public class ZkStateReader implements SolrCloseable {
Collection<String> safeCopy = new ArrayList<>(watchedCollectionStates.keySet());
Set<String> updatedCollections = new HashSet<>();
for (String coll : safeCopy) {
- DocCollection newState = fetchCollectionState(coll, null);
+ DocCollection newState = fetchCollectionState(coll, null, null);
if (updateWatchedCollection(coll, newState)) {
updatedCollections.add(coll);
}
@@ -444,7 +461,7 @@ public class ZkStateReader implements SolrCloseable {
} else if (watchedCollectionStates.containsKey(collection)) {
// Exists as a watched collection, force a refresh.
log.debug("Forcing refresh of watched collection state for {}", collection);
- DocCollection newState = fetchCollectionState(collection, null);
+ DocCollection newState = fetchCollectionState(collection, null, null);
if (updateWatchedCollection(collection, newState)) {
constructState(Collections.singleton(collection));
}
@@ -774,7 +791,7 @@ public class ZkStateReader implements SolrCloseable {
private class LazyCollectionRef extends ClusterState.CollectionRef {
private final String collName;
- private long lastUpdateTime;
+ private volatile long lastUpdateTime;
private DocCollection cachedDocCollection;
public LazyCollectionRef(String collName) {
@@ -794,7 +811,7 @@ public class ZkStateReader implements SolrCloseable {
exists = zkClient.exists(getCollectionPath(collName), null, true);
} catch (Exception e) {
}
- if (exists != null && exists.getVersion() == cachedDocCollection.getZNodeVersion()) {
+ if (exists != null && !cachedDocCollection.isModified(exists.getVersion(), exists.getCversion())) {
shouldFetch = false;
}
}
@@ -972,14 +989,18 @@ public class ZkStateReader implements SolrCloseable {
* Get shard leader properties, with retry if none exist.
*/
public Replica getLeaderRetry(String collection, String shard, int timeout) throws InterruptedException {
+ log.debug("getLeaderRetry:@{} {}/{}", System.currentTimeMillis(), collection, shard);
+ AtomicReference<DocCollection> coll = new AtomicReference<>();
AtomicReference<Replica> leader = new AtomicReference<>();
try {
waitForState(collection, timeout, TimeUnit.MILLISECONDS, (n, c) -> {
if (c == null)
return false;
+ coll.set(c);
Replica l = getLeader(n, c, shard);
if (l != null) {
+ log.debug("leader found for {}/{} to be {}", collection, shard, l);
leader.set(l);
return true;
}
@@ -1314,9 +1335,11 @@ public class ZkStateReader implements SolrCloseable {
*/
class StateWatcher implements Watcher {
private final String coll;
+ private final String collectionPath;
StateWatcher(String coll) {
this.coll = coll;
+ collectionPath = getCollectionPath(coll);
}
@Override
@@ -1338,18 +1361,29 @@ public class ZkStateReader implements SolrCloseable {
event, coll, liveNodes.size());
}
- refreshAndWatch();
+ refreshAndWatch(event.getType());
}
+ public void refreshAndWatch() {
+ refreshAndWatch(null);
+ }
/**
* Refresh collection state from ZK and leave a watch for future changes.
* As a side effect, updates {@link #clusterState} and {@link #watchedCollectionStates}
* with the results of the refresh.
*/
- public void refreshAndWatch() {
+ public void refreshAndWatch(EventType eventType) {
try {
- DocCollection newState = fetchCollectionState(coll, this);
+ if (eventType == null || eventType == EventType.NodeChildrenChanged) {
+ refreshAndWatchChildren();
+ if (eventType == EventType.NodeChildrenChanged) {
+ //only per-replica states modified. return
+ return;
+ }
+ }
+
+ DocCollection newState = fetchCollectionState(coll, this, collectionPath);
updateWatchedCollection(coll, newState);
synchronized (getUpdateLock()) {
constructState(Collections.singleton(coll));
@@ -1365,6 +1399,32 @@ public class ZkStateReader implements SolrCloseable {
log.error("Unwatched collection: [{}]", coll, e);
}
}
+
+ private void refreshAndWatchChildren() throws KeeperException, InterruptedException {
+ Stat stat = new Stat();
+ List<String> replicaStates = null;
+ try {
+ replicaStates = zkClient.getChildren(collectionPath, this, stat, true);
+ PerReplicaStates newStates = new PerReplicaStates(collectionPath, stat.getCversion(), replicaStates);
+ DocCollection oldState = watchedCollectionStates.get(coll);
+ DocCollection newState = null;
+ if (oldState != null) {
+ newState = oldState.copyWith(newStates);
+ } else {
+ newState = fetchCollectionState(coll, null, collectionPath);
+ }
+ updateWatchedCollection(coll, newState);
+ synchronized (getUpdateLock()) {
+ constructState(Collections.singleton(coll));
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("node : {} updated per-replica states changed for: {}, ver: {} , new vals: {}", nodeName, coll, stat.getCversion(), replicaStates);
+ }
+
+ } catch (NoNodeException e) {
+ log.info("{} is deleted, stop watching children", collectionPath);
+ }
+ }
}
/**
@@ -1567,7 +1627,7 @@ public class ZkStateReader implements SolrCloseable {
public static DocCollection getCollectionLive(ZkStateReader zkStateReader, String coll) {
try {
- return zkStateReader.fetchCollectionState(coll, null);
+ return zkStateReader.fetchCollectionState(coll, null, null);
} catch (KeeperException e) {
throw new SolrException(ErrorCode.BAD_REQUEST, "Could not load collection from ZK: " + coll, e);
} catch (InterruptedException e) {
@@ -1576,14 +1636,24 @@ public class ZkStateReader implements SolrCloseable {
}
}
- private DocCollection fetchCollectionState(String coll, Watcher watcher) throws KeeperException, InterruptedException {
- String collectionPath = getCollectionPath(coll);
+ public DocCollection fetchCollectionState(String coll, Watcher watcher, String path) throws KeeperException, InterruptedException {
+ String collectionPath = path == null ? getCollectionPath(coll) : path;
while (true) {
+ ClusterState.initReplicaStateProvider(() -> {
+ try {
+ PerReplicaStates replicaStates = getReplicaStates(new PerReplicaStates(collectionPath, 0, Collections.emptyList()));
+ log.info("per-replica-state ver: {} fetched for initializing {} ", replicaStates.cversion, collectionPath);
+ return replicaStates;
+ } catch (Exception e) {
+ //TODO
+ throw new RuntimeException(e);
+ }
+ });
try {
Stat stat = new Stat();
byte[] data = zkClient.getData(collectionPath, watcher, stat, true);
ClusterState state = ClusterState.load(stat.getVersion(), data,
- Collections.<String>emptySet(), collectionPath);
+ Collections.emptySet(), collectionPath);
ClusterState.CollectionRef collectionRef = state.getCollectionStates().get(coll);
return collectionRef == null ? null : collectionRef.get();
} catch (KeeperException.NoNodeException e) {
@@ -1597,6 +1667,8 @@ public class ZkStateReader implements SolrCloseable {
}
}
return null;
+ } finally {
+ ClusterState.clearReplicaStateProvider();
}
}
}
@@ -1715,6 +1787,7 @@ public class ZkStateReader implements SolrCloseable {
v = new CollectionWatch<>();
watchSet.set(true);
}
+ log.info("already watching , added to stateWatchers");
v.stateWatchers.add(stateWatcher);
return v;
});
@@ -1724,11 +1797,26 @@ public class ZkStateReader implements SolrCloseable {
}
DocCollection state = clusterState.getCollectionOrNull(collection);
+ state = updatePerReplicaState(state);
if (stateWatcher.onStateChanged(state) == true) {
removeDocCollectionWatcher(collection, stateWatcher);
}
}
+ private DocCollection updatePerReplicaState(DocCollection c) {
+ if (c == null || !c.isPerReplicaState()) return c;
+ PerReplicaStates current = c.getPerReplicaStates();
+ PerReplicaStates newPrs = PerReplicaStates.fetch(c.getZNode(), zkClient, current);
+ if (newPrs != current) {
+ log.debug("just-in-time update for a fresh per-replica-state {}", c.getName());
+ DocCollection modifiedColl = c.copyWith(newPrs);
+ updateWatchedCollection(c.getName(), modifiedColl);
+ return modifiedColl;
+ } else {
+ return c;
+ }
+ }
+
/**
* Block until a CollectionStatePredicate returns true, or the wait times out
*
@@ -1759,12 +1847,18 @@ public class ZkStateReader implements SolrCloseable {
throw new AlreadyClosedException();
}
+ long waitStartTime = System.currentTimeMillis();
+ log.debug("{} is waiting for collectionState at : {}", nodeName, waitStartTime);
final CountDownLatch latch = new CountDownLatch(1);
waitLatches.add(latch);
AtomicReference<DocCollection> docCollection = new AtomicReference<>();
CollectionStateWatcher watcher = (n, c) -> {
docCollection.set(c);
boolean matches = predicate.matches(n, c);
+ if (!matches) {
+ log.info(" CollectionStatePredicate failed for {}, after {} secs, cversion : {}", collection, (System.currentTimeMillis() - waitStartTime),
+ (c == null || c.getPerReplicaStates() == null ? "-1" : c.getPerReplicaStates()));
+ }
if (matches)
latch.countDown();
@@ -1962,7 +2056,9 @@ public class ZkStateReader implements SolrCloseable {
break;
}
} else {
- if (oldState.getZNodeVersion() >= newState.getZNodeVersion()) {
+ int oldCVersion = oldState.getPerReplicaStates() == null ? -1 : oldState.getPerReplicaStates().cversion;
+ int newCVersion = newState.getPerReplicaStates() == null ? -1 : newState.getPerReplicaStates().cversion;
+ if (oldState.getZNodeVersion() >= newState.getZNodeVersion() && oldCVersion >= newCVersion) {
// no change to state, but we might have been triggered by the addition of a
// state watcher, so run notifications
updated = true;
@@ -2356,7 +2452,17 @@ public class ZkStateReader implements SolrCloseable {
}
}
+ public PerReplicaStates getReplicaStates(String path) throws KeeperException, InterruptedException {
+ return PerReplicaStates.fetch(path, zkClient, null);
+
+ }
+
+ public PerReplicaStates getReplicaStates(PerReplicaStates current) throws KeeperException, InterruptedException {
+ return PerReplicaStates.fetch(current.path, zkClient, current);
+ }
+
public DocCollection getCollection(String collection) {
- return clusterState.getCollectionOrNull(collection);
+ return clusterState == null ? null : clusterState.getCollectionOrNull(collection);
}
+
}
diff --git a/solr/solrj/src/test/org/apache/solr/client/ref_guide_examples/IndexingNestedDocuments.java b/solr/solrj/src/test/org/apache/solr/client/ref_guide_examples/IndexingNestedDocuments.java
index 9d22119..566819b 100644
--- a/solr/solrj/src/test/org/apache/solr/client/ref_guide_examples/IndexingNestedDocuments.java
+++ b/solr/solrj/src/test/org/apache/solr/client/ref_guide_examples/IndexingNestedDocuments.java
@@ -70,7 +70,9 @@ public class IndexingNestedDocuments extends SolrCloudTestCase {
*/
public void testIndexingAnonKids() throws Exception {
final String collection = "test_anon";
- CollectionAdminRequest.createCollection(collection, ANON_KIDS_CONFIG, 1, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(collection, ANON_KIDS_CONFIG, 1, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.getSolrClient().setDefaultCollection(collection);
//
diff --git a/solr/solrj/src/test/org/apache/solr/client/ref_guide_examples/JsonRequestApiHeatmapFacetingTest.java b/solr/solrj/src/test/org/apache/solr/client/ref_guide_examples/JsonRequestApiHeatmapFacetingTest.java
index 4698c7e..3915a00 100644
--- a/solr/solrj/src/test/org/apache/solr/client/ref_guide_examples/JsonRequestApiHeatmapFacetingTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/ref_guide_examples/JsonRequestApiHeatmapFacetingTest.java
@@ -57,7 +57,9 @@ public class JsonRequestApiHeatmapFacetingTest extends SolrCloudTestCase {
final List<String> solrUrls = new ArrayList<>();
solrUrls.add(cluster.getJettySolrRunner(0).getBaseUrl().toString());
- CollectionAdminRequest.createCollection(COLLECTION_NAME, CONFIG_NAME, 1, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, CONFIG_NAME, 1, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
indexSpatialData();
}
diff --git a/solr/solrj/src/test/org/apache/solr/client/ref_guide_examples/JsonRequestApiTest.java b/solr/solrj/src/test/org/apache/solr/client/ref_guide_examples/JsonRequestApiTest.java
index 86a0d8f..51679be 100644
--- a/solr/solrj/src/test/org/apache/solr/client/ref_guide_examples/JsonRequestApiTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/ref_guide_examples/JsonRequestApiTest.java
@@ -64,7 +64,8 @@ public class JsonRequestApiTest extends SolrCloudTestCase {
final List<String> solrUrls = new ArrayList<>();
solrUrls.add(cluster.getJettySolrRunner(0).getBaseUrl().toString());
- CollectionAdminRequest.createCollection(COLLECTION_NAME, CONFIG_NAME, 1, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, CONFIG_NAME, 1, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
ContentStreamUpdateRequest up = new ContentStreamUpdateRequest("/update");
up.setParam("collection", COLLECTION_NAME);
diff --git a/solr/solrj/src/test/org/apache/solr/client/ref_guide_examples/UsingSolrJRefGuideExamplesTest.java b/solr/solrj/src/test/org/apache/solr/client/ref_guide_examples/UsingSolrJRefGuideExamplesTest.java
index 04776cc..acb6d3e 100644
--- a/solr/solrj/src/test/org/apache/solr/client/ref_guide_examples/UsingSolrJRefGuideExamplesTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/ref_guide_examples/UsingSolrJRefGuideExamplesTest.java
@@ -69,6 +69,7 @@ public class UsingSolrJRefGuideExamplesTest extends SolrCloudTestCase {
.configure();
CollectionAdminResponse response = CollectionAdminRequest.createCollection("techproducts", "conf", 1, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.process(cluster.getSolrClient());
cluster.waitForActiveCollection("techproducts", 1, 1);
}
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/impl/CloudSolrClientTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/impl/CloudSolrClientTest.java
index 9e38a12..2cf27a9 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/impl/CloudSolrClientTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/impl/CloudSolrClientTest.java
@@ -60,6 +60,7 @@ 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.DocRouter;
+import org.apache.solr.common.cloud.PerReplicaStates;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.params.CommonParams;
@@ -71,6 +72,7 @@ import org.apache.solr.common.util.SimpleOrderedMap;
import org.apache.solr.handler.admin.CollectionsHandler;
import org.apache.solr.handler.admin.ConfigSetsHandler;
import org.apache.solr.handler.admin.CoreAdminHandler;
+import org.apache.solr.util.LogLevel;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
@@ -85,10 +87,11 @@ import org.slf4j.LoggerFactory;
* This test would be faster if we simulated the zk state instead.
*/
@Slow
+@LogLevel("org.apache.solr.cloud.Overseer=INFO;org.apache.solr.common.cloud=INFO;org.apache.solr.cloud.api.collections=INFO;org.apache.solr.cloud.overseer=INFO")
public class CloudSolrClientTest extends SolrCloudTestCase {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
+
private static final String COLLECTION = "collection1";
private static final String COLLECTION2 = "2nd_collection";
@@ -111,8 +114,8 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
httpBasedCloudSolrClient = new CloudSolrClient.Builder(solrUrls).build();
}
-
- @After
+
+ @After
public void tearDown() throws Exception {
if (httpBasedCloudSolrClient != null) {
try {
@@ -121,7 +124,7 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
throw new RuntimeException(e);
}
}
-
+
shutdownCluster();
super.tearDown();
}
@@ -135,15 +138,17 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
* Randomly return the cluster's ZK based CSC, or HttpClusterProvider based CSC.
*/
private CloudSolrClient getRandomClient() {
- return random().nextBoolean()? cluster.getSolrClient(): httpBasedCloudSolrClient;
+ return random().nextBoolean() ? cluster.getSolrClient() : httpBasedCloudSolrClient;
}
@Test
public void testParallelUpdateQTime() throws Exception {
- CollectionAdminRequest.createCollection(COLLECTION, "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(COLLECTION, "conf", 2, 1)
+ .setPerReplicaState(USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection(COLLECTION, 2, 2);
UpdateRequest req = new UpdateRequest();
- for (int i=0; i<10; i++) {
+ for (int i = 0; i < 10; i++) {
SolrInputDocument doc = new SolrInputDocument();
doc.addField("id", String.valueOf(TestUtil.nextInt(random(), 1000, 1100)));
req.add(doc);
@@ -157,9 +162,10 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
public void testOverwriteOption() throws Exception {
CollectionAdminRequest.createCollection("overwrite", "conf", 1, 1)
+ .setPerReplicaState(USE_PER_REPLICA_STATE)
.processAndWait(cluster.getSolrClient(), TIMEOUT);
cluster.waitForActiveCollection("overwrite", 1, 1);
-
+
new UpdateRequest()
.add("id", "0", "a_t", "hello1")
.add("id", "0", "a_t", "hello2")
@@ -172,7 +178,7 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
.add(new SolrInputDocument(id, "1", "a_t", "hello1"), /* overwrite = */ false)
.add(new SolrInputDocument(id, "1", "a_t", "hello2"), false)
.commit(cluster.getSolrClient(), "overwrite");
-
+
resp = getRandomClient().query("overwrite", new SolrQuery("*:*"));
assertEquals("There should be 3 documents because there should be two id=1 docs due to overwrite=false", 3, resp.getResults().getNumFound());
@@ -180,10 +186,14 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
@Test
public void testAliasHandling() throws Exception {
- CollectionAdminRequest.createCollection(COLLECTION, "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(COLLECTION, "conf", 2, 1)
+ .setPerReplicaState(USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection(COLLECTION, 2, 2);
- CollectionAdminRequest.createCollection(COLLECTION2, "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(COLLECTION2, "conf", 2, 1)
+ .setPerReplicaState(USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection(COLLECTION2, 2, 2);
CloudSolrClient client = getRandomClient();
@@ -228,25 +238,27 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
@Test
public void testRouting() throws Exception {
- CollectionAdminRequest.createCollection("routing_collection", "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection("routing_collection", "conf", 2, 1)
+ .setPerReplicaState(USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection("routing_collection", 2, 2);
-
+
AbstractUpdateRequest request = new UpdateRequest()
.add(id, "0", "a_t", "hello1")
.add(id, "2", "a_t", "hello2")
.setAction(AbstractUpdateRequest.ACTION.COMMIT, true, true);
-
+
// Test single threaded routed updates for UpdateRequest
NamedList<Object> response = getRandomClient().request(request, "routing_collection");
if (getRandomClient().isDirectUpdatesToLeadersOnly()) {
checkSingleServer(response);
}
CloudSolrClient.RouteResponse rr = (CloudSolrClient.RouteResponse) response;
- Map<String,LBHttpSolrClient.Req> routes = rr.getRoutes();
- Iterator<Map.Entry<String,LBHttpSolrClient.Req>> it = routes.entrySet()
+ Map<String, LBHttpSolrClient.Req> routes = rr.getRoutes();
+ Iterator<Map.Entry<String, LBHttpSolrClient.Req>> it = routes.entrySet()
.iterator();
while (it.hasNext()) {
- Map.Entry<String,LBHttpSolrClient.Req> entry = it.next();
+ Map.Entry<String, LBHttpSolrClient.Req> entry = it.next();
String url = entry.getKey();
UpdateRequest updateRequest = (UpdateRequest) entry.getValue()
.getRequest();
@@ -262,9 +274,9 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
assertTrue(docList.getNumFound() == 1);
}
}
-
+
// Test the deleteById routing for UpdateRequest
-
+
final UpdateResponse uResponse = new UpdateRequest()
.deleteById("0")
.deleteById("2")
@@ -276,7 +288,7 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
QueryResponse qResponse = getRandomClient().query("routing_collection", new SolrQuery("*:*"));
SolrDocumentList docs = qResponse.getResults();
assertEquals(0, docs.getNumFound());
-
+
// Test Multi-Threaded routed updates for UpdateRequest
try (CloudSolrClient threadedClient = new CloudSolrClientBuilder
(Collections.singletonList(cluster.getZkServer().getZkAddress()), Optional.empty())
@@ -292,7 +304,7 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
it = routes.entrySet()
.iterator();
while (it.hasNext()) {
- Map.Entry<String,LBHttpSolrClient.Req> entry = it.next();
+ Map.Entry<String, LBHttpSolrClient.Req> entry = it.next();
String url = entry.getKey();
UpdateRequest updateRequest = (UpdateRequest) entry.getValue()
.getRequest();
@@ -334,7 +346,7 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
}
assertTrue("expected urls is not fewer than all urls! expected=" + expectedBaseURLs
- + "; all=" + requestCountsMap.keySet(),
+ + "; all=" + requestCountsMap.keySet(),
expectedBaseURLs.size() < requestCountsMap.size());
// Calculate a number of shard keys that route to the same shard.
@@ -344,7 +356,7 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
} else {
n = random().nextInt(9) + 2;
}
-
+
List<String> sameShardRoutes = Lists.newArrayList();
sameShardRoutes.add("0");
for (int i = 1; i < n; i++) {
@@ -412,6 +424,7 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
// all its cores on the same node.
// Hence the below configuration for our collection
CollectionAdminRequest.createCollection(collectionName, "conf", liveNodes, liveNodes)
+ .setPerReplicaState(USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(liveNodes * liveNodes)
.processAndWait(cluster.getSolrClient(), TIMEOUT);
cluster.waitForActiveCollection(collectionName, liveNodes, liveNodes * liveNodes);
@@ -431,8 +444,7 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
private void queryWithShardsPreferenceRules(CloudSolrClient cloudClient,
boolean useShardsPreference,
String collectionName)
- throws Exception
- {
+ throws Exception {
SolrQuery qRequest = new SolrQuery("*:*");
ModifiableSolrParams qParams = new ModifiableSolrParams();
@@ -451,18 +463,18 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
QueryResponse qResponse = cloudClient.query(collectionName, qRequest);
Object shardsInfo = qResponse.getResponse().get(ShardParams.SHARDS_INFO);
- assertNotNull("Unable to obtain "+ShardParams.SHARDS_INFO, shardsInfo);
+ assertNotNull("Unable to obtain " + ShardParams.SHARDS_INFO, shardsInfo);
// Iterate over shards-info and check what cores responded
- SimpleOrderedMap<?> shardsInfoMap = (SimpleOrderedMap<?>)shardsInfo;
+ SimpleOrderedMap<?> shardsInfoMap = (SimpleOrderedMap<?>) shardsInfo;
@SuppressWarnings({"unchecked"})
Iterator<Map.Entry<String, ?>> itr = shardsInfoMap.asMap(100).entrySet().iterator();
List<String> shardAddresses = new ArrayList<String>();
while (itr.hasNext()) {
Map.Entry<String, ?> e = itr.next();
- assertTrue("Did not find map-type value in "+ShardParams.SHARDS_INFO, e.getValue() instanceof Map);
- String shardAddress = (String)((Map)e.getValue()).get("shardAddress");
- assertNotNull(ShardParams.SHARDS_INFO+" did not return 'shardAddress' parameter", shardAddress);
+ assertTrue("Did not find map-type value in " + ShardParams.SHARDS_INFO, e.getValue() instanceof Map);
+ String shardAddress = (String) ((Map) e.getValue()).get("shardAddress");
+ assertNotNull(ShardParams.SHARDS_INFO + " did not return 'shardAddress' parameter", shardAddress);
shardAddresses.add(shardAddress);
}
if (log.isInfoEnabled()) {
@@ -471,14 +483,14 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
// Make sure the distributed queries were directed to a single node only
Set<Integer> ports = new HashSet<Integer>();
- for (String shardAddr: shardAddresses) {
- URL url = new URL (shardAddr);
+ for (String shardAddr : shardAddresses) {
+ URL url = new URL(shardAddr);
ports.add(url.getPort());
}
// This assertion would hold true as long as every shard has a core on each node
- assertTrue ("Response was not received from shards on a single node",
- shardAddresses.size() > 1 && ports.size()==1);
+ assertTrue("Response was not received from shards on a single node",
+ shardAddresses.size() > 1 && ports.size() == 1);
}
/**
@@ -491,7 +503,8 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
int liveNodes = cluster.getJettySolrRunners().size();
// For testing replica.type, we want to have all replica types available for the collection
- CollectionAdminRequest.createCollection(collectionName, "conf", 1, liveNodes/3, liveNodes/3, liveNodes/3)
+ CollectionAdminRequest.createCollection(collectionName, "conf", 1, liveNodes / 3, liveNodes / 3, liveNodes / 3)
+ .setPerReplicaState(USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(liveNodes)
.processAndWait(cluster.getSolrClient(), TIMEOUT);
cluster.waitForActiveCollection(collectionName, 1, liveNodes);
@@ -510,10 +523,9 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
}
private void queryReplicaType(CloudSolrClient cloudClient,
- Replica.Type typeToQuery,
- String collectionName)
- throws Exception
- {
+ Replica.Type typeToQuery,
+ String collectionName)
+ throws Exception {
SolrQuery qRequest = new SolrQuery("*:*");
ModifiableSolrParams qParams = new ModifiableSolrParams();
@@ -526,21 +538,21 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
QueryResponse qResponse = cloudClient.query(collectionName, qRequest);
Object shardsInfo = qResponse.getResponse().get(ShardParams.SHARDS_INFO);
- assertNotNull("Unable to obtain "+ShardParams.SHARDS_INFO, shardsInfo);
+ assertNotNull("Unable to obtain " + ShardParams.SHARDS_INFO, shardsInfo);
// Iterate over shards-info and check what cores responded
- SimpleOrderedMap<?> shardsInfoMap = (SimpleOrderedMap<?>)shardsInfo;
+ SimpleOrderedMap<?> shardsInfoMap = (SimpleOrderedMap<?>) shardsInfo;
@SuppressWarnings({"unchecked"})
Iterator<Map.Entry<String, ?>> itr = shardsInfoMap.asMap(100).entrySet().iterator();
List<String> shardAddresses = new ArrayList<String>();
while (itr.hasNext()) {
Map.Entry<String, ?> e = itr.next();
- assertTrue("Did not find map-type value in "+ShardParams.SHARDS_INFO, e.getValue() instanceof Map);
- String shardAddress = (String)((Map)e.getValue()).get("shardAddress");
+ assertTrue("Did not find map-type value in " + ShardParams.SHARDS_INFO, e.getValue() instanceof Map);
+ String shardAddress = (String) ((Map) e.getValue()).get("shardAddress");
if (shardAddress.endsWith("/")) {
shardAddress = shardAddress.substring(0, shardAddress.length() - 1);
}
- assertNotNull(ShardParams.SHARDS_INFO+" did not return 'shardAddress' parameter", shardAddress);
+ assertNotNull(ShardParams.SHARDS_INFO + " did not return 'shardAddress' parameter", shardAddress);
shardAddresses.add(shardAddress);
}
assertEquals("Shard addresses must be of size 1, since there is only 1 shard in the collection", 1, shardAddresses.size());
@@ -557,7 +569,7 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
SolrServerException, IOException {
NamedList<Object> resp;
- try (HttpSolrClient client = getHttpSolrClient(baseUrl + "/"+ collectionName, 15000, 60000)) {
+ try (HttpSolrClient client = getHttpSolrClient(baseUrl + "/" + collectionName, 15000, 60000)) {
ModifiableSolrParams params = new ModifiableSolrParams();
params.set("qt", "/admin/mbeans");
params.set("stats", "true");
@@ -574,12 +586,12 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
name = category + "." + (scope != null ? scope : key) + ".requests";
}
@SuppressWarnings({"unchecked"})
- Map<String,Object> map = (Map<String,Object>)resp.findRecursive("solr-mbeans", category, key, "stats");
+ Map<String, Object> map = (Map<String, Object>) resp.findRecursive("solr-mbeans", category, key, "stats");
if (map == null) {
return null;
}
if (scope != null) { // admin handler uses a meter instead of counter here
- return (Long)map.get(name + ".count");
+ return (Long) map.get(name + ".count");
} else {
return (Long) map.get(name);
}
@@ -642,8 +654,10 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
try (CloudSolrClient client = getCloudSolrClient(cluster.getZkServer().getZkAddress())) {
String async1 = CollectionAdminRequest.createCollection("multicollection1", "conf", 2, 1)
+ .setPerReplicaState(USE_PER_REPLICA_STATE)
.processAsync(client);
String async2 = CollectionAdminRequest.createCollection("multicollection2", "conf", 2, 1)
+ .setPerReplicaState(USE_PER_REPLICA_STATE)
.processAsync(client);
CollectionAdminRequest.waitForAsyncRequest(async1, client, TIMEOUT);
@@ -709,7 +723,7 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
QueryResponse rsp = solrClient.query(q);
@SuppressWarnings({"rawtypes"})
- Map m = (Map) rsp.getResponse().get(CloudSolrClient.STATE_VERSION, rsp.getResponse().size()-1);
+ Map m = (Map) rsp.getResponse().get(CloudSolrClient.STATE_VERSION, rsp.getResponse().size() - 1);
assertNotNull("Expected an extra information from server with the list of invalid collection states", m);
assertNotNull(m.get(COLLECTION));
}
@@ -726,19 +740,19 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
Set<String> liveNodes = cluster.getSolrClient().getZkStateReader().getClusterState().getLiveNodes();
for (String s : liveNodes) {
String n = cluster.getSolrClient().getZkStateReader().getBaseUrlForNodeName(s);
- if(!allNodesOfColl.contains(n)){
+ if (!allNodesOfColl.contains(n)) {
theNode = n;
break;
}
}
- log.info("the node which does not serve this collection{} ",theNode);
+ log.info("the node which does not serve this collection{} ", theNode);
assertNotNull(theNode);
final String solrClientUrl = theNode + "/" + COLLECTION;
try (SolrClient solrClient = getHttpSolrClient(solrClientUrl)) {
- q.setParam(CloudSolrClient.STATE_VERSION, COLLECTION + ":" + (coll.getZNodeVersion()-1));
+ q.setParam(CloudSolrClient.STATE_VERSION, COLLECTION + ":" + (coll.getZNodeVersion() - 1));
try {
QueryResponse rsp = solrClient.query(q);
log.info("error was expected");
@@ -786,9 +800,11 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
@Test
public void testVersionsAreReturned() throws Exception {
- CollectionAdminRequest.createCollection("versions_collection", "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection("versions_collection", "conf", 2, 1)
+ .setPerReplicaState(USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection("versions_collection", 2, 2);
-
+
// assert that "adds" are returned
UpdateRequest updateRequest = new UpdateRequest()
.add("id", "1", "a_t", "hello1")
@@ -797,7 +813,7 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
NamedList<Object> response = updateRequest.commit(getRandomClient(), "versions_collection").getResponse();
Object addsObject = response.get("adds");
-
+
assertNotNull("There must be a adds parameter", addsObject);
assertTrue(addsObject instanceof NamedList<?>);
NamedList<?> adds = (NamedList<?>) addsObject;
@@ -832,10 +848,12 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
NamedList deletes = (NamedList) deletesObject;
assertEquals("There must be 1 version", 1, deletes.size());
}
-
+
@Test
public void testInitializationWithSolrUrls() throws Exception {
- CollectionAdminRequest.createCollection(COLLECTION, "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(COLLECTION, "conf", 2, 1)
+ .setPerReplicaState(USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection(COLLECTION, 2, 2);
CloudSolrClient client = httpBasedCloudSolrClient;
SolrInputDocument doc = new SolrInputDocument("id", "1", "title_s", "my doc");
@@ -858,12 +876,13 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
final JettySolrRunner old_leader_node = cluster.getJettySolrRunners().get(0);
final JettySolrRunner new_leader_node = cluster.getJettySolrRunners().get(1);
-
+
// start with exactly 1 shard/replica...
assertEquals("Couldn't create collection", 0,
- CollectionAdminRequest.createCollection(COL, "conf", 1, 1)
- .setCreateNodeSet(old_leader_node.getNodeName())
- .process(cluster.getSolrClient()).getStatus());
+ CollectionAdminRequest.createCollection(COL, "conf", 1, 1)
+ .setPerReplicaState(USE_PER_REPLICA_STATE)
+ .setCreateNodeSet(old_leader_node.getNodeName())
+ .process(cluster.getSolrClient()).getStatus());
cluster.waitForActiveCollection(COL, 1, 1);
// determine the coreNodeName of only current replica
@@ -882,47 +901,47 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
// don't let collection cache entries get expired, even on a slow machine...
stale_client.setCollectionCacheTTl(Integer.MAX_VALUE);
stale_client.setDefaultCollection(COL);
-
+
// do a query to populate stale_client's cache...
assertEquals(0, stale_client.query(new SolrQuery("*:*")).getResults().getNumFound());
-
+
// add 1 replica on a diff node...
assertEquals("Couldn't create collection", 0,
- CollectionAdminRequest.addReplicaToShard(COL, "shard1")
- .setNode(new_leader_node.getNodeName())
- // NOTE: don't use our stale_client for this -- don't tip it off of a collection change
- .process(cluster.getSolrClient()).getStatus());
+ CollectionAdminRequest.addReplicaToShard(COL, "shard1")
+ .setNode(new_leader_node.getNodeName())
+ // NOTE: don't use our stale_client for this -- don't tip it off of a collection change
+ .process(cluster.getSolrClient()).getStatus());
AbstractDistribZkTestBase.waitForRecoveriesToFinish
- (COL, cluster.getSolrClient().getZkStateReader(), true, true, 330);
-
+ (COL, cluster.getSolrClient().getZkStateReader(), true, true, 330);
+
// ...and delete our original leader.
assertEquals("Couldn't create collection", 0,
- CollectionAdminRequest.deleteReplica(COL, "shard1", old_leader_core_node_name)
- // NOTE: don't use our stale_client for this -- don't tip it off of a collection change
- .process(cluster.getSolrClient()).getStatus());
+ CollectionAdminRequest.deleteReplica(COL, "shard1", old_leader_core_node_name)
+ // NOTE: don't use our stale_client for this -- don't tip it off of a collection change
+ .process(cluster.getSolrClient()).getStatus());
AbstractDistribZkTestBase.waitForRecoveriesToFinish
- (COL, cluster.getSolrClient().getZkStateReader(), true, true, 330);
+ (COL, cluster.getSolrClient().getZkStateReader(), true, true, 330);
// stale_client's collection state cache should now only point at a leader that no longer exists.
-
+
// attempt a (direct) update that should succeed in spite of cached cluster state
// pointing solely to a node that's no longer part of our collection...
assertEquals(0, (new UpdateRequest().add("id", "1").commit(stale_client, COL)).getStatus());
assertEquals(1, stale_client.query(new SolrQuery("*:*")).getResults().getNumFound());
-
+
}
}
-
+
private static void checkSingleServer(NamedList<Object> response) {
final CloudSolrClient.RouteResponse rr = (CloudSolrClient.RouteResponse) response;
- final Map<String,LBHttpSolrClient.Req> routes = rr.getRoutes();
- final Iterator<Map.Entry<String,LBHttpSolrClient.Req>> it =
+ final Map<String, LBHttpSolrClient.Req> routes = rr.getRoutes();
+ final Iterator<Map.Entry<String, LBHttpSolrClient.Req>> it =
routes.entrySet().iterator();
while (it.hasNext()) {
- Map.Entry<String,LBHttpSolrClient.Req> entry = it.next();
- assertEquals("wrong number of servers: "+entry.getValue().getServers(),
- 1, entry.getValue().getServers().size());
+ Map.Entry<String, LBHttpSolrClient.Req> entry = it.next();
+ assertEquals("wrong number of servers: " + entry.getValue().getServers(),
+ 1, entry.getValue().getServers().size());
}
}
@@ -942,10 +961,11 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
// Hence the below configuration for our collection
int pullReplicas = Math.max(1, liveNodes - 2);
CollectionAdminRequest.createCollection(collectionName, "conf", liveNodes, 1, 1, pullReplicas)
+ .setPerReplicaState(USE_PER_REPLICA_STATE)
.setMaxShardsPerNode(liveNodes)
.processAndWait(cluster.getSolrClient(), TIMEOUT);
cluster.waitForActiveCollection(collectionName, liveNodes, liveNodes * (2 + pullReplicas));
-
+
// Add some new documents
new UpdateRequest()
.add(id, "0", "a_t", "hello1")
@@ -975,8 +995,7 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
String preferReplicaTypes,
boolean preferLocalShards,
String collectionName)
- throws Exception
- {
+ throws Exception {
SolrQuery qRequest = new SolrQuery("*:*");
ModifiableSolrParams qParams = new ModifiableSolrParams();
@@ -997,7 +1016,7 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
rule.append(ShardParams.SHARDS_PREFERENCE_REPLICA_LOCATION);
rule.append(":local");
}
- qParams.add(ShardParams.SHARDS_PREFERENCE, rule.toString());
+ qParams.add(ShardParams.SHARDS_PREFERENCE, rule.toString());
qParams.add(ShardParams.SHARDS_INFO, "true");
qRequest.add(qParams);
@@ -1008,7 +1027,7 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
QueryResponse qResponse = cloudClient.query(collectionName, qRequest);
Object shardsInfo = qResponse.getResponse().get(ShardParams.SHARDS_INFO);
- assertNotNull("Unable to obtain "+ShardParams.SHARDS_INFO, shardsInfo);
+ assertNotNull("Unable to obtain " + ShardParams.SHARDS_INFO, shardsInfo);
Map<String, String> replicaTypeMap = new HashMap<String, String>();
DocCollection collection = getCollectionState(collectionName);
@@ -1025,15 +1044,15 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
}
// Iterate over shards-info and check that replicas of correct type responded
- SimpleOrderedMap<?> shardsInfoMap = (SimpleOrderedMap<?>)shardsInfo;
+ SimpleOrderedMap<?> shardsInfoMap = (SimpleOrderedMap<?>) shardsInfo;
@SuppressWarnings({"unchecked"})
Iterator<Map.Entry<String, ?>> itr = shardsInfoMap.asMap(100).entrySet().iterator();
List<String> shardAddresses = new ArrayList<String>();
while (itr.hasNext()) {
Map.Entry<String, ?> e = itr.next();
- assertTrue("Did not find map-type value in "+ShardParams.SHARDS_INFO, e.getValue() instanceof Map);
- String shardAddress = (String)((Map)e.getValue()).get("shardAddress");
- assertNotNull(ShardParams.SHARDS_INFO+" did not return 'shardAddress' parameter", shardAddress);
+ assertTrue("Did not find map-type value in " + ShardParams.SHARDS_INFO, e.getValue() instanceof Map);
+ String shardAddress = (String) ((Map) e.getValue()).get("shardAddress");
+ assertNotNull(ShardParams.SHARDS_INFO + " did not return 'shardAddress' parameter", shardAddress);
assertTrue(replicaTypeMap.containsKey(shardAddress));
assertTrue(preferredTypes.indexOf(replicaTypeMap.get(shardAddress)) == 0);
shardAddresses.add(shardAddress);
@@ -1047,7 +1066,9 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
@Test
public void testPing() throws Exception {
final String testCollection = "ping_test";
- CollectionAdminRequest.createCollection(testCollection, "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(testCollection, "conf", 2, 1)
+ .setPerReplicaState(USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection(testCollection, 2, 2);
final SolrClient clientUnderTest = getRandomClient();
@@ -1056,4 +1077,24 @@ public class CloudSolrClientTest extends SolrCloudTestCase {
assertEquals("This should be OK", 0, response.getStatus());
}
+ public void testPerReplicaStateCollection() throws Exception {
+ CollectionAdminRequest.createCollection("versions_collection", "conf", 2, 1)
+ .process(cluster.getSolrClient());
+
+ final String testCollection = "perReplicaState_test";
+ int liveNodes = cluster.getJettySolrRunners().size();
+ CollectionAdminRequest.createCollection(testCollection, "conf", 2, 2)
+ .setMaxShardsPerNode(liveNodes)
+ .setPerReplicaState(Boolean.TRUE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(testCollection, 2, 4);
+ final SolrClient clientUnderTest = getRandomClient();
+ final SolrPingResponse response = clientUnderTest.ping(testCollection);
+ assertEquals("This should be OK", 0, response.getStatus());
+ DocCollection c = cluster.getSolrClient().getZkStateReader().getCollection(testCollection);
+ c.forEachReplica((s, replica) -> assertNotNull(replica.getReplicaState()));
+ PerReplicaStates prs = PerReplicaStates.fetch(ZkStateReader.getCollectionPath(testCollection), cluster.getZkClient(), null);
+ assertEquals(4, prs.states.size());
+ }
+
}
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/CloudAuthStreamTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/CloudAuthStreamTest.java
index e8aef51..22d6a0c 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/CloudAuthStreamTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/CloudAuthStreamTest.java
@@ -126,7 +126,8 @@ public class CloudAuthStreamTest extends SolrCloudTestCase {
for (String collection : Arrays.asList(COLLECTION_X, COLLECTION_Y)) {
CollectionAdminRequest.createCollection(collection, "_default", 2, 2)
- .setBasicAuthCredentials(ADMIN_USER, ADMIN_USER)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .setBasicAuthCredentials(ADMIN_USER, ADMIN_USER)
.process(cluster.getSolrClient());
}
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/JDBCStreamTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/JDBCStreamTest.java
index 397d655..012b21a 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/JDBCStreamTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/JDBCStreamTest.java
@@ -74,7 +74,9 @@ public class JDBCStreamTest extends SolrCloudTestCase {
} else {
collection = COLLECTIONORALIAS;
}
- CollectionAdminRequest.createCollection(collection, "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(collection, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
AbstractDistribZkTestBase.waitForRecoveriesToFinish(collection, cluster.getSolrClient().getZkStateReader(),
false, true, TIMEOUT);
if (useAlias) {
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/MathExpressionTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/MathExpressionTest.java
index edef269..b7eaa64 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/MathExpressionTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/MathExpressionTest.java
@@ -65,7 +65,10 @@ public class MathExpressionTest extends SolrCloudTestCase {
collection = COLLECTIONORALIAS;
}
- CollectionAdminRequest.createCollection(collection, "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(collection, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
AbstractDistribZkTestBase.waitForRecoveriesToFinish(collection, cluster.getSolrClient().getZkStateReader(),
false, true, TIMEOUT);
if (useAlias) {
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/SelectWithEvaluatorsTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/SelectWithEvaluatorsTest.java
index add4331..2fa0dd0 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/SelectWithEvaluatorsTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/SelectWithEvaluatorsTest.java
@@ -68,7 +68,9 @@ public class SelectWithEvaluatorsTest extends SolrCloudTestCase {
} else {
collection = COLLECTIONORALIAS;
}
- CollectionAdminRequest.createCollection(collection, "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(collection, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
AbstractDistribZkTestBase.waitForRecoveriesToFinish(collection, cluster.getSolrClient().getZkStateReader(),
false, true, TIMEOUT);
if (useAlias) {
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamDecoratorTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamDecoratorTest.java
index 6c88ffe..4d77540 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamDecoratorTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamDecoratorTest.java
@@ -96,7 +96,8 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
collection = COLLECTIONORALIAS;
}
- CollectionAdminRequest.createCollection(collection, "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(collection, "conf", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection(collection, 2, 2);
@@ -2679,7 +2680,8 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
@Test
public void testUpdateStream() throws Exception {
- CollectionAdminRequest.createCollection("destinationCollection", "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection("destinationCollection", "conf", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection("destinationCollection", 2, 2);
new UpdateRequest()
@@ -2773,7 +2775,8 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
@Test
public void testParallelUpdateStream() throws Exception {
- CollectionAdminRequest.createCollection("parallelDestinationCollection", "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection("parallelDestinationCollection", "conf", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection("parallelDestinationCollection", 2, 2);
new UpdateRequest()
@@ -2871,7 +2874,8 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
@Test
public void testParallelDaemonUpdateStream() throws Exception {
- CollectionAdminRequest.createCollection("parallelDestinationCollection1", "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection("parallelDestinationCollection1", "conf", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection("parallelDestinationCollection1", 2, 2);
new UpdateRequest()
@@ -3045,7 +3049,8 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
public void testParallelTerminatingDaemonUpdateStream() throws Exception {
Assume.assumeTrue(!useAlias);
- CollectionAdminRequest.createCollection("parallelDestinationCollection1", "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection("parallelDestinationCollection1", "conf", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection("parallelDestinationCollection1", 2, 2);
new UpdateRequest()
@@ -3231,7 +3236,8 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
@Test
public void testCommitStream() throws Exception {
- CollectionAdminRequest.createCollection("destinationCollection", "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection("destinationCollection", "conf", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection("destinationCollection", 2, 2);
new UpdateRequest()
@@ -3324,7 +3330,8 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
@Test
public void testParallelCommitStream() throws Exception {
- CollectionAdminRequest.createCollection("parallelDestinationCollection", "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection("parallelDestinationCollection", "conf", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection("parallelDestinationCollection", 2, 2);
new UpdateRequest()
@@ -3422,7 +3429,8 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
@Test
public void testParallelDaemonCommitStream() throws Exception {
- CollectionAdminRequest.createCollection("parallelDestinationCollection1", "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection("parallelDestinationCollection1", "conf", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection("parallelDestinationCollection1", 2, 2);
new UpdateRequest()
@@ -3639,11 +3647,14 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
public void testClassifyStream() throws Exception {
Assume.assumeTrue(!useAlias);
- CollectionAdminRequest.createCollection("modelCollection", "ml", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection("modelCollection", "ml", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection("modelCollection", 2, 2);
- CollectionAdminRequest.createCollection("uknownCollection", "ml", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection("uknownCollection", "ml", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection("uknownCollection", 2, 2);
- CollectionAdminRequest.createCollection("checkpointCollection", "ml", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection("checkpointCollection", "ml", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
cluster.waitForActiveCollection("checkpointCollection", 2, 2);
UpdateRequest updateRequest = new UpdateRequest();
@@ -3867,11 +3878,14 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
@Test
public void testExecutorStream() throws Exception {
- CollectionAdminRequest.createCollection("workQueue", "conf", 2, 1).processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
+ CollectionAdminRequest.createCollection("workQueue", "conf", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
cluster.waitForActiveCollection("workQueue", 2, 2);
- CollectionAdminRequest.createCollection("mainCorpus", "conf", 2, 1).processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
+ CollectionAdminRequest.createCollection("mainCorpus", "conf", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
cluster.waitForActiveCollection("mainCorpus", 2, 2);
- CollectionAdminRequest.createCollection("destination", "conf", 2, 1).processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
+ CollectionAdminRequest.createCollection("destination", "conf", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
cluster.waitForActiveCollection("destination", 2, 2);
UpdateRequest workRequest = new UpdateRequest();
@@ -3933,11 +3947,14 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
@Test
public void testParallelExecutorStream() throws Exception {
- CollectionAdminRequest.createCollection("workQueue1", "conf", 2, 1).processAndWait(cluster.getSolrClient(),DEFAULT_TIMEOUT);
+ CollectionAdminRequest.createCollection("workQueue1", "conf", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .processAndWait(cluster.getSolrClient(),DEFAULT_TIMEOUT);
- CollectionAdminRequest.createCollection("mainCorpus1", "conf", 2, 1).processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
+ CollectionAdminRequest.createCollection("mainCorpus1", "conf", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
- CollectionAdminRequest.createCollection("destination1", "conf", 2, 1).processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
+ CollectionAdminRequest.createCollection("destination1", "conf", 2, 1).setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
cluster.waitForActiveCollection("workQueue1", 2, 2);
cluster.waitForActiveCollection("mainCorpus1", 2, 2);
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/request/json/DirectJsonQueryRequestFacetingIntegrationTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/request/json/DirectJsonQueryRequestFacetingIntegrationTest.java
index 48f13c2..9563da7 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/request/json/DirectJsonQueryRequestFacetingIntegrationTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/request/json/DirectJsonQueryRequestFacetingIntegrationTest.java
@@ -56,7 +56,9 @@ public class DirectJsonQueryRequestFacetingIntegrationTest extends SolrCloudTest
final List<String> solrUrls = new ArrayList<>();
solrUrls.add(cluster.getJettySolrRunner(0).getBaseUrl().toString());
- CollectionAdminRequest.createCollection(COLLECTION_NAME, CONFIG_NAME, 1, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, CONFIG_NAME, 1, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
ContentStreamUpdateRequest up = new ContentStreamUpdateRequest("/update");
up.setParam("collection", COLLECTION_NAME);
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/request/json/JsonQueryRequestFacetingIntegrationTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/request/json/JsonQueryRequestFacetingIntegrationTest.java
index f4406c1..9d5c3e6 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/request/json/JsonQueryRequestFacetingIntegrationTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/request/json/JsonQueryRequestFacetingIntegrationTest.java
@@ -63,7 +63,9 @@ public class JsonQueryRequestFacetingIntegrationTest extends SolrCloudTestCase {
final List<String> solrUrls = new ArrayList<>();
solrUrls.add(cluster.getJettySolrRunner(0).getBaseUrl().toString());
- CollectionAdminRequest.createCollection(COLLECTION_NAME, CONFIG_NAME, 1, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, CONFIG_NAME, 1, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
ContentStreamUpdateRequest up = new ContentStreamUpdateRequest("/update");
up.setParam("collection", COLLECTION_NAME);
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/request/json/JsonQueryRequestHeatmapFacetingTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/request/json/JsonQueryRequestHeatmapFacetingTest.java
index 1ccd581..977d405 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/request/json/JsonQueryRequestHeatmapFacetingTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/request/json/JsonQueryRequestHeatmapFacetingTest.java
@@ -47,7 +47,9 @@ public class JsonQueryRequestHeatmapFacetingTest extends SolrCloudTestCase {
final List<String> solrUrls = new ArrayList<>();
solrUrls.add(cluster.getJettySolrRunner(0).getBaseUrl().toString());
- CollectionAdminRequest.createCollection(COLLECTION_NAME, CONFIG_NAME, 1, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, CONFIG_NAME, 1, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
indexSpatialData();
}
diff --git a/solr/solrj/src/test/org/apache/solr/common/cloud/TestCloudCollectionsListeners.java b/solr/solrj/src/test/org/apache/solr/common/cloud/TestCloudCollectionsListeners.java
index 1ef806e..6d546a4 100644
--- a/solr/solrj/src/test/org/apache/solr/common/cloud/TestCloudCollectionsListeners.java
+++ b/solr/solrj/src/test/org/apache/solr/common/cloud/TestCloudCollectionsListeners.java
@@ -97,6 +97,7 @@ public class TestCloudCollectionsListeners extends SolrCloudTestCase {
assertFalse("CloudCollectionsListener not triggered after registration", newResults.get(2).contains("testcollection1"));
CollectionAdminRequest.createCollection("testcollection1", "config", 4, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.processAndWait(client, MAX_WAIT_TIMEOUT);
client.waitForState("testcollection1", MAX_WAIT_TIMEOUT, TimeUnit.SECONDS,
(n, c) -> DocCollection.isFullyActive(n, c, 4, 1));
@@ -110,6 +111,7 @@ public class TestCloudCollectionsListeners extends SolrCloudTestCase {
client.getZkStateReader().removeCloudCollectionsListener(watcher1);
CollectionAdminRequest.createCollection("testcollection2", "config", 4, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.processAndWait(client, MAX_WAIT_TIMEOUT);
cluster.waitForActiveCollection("testcollection2", 4, 4);
@@ -136,10 +138,12 @@ public class TestCloudCollectionsListeners extends SolrCloudTestCase {
CloudSolrClient client = cluster.getSolrClient();
CollectionAdminRequest.createCollection("testcollection1", "config", 4, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.processAndWait(client, MAX_WAIT_TIMEOUT);
cluster.waitForActiveCollection("testcollection1", 4, 4);
CollectionAdminRequest.createCollection("testcollection2", "config", 4, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
.processAndWait(client, MAX_WAIT_TIMEOUT);
cluster.waitForActiveCollection("testcollection2", 4, 4);
diff --git a/solr/solrj/src/test/org/apache/solr/common/cloud/TestCollectionStateWatchers.java b/solr/solrj/src/test/org/apache/solr/common/cloud/TestCollectionStateWatchers.java
index ab3dc95..e7587fe 100644
--- a/solr/solrj/src/test/org/apache/solr/common/cloud/TestCollectionStateWatchers.java
+++ b/solr/solrj/src/test/org/apache/solr/common/cloud/TestCollectionStateWatchers.java
@@ -33,6 +33,7 @@ import org.apache.solr.cloud.SolrCloudTestCase;
import org.apache.solr.common.util.ExecutorUtil;
import org.junit.After;
import org.junit.Before;
+import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -121,7 +122,8 @@ public class TestCollectionStateWatchers extends SolrCloudTestCase {
// note: one node in our cluster is unsed by collection
CollectionAdminRequest.createCollection("testcollection", "config", CLUSTER_SIZE, 1)
- .processAndWait(client, MAX_WAIT_TIMEOUT);
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .processAndWait(client, MAX_WAIT_TIMEOUT);
client.waitForState("testcollection", MAX_WAIT_TIMEOUT, TimeUnit.SECONDS,
(n, c) -> DocCollection.isFullyActive(n, c, CLUSTER_SIZE, 1));
@@ -169,7 +171,8 @@ public class TestCollectionStateWatchers extends SolrCloudTestCase {
CloudSolrClient client = cluster.getSolrClient();
CollectionAdminRequest.createCollection("currentstate", "config", 1, 1)
- .processAndWait(client, MAX_WAIT_TIMEOUT);
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .processAndWait(client, MAX_WAIT_TIMEOUT);
final CountDownLatch latch = new CountDownLatch(1);
client.registerCollectionStateWatcher("currentstate", (n, c) -> {
@@ -199,7 +202,8 @@ public class TestCollectionStateWatchers extends SolrCloudTestCase {
CloudSolrClient client = cluster.getSolrClient();
CollectionAdminRequest.createCollection("waitforstate", "config", 1, 1)
- .processAndWait(client, MAX_WAIT_TIMEOUT);
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .processAndWait(client, MAX_WAIT_TIMEOUT);
client.waitForState("waitforstate", MAX_WAIT_TIMEOUT, TimeUnit.SECONDS,
(n, c) -> DocCollection.isFullyActive(n, c, 1, 1));
@@ -217,6 +221,8 @@ public class TestCollectionStateWatchers extends SolrCloudTestCase {
}
@Test
+ @Ignore
+ //nocommit
public void testCanWaitForNonexistantCollection() throws Exception {
Future<Boolean> future = waitInBackground("delayed", MAX_WAIT_TIMEOUT, TimeUnit.SECONDS,
@@ -247,7 +253,8 @@ public class TestCollectionStateWatchers extends SolrCloudTestCase {
CloudSolrClient client = cluster.getSolrClient();
CollectionAdminRequest.createCollection("falsepredicate", "config", 4, 1)
- .processAndWait(client, MAX_WAIT_TIMEOUT);
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .processAndWait(client, MAX_WAIT_TIMEOUT);
client.waitForState("falsepredicate", MAX_WAIT_TIMEOUT, TimeUnit.SECONDS,
(n, c) -> DocCollection.isFullyActive(n, c, 4, 1));
@@ -301,7 +308,8 @@ public class TestCollectionStateWatchers extends SolrCloudTestCase {
@Test
public void testDeletionsTriggerWatches() throws Exception {
CollectionAdminRequest.createCollection("tobedeleted", "config", 1, 1)
- .process(cluster.getSolrClient());
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
Future<Boolean> future = waitInBackground("tobedeleted", MAX_WAIT_TIMEOUT, TimeUnit.SECONDS,
(l, c) -> c == null);
@@ -315,7 +323,9 @@ public class TestCollectionStateWatchers extends SolrCloudTestCase {
public void testLiveNodeChangesTriggerWatches() throws Exception {
final CloudSolrClient client = cluster.getSolrClient();
- CollectionAdminRequest.createCollection("test_collection", "config", 1, 1).process(client);
+ CollectionAdminRequest.createCollection("test_collection", "config", 1, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(client);
Future<Boolean> future = waitInBackground("test_collection", MAX_WAIT_TIMEOUT, TimeUnit.SECONDS,
(l, c) -> (l.size() == 1 + CLUSTER_SIZE));
diff --git a/solr/solrj/src/test/org/apache/solr/common/cloud/TestPerReplicaStates.java b/solr/solrj/src/test/org/apache/solr/common/cloud/TestPerReplicaStates.java
new file mode 100644
index 0000000..c70807a
--- /dev/null
+++ b/solr/solrj/src/test/org/apache/solr/common/cloud/TestPerReplicaStates.java
@@ -0,0 +1,137 @@
+/*
+ * 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.common.cloud;
+
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.cloud.Replica.State;
+import org.apache.zookeeper.CreateMode;
+import org.junit.After;
+import org.junit.Before;
+
+public class TestPerReplicaStates extends SolrCloudTestCase {
+ @Before
+ public void prepareCluster() throws Exception {
+ configureCluster(4)
+ .configure();
+ }
+
+ @After
+ public void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ public void testBasic() {
+ PerReplicaStates.State rs = new PerReplicaStates.State("R1", State.ACTIVE, Boolean.FALSE, 1);
+ assertEquals("R1:1:A", rs.asString);
+
+ rs = new PerReplicaStates.State("R1", State.DOWN, Boolean.TRUE, 1);
+ assertEquals("R1:1:D:L", rs.asString);
+ rs = PerReplicaStates.State.parse (rs.asString);
+ assertEquals(State.DOWN, rs.state);
+
+ }
+
+ public void testEntries() {
+ PerReplicaStates entries = new PerReplicaStates("state.json", 0, ImmutableList.of("R1:2:A", "R1:1:A:L", "R1:0:D", "R2:0:D", "R3:0:A"));
+ assertEquals(2, entries.get("R1").version);
+ entries = new PerReplicaStates("state.json", 0, ImmutableList.of("R1:1:A:L", "R1:2:A", "R2:0:D", "R3:0:A", "R1:0:D"));
+ assertEquals(2, entries.get("R1").version);
+ assertEquals(2, entries.get("R1").getDuplicates().size());
+ Set<String> modified = PerReplicaStates.findModifiedReplicas(entries, new PerReplicaStates("state.json", 0, ImmutableList.of("R1:1:A:L", "R1:2:A", "R2:0:D", "R3:1:A", "R1:0:D")));
+ assertEquals(1, modified.size());
+ assertTrue(modified.contains("R3"));
+ modified = PerReplicaStates.findModifiedReplicas( entries,
+ new PerReplicaStates("state.json", 0, ImmutableList.of("R1:1:A:L", "R1:2:A", "R2:0:D", "R3:1:A", "R1:0:D", "R4:0:A")));
+ assertEquals(2, modified.size());
+ assertTrue(modified.contains("R3"));
+ assertTrue(modified.contains("R4"));
+ modified = PerReplicaStates.findModifiedReplicas( entries,
+ new PerReplicaStates("state.json", 0, ImmutableList.of("R1:1:A:L", "R1:2:A", "R3:1:A", "R1:0:D", "R4:0:A")));
+ assertEquals(3, modified.size());
+ assertTrue(modified.contains("R3"));
+ assertTrue(modified.contains("R4"));
+ assertTrue(modified.contains("R2"));
+
+
+ }
+
+ public void testReplicaStateOperations() throws Exception {
+ String root = "/testReplicaStateOperations";
+ cluster.getZkClient().create(root, null, CreateMode.PERSISTENT, true);
+
+ ImmutableList<String> states = ImmutableList.of("R1:2:A", "R1:1:A:L", "R1:0:D", "R3:0:A", "R4:13:A");
+
+ for (String state : states) {
+ cluster.getZkClient().create(root + "/" + state, null, CreateMode.PERSISTENT, true);
+ }
+
+ ZkStateReader zkStateReader = cluster.getSolrClient().getZkStateReader();
+ PerReplicaStates rs = zkStateReader.getReplicaStates(new PerReplicaStates(root, 0, Collections.emptyList()));
+ assertEquals(3, rs.states.size());
+ assertTrue(rs.cversion >= 5);
+
+ List<PerReplicaStates.Op> ops = PerReplicaStates.WriteOps.addReplica("R5",State.ACTIVE, false, rs).get();
+
+ assertEquals(1, ops.size());
+ assertEquals(PerReplicaStates.Op.Type.ADD ,ops.get(0).typ );
+ PerReplicaStates.persist(ops, root,cluster.getZkClient());
+ rs = zkStateReader.getReplicaStates(root);
+ assertEquals(4, rs.states.size());
+ assertTrue(rs.cversion >= 6);
+ assertEquals(6, cluster.getZkClient().getChildren(root, null,true).size());
+ ops = PerReplicaStates.WriteOps.flipState("R1", State.DOWN , rs).get();
+
+ assertEquals(4, ops.size());
+ assertEquals(PerReplicaStates.Op.Type.ADD, ops.get(0).typ);
+ assertEquals(PerReplicaStates.Op.Type.DELETE, ops.get(1).typ);
+ assertEquals(PerReplicaStates.Op.Type.DELETE, ops.get(2).typ);
+ assertEquals(PerReplicaStates.Op.Type.DELETE, ops.get(3).typ);
+ PerReplicaStates.persist(ops, root,cluster.getZkClient());
+ rs = zkStateReader.getReplicaStates(root);
+ assertEquals(4, rs.states.size());
+ assertEquals(3, rs.states.get("R1").version);
+
+ ops = PerReplicaStates.WriteOps.deleteReplica("R5" , rs).get();
+ assertEquals(1, ops.size());
+ PerReplicaStates.persist(ops, root,cluster.getZkClient());
+
+ rs = zkStateReader.getReplicaStates(root);
+ assertEquals(3, rs.states.size());
+
+ ops = PerReplicaStates.WriteOps.flipLeader(ImmutableSet.of("R4","R3","R1"), "R4",rs).get();
+
+ assertEquals(2, ops.size());
+ assertEquals(PerReplicaStates.Op.Type.ADD, ops.get(0).typ);
+ assertEquals(PerReplicaStates.Op.Type.DELETE, ops.get(1).typ);
+ PerReplicaStates.persist(ops, root,cluster.getZkClient());
+ rs = zkStateReader.getReplicaStates(root);
+ ops = PerReplicaStates.WriteOps.flipLeader(ImmutableSet.of("R4","R3","R1"),"R3",rs).get();
+ assertEquals(4, ops.size());
+ PerReplicaStates.persist(ops, root,cluster.getZkClient());
+ rs = zkStateReader.getReplicaStates(root);
+ assertTrue(rs.get("R3").isLeader);
+ }
+
+}
diff --git a/solr/test-framework/src/java/org/apache/solr/cloud/SolrCloudTestCase.java b/solr/test-framework/src/java/org/apache/solr/cloud/SolrCloudTestCase.java
index b646e2e..c6f26c6 100644
--- a/solr/test-framework/src/java/org/apache/solr/cloud/SolrCloudTestCase.java
+++ b/solr/test-framework/src/java/org/apache/solr/cloud/SolrCloudTestCase.java
@@ -81,6 +81,7 @@ import org.slf4j.LoggerFactory;
public class SolrCloudTestCase extends SolrTestCaseJ4 {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ public static final Boolean USE_PER_REPLICA_STATE = Boolean.parseBoolean(System.getProperty("use.per-replica", "false"));
public static final int DEFAULT_TIMEOUT = 45; // this is an important timeout for test stability - can't be too short