You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by no...@apache.org on 2019/12/06 02:39:27 UTC
[lucene-solr] branch jira/solr14003 updated: eliminated all uses of
Replica#isActive()
This is an automated email from the ASF dual-hosted git repository.
noble pushed a commit to branch jira/solr14003
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
The following commit(s) were added to refs/heads/jira/solr14003 by this push:
new 38c5b8f eliminated all uses of Replica#isActive()
38c5b8f is described below
commit 38c5b8fec85541972e6234bcd1e099d4fee1fafb
Author: noble <no...@apache.org>
AuthorDate: Fri Dec 6 13:39:03 2019 +1100
eliminated all uses of Replica#isActive()
---
.../apache/solr/cloud/ActiveReplicaWatcher.java | 9 ++++---
.../src/java/org/apache/solr/cloud/CloudUtil.java | 4 +--
.../src/java/org/apache/solr/cloud/Overseer.java | 16 ++++++-----
.../java/org/apache/solr/cloud/ZkController.java | 31 ++++------------------
.../api/collections/LeaderRecoveryWatcher.java | 5 ++--
.../autoscaling/sim/SimClusterStateProvider.java | 8 +++---
.../solr/cloud/autoscaling/sim/SimUtils.java | 4 ++-
.../org/apache/solr/handler/admin/ColStatus.java | 19 ++++++-------
.../solr/handler/admin/CollectionsHandler.java | 2 +-
.../solr/handler/admin/MetricsHistoryHandler.java | 7 +++--
.../solr/handler/admin/RebalanceLeaders.java | 7 +++--
.../apache/solr/cloud/CollectionsAPISolrJTest.java | 2 +-
.../org/apache/solr/cloud/MoveReplicaTest.java | 7 +++--
.../solr/cloud/TestCloudSearcherWarming.java | 5 ++--
.../org/apache/solr/cloud/TestPullReplica.java | 16 ++++++-----
.../solr/cloud/TestPullReplicaErrorHandling.java | 2 +-
.../apache/solr/cloud/TestRebalanceLeaders.java | 13 ++++-----
.../org/apache/solr/cloud/TestTlogReplica.java | 10 ++++---
.../solr/cloud/api/collections/ShardSplitTest.java | 8 +++---
.../cloud/autoscaling/ComputePlanActionTest.java | 8 +++---
.../ScheduledMaintenanceTriggerTest.java | 2 +-
.../solr/cloud/autoscaling/TestPolicyCloud.java | 6 ++---
.../autoscaling/sim/TestSimComputePlanAction.java | 2 +-
.../test/org/apache/solr/cloud/rule/RulesTest.java | 9 +++----
.../admin/AutoscalingHistoryHandlerTest.java | 7 +++--
.../solr/client/solrj/cloud/DirectShardState.java | 4 +++
.../client/solrj/cloud/ShardStateProvider.java | 2 ++
.../client/solrj/impl/BaseCloudSolrClient.java | 4 ++-
.../solrj/impl/BaseHttpClusterStateProvider.java | 2 +-
.../client/solrj/impl/ClusterStateProvider.java | 2 +-
.../solrj/impl/ZkClientClusterStateProvider.java | 2 +-
.../solr/common/cloud/CollectionStateWatcher.java | 4 ++-
.../solr/common/cloud/DocCollectionWatcher.java | 4 ++-
.../java/org/apache/solr/common/cloud/Replica.java | 1 +
.../apache/solr/common/cloud/ZkStateReader.java | 25 ++++++++++-------
.../common/cloud/TestCollectionStateWatchers.java | 8 +++---
.../common/cloud/TestDocCollectionWatcher.java | 4 +--
.../solr/cloud/AbstractFullDistribZkTestBase.java | 4 ++-
.../org/apache/solr/cloud/SolrCloudTestCase.java | 9 ++++---
39 files changed, 157 insertions(+), 127 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/cloud/ActiveReplicaWatcher.java b/solr/core/src/java/org/apache/solr/cloud/ActiveReplicaWatcher.java
index c6bd807..9c6488a 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ActiveReplicaWatcher.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ActiveReplicaWatcher.java
@@ -21,6 +21,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Set;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.common.SolrCloseableLatch;
import org.apache.solr.common.cloud.CollectionStateWatcher;
import org.apache.solr.common.cloud.DocCollection;
@@ -32,7 +33,7 @@ import org.slf4j.LoggerFactory;
/**
* Watch for replicas to become {@link org.apache.solr.common.cloud.Replica.State#ACTIVE}. Watcher is
- * terminated (its {@link #onStateChanged(Set, DocCollection)} method returns false) when all listed
+ * terminated (its {@link #onStateChanged(ShardStateProvider, Set, DocCollection)} method returns false) when all listed
* replicas become active.
* <p>Additionally, the provided {@link SolrCloseableLatch} instance can be used to await
* for all listed replicas to become active.</p>
@@ -114,7 +115,7 @@ public class ActiveReplicaWatcher implements CollectionStateWatcher {
// synchronized due to SOLR-11535
@Override
- public synchronized boolean onStateChanged(Set<String> liveNodes, DocCollection collectionState) {
+ public synchronized boolean onStateChanged(ShardStateProvider ssp, Set<String> liveNodes, DocCollection collectionState) {
log.debug("-- onStateChanged@" + Long.toHexString(hashCode()) + ": replicaIds=" + replicaIds + ", solrCoreNames=" + solrCoreNames +
(latch != null ? "\nlatch count=" + latch.getCount() : "") +
"\ncollectionState=" + collectionState);
@@ -142,7 +143,7 @@ public class ActiveReplicaWatcher implements CollectionStateWatcher {
for (Slice slice : collectionState.getSlices()) {
for (Replica replica : slice.getReplicas()) {
if (replicaIds.contains(replica.getName())) {
- if (replica.isActive(liveNodes)) {
+ if (ssp.isActive(replica)) {
activeReplicas.add(replica);
replicaIds.remove(replica.getName());
if (latch != null) {
@@ -150,7 +151,7 @@ public class ActiveReplicaWatcher implements CollectionStateWatcher {
}
}
} else if (solrCoreNames.contains(replica.getStr(ZkStateReader.CORE_NAME_PROP))) {
- if (replica.isActive(liveNodes)) {
+ if (ssp.isActive(replica)) {
activeReplicas.add(replica);
solrCoreNames.remove(replica.getStr(ZkStateReader.CORE_NAME_PROP));
if (latch != null) {
diff --git a/solr/core/src/java/org/apache/solr/cloud/CloudUtil.java b/solr/core/src/java/org/apache/solr/cloud/CloudUtil.java
index aa7d3e6..d436c21 100644
--- a/solr/core/src/java/org/apache/solr/cloud/CloudUtil.java
+++ b/solr/core/src/java/org/apache/solr/cloud/CloudUtil.java
@@ -208,7 +208,7 @@ public class CloudUtil {
timeout.sleep(100);
continue;
}
- if (predicate.matches(state.getLiveNodes(), coll, cloudManager.getClusterStateProvider().getReplicaStateProvider(collection))) {
+ if (predicate.matches(state.getLiveNodes(), coll, cloudManager.getClusterStateProvider().getShardStateProvider(collection))) {
log.trace("-- predicate matched with state {}", state);
return timeout.timeElapsed(TimeUnit.MILLISECONDS);
}
@@ -264,7 +264,7 @@ public class CloudUtil {
continue;
}
for (Replica replica : slice) {
- if (replica.isActive(liveNodes))
+ if (ssp.isActive(replica))
activeReplicas++;
}
if (activeReplicas != expectedReplicas) {
diff --git a/solr/core/src/java/org/apache/solr/cloud/Overseer.java b/solr/core/src/java/org/apache/solr/cloud/Overseer.java
index a89926f..5a83a6c 100644
--- a/solr/core/src/java/org/apache/solr/cloud/Overseer.java
+++ b/solr/core/src/java/org/apache/solr/cloud/Overseer.java
@@ -16,8 +16,6 @@
*/
package org.apache.solr.cloud;
-import static org.apache.solr.common.params.CommonParams.ID;
-
import java.io.Closeable;
import java.io.IOException;
import java.lang.invoke.MethodHandles;
@@ -31,8 +29,10 @@ import java.util.Optional;
import java.util.Set;
import java.util.function.BiConsumer;
+import com.codahale.metrics.Timer;
import org.apache.lucene.util.Version;
import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.ClusterStateProvider;
@@ -54,6 +54,7 @@ import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.ConnectionManager;
import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkNodeProps;
@@ -76,7 +77,7 @@ import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.codahale.metrics.Timer;
+import static org.apache.solr.common.params.CommonParams.ID;
/**
* Cluster leader. Responsible for processing state updates, node assignments, creating/deleting
@@ -610,10 +611,12 @@ public class Overseer implements SolrCloseable {
if (coll == null) {
return;
}
+ ShardStateProvider ssp = zkController.getZkStateReader().getShardStateProvider(coll.getName());
// check that all shard leaders are active
boolean allActive = true;
for (Slice s : coll.getActiveSlices()) {
- if (s.getLeader() == null || !s.getLeader().isActive(clusterState.getLiveNodes())) {
+ Replica leader = ssp.getLeader(s);
+ if (leader == null || !ssp.isActive(leader)) {
allActive = false;
break;
}
@@ -622,13 +625,14 @@ public class Overseer implements SolrCloseable {
doCompatCheck(consumer);
} else {
// wait for all leaders to become active and then check
- zkController.zkStateReader.registerCollectionStateWatcher(CollectionAdminParams.SYSTEM_COLL, (liveNodes, state) -> {
+ zkController.zkStateReader.registerCollectionStateWatcher(CollectionAdminParams.SYSTEM_COLL, (stateProvider, liveNodes, state) -> {
boolean active = true;
if (state == null || liveNodes.isEmpty()) {
return true;
}
for (Slice s : state.getActiveSlices()) {
- if (s.getLeader() == null || !s.getLeader().isActive(liveNodes)) {
+ Replica leader = stateProvider.getLeader(s);
+ if (leader == null || !stateProvider.isActive(leader)) {
active = false;
break;
}
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 b5646ba..fc9af6f 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkController.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
@@ -53,43 +53,22 @@ import java.util.concurrent.atomic.AtomicReference;
import com.google.common.base.Strings;
import org.apache.commons.lang3.StringUtils;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
+import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventType;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.impl.HttpSolrClient.Builder;
import org.apache.solr.client.solrj.impl.SolrClientCloudManager;
import org.apache.solr.client.solrj.impl.ZkClientClusterStateProvider;
import org.apache.solr.client.solrj.request.CoreAdminRequest.WaitForState;
-import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventType;
import org.apache.solr.cloud.overseer.OverseerAction;
import org.apache.solr.cloud.overseer.SliceMutator;
import org.apache.solr.common.AlreadyClosedException;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrException.ErrorCode;
-import org.apache.solr.common.cloud.BeforeReconnect;
-import org.apache.solr.common.cloud.ClusterState;
-import org.apache.solr.common.cloud.ConnectionManager;
-import org.apache.solr.common.cloud.DefaultConnectionStrategy;
-import org.apache.solr.common.cloud.DefaultZkACLProvider;
-import org.apache.solr.common.cloud.DefaultZkCredentialsProvider;
-import org.apache.solr.common.cloud.DocCollection;
-import org.apache.solr.common.cloud.DocCollectionWatcher;
-import org.apache.solr.common.cloud.LiveNodesListener;
-import org.apache.solr.common.cloud.NodesSysPropsCacher;
-import org.apache.solr.common.cloud.OnReconnect;
-import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.*;
import org.apache.solr.common.cloud.Replica.Type;
-import org.apache.solr.common.cloud.Slice;
-import org.apache.solr.common.cloud.SolrZkClient;
-import org.apache.solr.common.cloud.ZkACLProvider;
-import org.apache.solr.common.cloud.ZkCmdExecutor;
-import org.apache.solr.common.cloud.ZkConfigManager;
-import org.apache.solr.common.cloud.ZkCoreNodeProps;
-import org.apache.solr.common.cloud.ZkCredentialsProvider;
-import org.apache.solr.common.cloud.ZkMaintenanceUtils;
-import org.apache.solr.common.cloud.ZkNodeProps;
-import org.apache.solr.common.cloud.ZkStateReader;
-import org.apache.solr.common.cloud.ZooKeeperException;
import org.apache.solr.common.params.CollectionParams;
import org.apache.solr.common.params.CommonParams;
import org.apache.solr.common.params.CoreAdminParams;
@@ -1053,7 +1032,7 @@ public class ZkController implements Closeable {
CountDownLatch latch = new CountDownLatch(collectionsWithLocalReplica.size());
for (String collectionWithLocalReplica : collectionsWithLocalReplica) {
- zkStateReader.registerDocCollectionWatcher(collectionWithLocalReplica, (collectionState) -> {
+ zkStateReader.registerDocCollectionWatcher(collectionWithLocalReplica, (collectionState, ssp) -> {
if (collectionState == null) return false;
boolean foundStates = true;
for (CoreDescriptor coreDescriptor : cc.getCoreDescriptors()) {
@@ -2559,7 +2538,7 @@ public class ZkController implements Closeable {
@Override
// synchronized due to SOLR-11535
- public synchronized boolean onStateChanged(DocCollection collectionState) {
+ public synchronized boolean onStateChanged(DocCollection collectionState, ShardStateProvider ssp) {
if (getCoreContainer().getCoreDescriptor(coreName) == null) return true;
boolean replicaRemoved = getReplicaOrNull(collectionState, shard, coreNodeName) == null;
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/LeaderRecoveryWatcher.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/LeaderRecoveryWatcher.java
index a80fdc0..8829fb0 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/LeaderRecoveryWatcher.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/LeaderRecoveryWatcher.java
@@ -18,6 +18,7 @@ package org.apache.solr.cloud.api.collections;
import java.util.Set;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.common.SolrCloseableLatch;
import org.apache.solr.common.cloud.CollectionStateWatcher;
import org.apache.solr.common.cloud.DocCollection;
@@ -53,7 +54,7 @@ public class LeaderRecoveryWatcher implements CollectionStateWatcher {
}
@Override
- public boolean onStateChanged(Set<String> liveNodes, DocCollection collectionState) {
+ public boolean onStateChanged(ShardStateProvider ssp, Set<String> liveNodes, DocCollection collectionState) {
if (collectionState == null) { // collection has been deleted - don't wait
latch.countDown();
return true;
@@ -76,7 +77,7 @@ public class LeaderRecoveryWatcher implements CollectionStateWatcher {
if (targetCore != null && !targetCore.equals(coreName)) {
continue;
}
- if (replica.isActive(liveNodes)) { // recovered - stop waiting
+ if (ssp.isActive(replica)) { // recovered - stop waiting
latch.countDown();
return true;
}
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
index 2a81716..6d731a6 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
@@ -48,6 +48,7 @@ import java.util.stream.Collectors;
import com.google.common.util.concurrent.AtomicDouble;
import org.apache.commons.math3.stat.descriptive.SummaryStatistics;
import org.apache.solr.client.solrj.cloud.DistribStateManager;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
@@ -749,6 +750,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
log.trace("Attempting leader election ({} / {})", collection, slice);
final AtomicBoolean stateChanged = new AtomicBoolean(Boolean.FALSE);
+ ShardStateProvider ssp = getShardStateProvider(collection);
lock.lockInterruptibly();
try {
@@ -776,14 +778,14 @@ public class SimClusterStateProvider implements ClusterStateProvider {
return;
}
- final Replica leader = s.getLeader();
+ final Replica leader = ssp.getLeader(s) ;
if (null != leader && liveNodes.contains(leader.getNodeName())) {
log.trace("-- already has livenode leader, skipping leader election {} / {}",
collection, slice);
return;
}
- if (s.getState() != Slice.State.ACTIVE) {
+ if (ssp.isActive(s)) {
log.trace("-- slice state is {}, but I will run leader election anyway ({} / {})",
s.getState(), collection, slice);
}
@@ -801,7 +803,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
throw new IllegalStateException("-- could not find ReplicaInfo for replica " + r);
}
synchronized (ri) {
- if (r.isActive(liveNodes.get())) {
+ if (ssp.isActive(r)) {
if (ri.getVariables().get(ZkStateReader.LEADER_PROP) != null) {
log.trace("-- found existing leader {} / {}: {}, {}", collection, s.getName(), ri, r);
alreadyHasLeader.set(true);
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimUtils.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimUtils.java
index b7d59de..27d1b4a 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimUtils.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SimUtils.java
@@ -32,6 +32,7 @@ import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
import org.apache.solr.client.solrj.cloud.autoscaling.Cell;
@@ -135,13 +136,14 @@ public class SimUtils {
allReplicaInfos.keySet().forEach(collection -> {
Set<String> infosCores = allReplicaInfos.getOrDefault(collection, Collections.emptyMap()).keySet();
Map<String, Replica> replicas = allReplicas.getOrDefault(collection, Collections.emptyMap());
+ ShardStateProvider ssp = solrCloudManager.getClusterStateProvider().getShardStateProvider(collection);
Set<String> csCores = replicas.keySet();
if (!infosCores.equals(csCores)) {
Set<String> notInClusterState = infosCores.stream()
.filter(k -> !csCores.contains(k))
.collect(Collectors.toSet());
Set<String> notInNodeProvider = csCores.stream()
- .filter(k -> !infosCores.contains(k) && replicas.get(k).isActive(solrCloudManager.getClusterStateProvider().getLiveNodes()))
+ .filter(k -> !infosCores.contains(k) && ssp.isActive(replicas.get(k)))
.collect(Collectors.toSet());
if (!notInClusterState.isEmpty() || !notInNodeProvider.isEmpty()) {
throw new RuntimeException("Mismatched replica data for collection " + collection + " between ClusterState and NodeStateProvider:\n\t" +
diff --git a/solr/core/src/java/org/apache/solr/handler/admin/ColStatus.java b/solr/core/src/java/org/apache/solr/handler/admin/ColStatus.java
index 1fb2533..434905b 100644
--- a/solr/core/src/java/org/apache/solr/handler/admin/ColStatus.java
+++ b/solr/core/src/java/org/apache/solr/handler/admin/ColStatus.java
@@ -30,9 +30,9 @@ import java.util.TreeSet;
import org.apache.http.client.HttpClient;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.client.solrj.io.SolrClientCache;
import org.apache.solr.client.solrj.request.QueryRequest;
-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.RoutingRule;
@@ -53,7 +53,6 @@ import org.slf4j.LoggerFactory;
public class ColStatus {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final ClusterState clusterState;
private final ZkNodeProps props;
private final SolrClientCache solrClientCache;
@@ -65,18 +64,19 @@ public class ColStatus {
public static final String RAW_SIZE_DETAILS_PROP = SegmentsInfoRequestHandler.RAW_SIZE_DETAILS_PARAM;
public static final String RAW_SIZE_SAMPLING_PERCENT_PROP = SegmentsInfoRequestHandler.RAW_SIZE_SAMPLING_PERCENT_PARAM;
public static final String SEGMENTS_PROP = "segments";
+ private final ZkStateReader zkStateReader;
- public ColStatus(HttpClient httpClient, ClusterState clusterState, ZkNodeProps props) {
+ public ColStatus(HttpClient httpClient, ZkStateReader zkStateReader, ZkNodeProps props) {
this.props = props;
this.solrClientCache = new SolrClientCache(httpClient);
- this.clusterState = clusterState;
+ this.zkStateReader = zkStateReader;
}
public void getColStatus(NamedList<Object> results) {
Collection<String> collections;
String col = props.getStr(ZkStateReader.COLLECTION_PROP);
if (col == null) {
- collections = new HashSet<>(clusterState.getCollectionsMap().keySet());
+ collections = new HashSet<>(zkStateReader.getCollections());
} else {
collections = Collections.singleton(col);
}
@@ -96,7 +96,8 @@ public class ColStatus {
withSegments = true;
}
for (String collection : collections) {
- DocCollection coll = clusterState.getCollectionOrNull(collection);
+ ShardStateProvider ssp = zkStateReader.getShardStateProvider(collection);
+ DocCollection coll = zkStateReader.getCollection(collection);
if (coll == null) {
continue;
}
@@ -124,7 +125,7 @@ public class ColStatus {
int recoveryFailedReplicas = 0;
for (Replica r : s.getReplicas()) {
// replica may still be marked as ACTIVE even though its node is no longer live
- if (! r.isActive(clusterState.getLiveNodes())) {
+ if (! ssp.isActive(r)) {
downReplicas++;
continue;
}
@@ -155,7 +156,7 @@ public class ColStatus {
sliceMap.add("routingRules", rules);
}
sliceMap.add("replicas", replicaMap);
- Replica leader = s.getLeader();
+ Replica leader = ssp.getLeader(s);
if (leader == null) { // pick the first one
leader = s.getReplicas().size() > 0 ? s.getReplicas().iterator().next() : null;
}
@@ -166,7 +167,7 @@ public class ColStatus {
sliceMap.add("leader", leaderMap);
leaderMap.add("coreNode", leader.getName());
leaderMap.addAll(leader.getProperties());
- if (!leader.isActive(clusterState.getLiveNodes())) {
+ if (!ssp.isActive(leader)) {
continue;
}
String url = ZkCoreNodeProps.getCoreUrl(leader);
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 6ba09d1..09fc02c 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
@@ -542,7 +542,7 @@ public class CollectionsHandler extends RequestHandlerBase implements Permission
props.put(COLLECTION_PROP, props.get(CoreAdminParams.NAME));
}
new ColStatus(h.coreContainer.getUpdateShardHandler().getDefaultHttpClient(),
- h.coreContainer.getZkController().getZkStateReader().getClusterState(), new ZkNodeProps(props))
+ h.coreContainer.getZkController().getZkStateReader(), new ZkNodeProps(props))
.getColStatus(rsp.getValues());
return null;
}),
diff --git a/solr/core/src/java/org/apache/solr/handler/admin/MetricsHistoryHandler.java b/solr/core/src/java/org/apache/solr/handler/admin/MetricsHistoryHandler.java
index c69f99c..c5bf7a0 100644
--- a/solr/core/src/java/org/apache/solr/handler/admin/MetricsHistoryHandler.java
+++ b/solr/core/src/java/org/apache/solr/handler/admin/MetricsHistoryHandler.java
@@ -59,6 +59,7 @@ import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.cloud.NodeStateProvider;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
import org.apache.solr.client.solrj.cloud.autoscaling.ReplicaInfo;
import org.apache.solr.client.solrj.cloud.autoscaling.Variable;
@@ -258,9 +259,10 @@ public class MetricsHistoryHandler extends RequestHandlerBase implements Permiss
factory.setPersistent(false);
return;
} else {
+ ShardStateProvider ssp = cloudManager.getClusterStateProvider().getShardStateProvider(systemColl.getName());
boolean ready = false;
for (Replica r : systemColl.getReplicas()) {
- if (r.isActive(clusterState.getLiveNodes())) {
+ if (ssp.isActive(r)) {
ready = true;
break;
}
@@ -532,6 +534,7 @@ public class MetricsHistoryHandler extends RequestHandlerBase implements Permiss
try {
ClusterState state = cloudManager.getClusterStateProvider().getClusterState();
state.forEachCollection(coll -> {
+ ShardStateProvider ssp = cloudManager.getClusterStateProvider().getShardStateProvider(coll.getName());
String registry = SolrMetricManager.getRegistryName(Group.collection, coll.getName());
Map<String, Number> perReg = totals
.computeIfAbsent(Group.collection, g -> new HashMap<>())
@@ -541,7 +544,7 @@ public class MetricsHistoryHandler extends RequestHandlerBase implements Permiss
DoubleAdder numActiveReplicas = new DoubleAdder();
for (Slice s : slices) {
s.forEach(r -> {
- if (r.isActive(state.getLiveNodes())) {
+ if (ssp.isActive(r)) {
numActiveReplicas.add(1.0);
}
});
diff --git a/solr/core/src/java/org/apache/solr/handler/admin/RebalanceLeaders.java b/solr/core/src/java/org/apache/solr/handler/admin/RebalanceLeaders.java
index 0fa43a8..83b0550 100644
--- a/solr/core/src/java/org/apache/solr/handler/admin/RebalanceLeaders.java
+++ b/solr/core/src/java/org/apache/solr/handler/admin/RebalanceLeaders.java
@@ -27,6 +27,7 @@ import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.cloud.LeaderElector;
import org.apache.solr.cloud.OverseerTaskProcessor;
import org.apache.solr.cloud.overseer.SliceMutator;
@@ -191,9 +192,10 @@ class RebalanceLeaders {
ClusterState clusterState = coreContainer.getZkController().getClusterState();
Set<String> liveNodes = clusterState.getLiveNodes();
DocCollection dc = clusterState.getCollection(collectionName);
+ ShardStateProvider ssp = coreContainer.getZkController().getZkStateReader().getShardStateProvider(collectionName);
for (Slice slice : dc.getSlices()) {
for (Replica replica : slice.getReplicas()) {
- if (replica.isActive(liveNodes) && replica.getBool(SliceMutator.PREFERRED_LEADER_PROP, false)) {
+ if (ssp.isActive(replica)&& replica.getBool(SliceMutator.PREFERRED_LEADER_PROP, false)) {
if (replica.getBool(LEADER_PROP, false)) {
if (pendingOps.containsKey(slice.getName())) {
// Record for return that the leader changed successfully
@@ -231,8 +233,9 @@ class RebalanceLeaders {
return; // already the leader, do nothing.
}
ZkStateReader zkStateReader = coreContainer.getZkController().getZkStateReader();
+ ShardStateProvider ssp = zkStateReader.getShardStateProvider(slice.getCollection());
// We're the preferred leader, but someone else is leader. Only become leader if we're active.
- if (replica.isActive(zkStateReader.getClusterState().getLiveNodes()) == false) {
+ if (ssp.isActive(replica) == false) {
addInactiveToResults(slice, replica);
return; // Don't try to become the leader if we're not active!
}
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 feeda0c..ad1927c 100644
--- a/solr/core/src/test/org/apache/solr/cloud/CollectionsAPISolrJTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/CollectionsAPISolrJTest.java
@@ -456,7 +456,7 @@ public class CollectionsAPISolrJTest extends SolrCloudTestCase {
if (c.getSlice("shard1").getState() == Slice.State.ACTIVE)
return false;
for (Replica r : c.getReplicas()) {
- if (r.isActive(n) == false)
+ if (!ssp.isActive(r))
return false;
}
return true;
diff --git a/solr/core/src/test/org/apache/solr/cloud/MoveReplicaTest.java b/solr/core/src/test/org/apache/solr/cloud/MoveReplicaTest.java
index 025460c..0a97c49 100644
--- a/solr/core/src/test/org/apache/solr/cloud/MoveReplicaTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/MoveReplicaTest.java
@@ -29,6 +29,7 @@ import java.util.Set;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
@@ -154,6 +155,7 @@ public class MoveReplicaTest extends SolrCloudTestCase {
boolean recovered = false;
for (int i = 0; i < 300; i++) {
DocCollection collState = getCollectionState(coll);
+ ShardStateProvider ssp = cloudClient.getClusterStateProvider().getShardStateProvider(coll);
log.debug("###### " + collState);
Collection<Replica> replicas = collState.getSlice(shardId).getReplicas();
boolean allActive = true;
@@ -163,7 +165,7 @@ public class MoveReplicaTest extends SolrCloudTestCase {
if (!r.getNodeName().equals(targetNode)) {
continue;
}
- if (!r.isActive(Collections.singleton(targetNode))) {
+ if (!ssp.isActive(r)) {
log.info("Not active: " + r);
allActive = false;
}
@@ -198,6 +200,7 @@ public class MoveReplicaTest extends SolrCloudTestCase {
recovered = false;
for (int i = 0; i < 300; i++) {
DocCollection collState = getCollectionState(coll);
+ ShardStateProvider ssp = cloudClient.getClusterStateProvider().getShardStateProvider(coll);
log.debug("###### " + collState);
Collection<Replica> replicas = collState.getSlice(shardId).getReplicas();
boolean allActive = true;
@@ -207,7 +210,7 @@ public class MoveReplicaTest extends SolrCloudTestCase {
if (!r.getNodeName().equals(replica.getNodeName())) {
continue;
}
- if (!r.isActive(Collections.singleton(replica.getNodeName()))) {
+ if (!ssp.isActive(replica)) {
log.info("Not active yet: " + r);
allActive = false;
}
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestCloudSearcherWarming.java b/solr/core/src/test/org/apache/solr/cloud/TestCloudSearcherWarming.java
index a6b97fa..5e5a11e 100644
--- a/solr/core/src/test/org/apache/solr/cloud/TestCloudSearcherWarming.java
+++ b/solr/core/src/test/org/apache/solr/cloud/TestCloudSearcherWarming.java
@@ -23,6 +23,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.solr.client.solrj.SolrQuery;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
@@ -247,7 +248,7 @@ public class TestCloudSearcherWarming extends SolrCloudTestCase {
private CollectionStateWatcher createActiveReplicaSearcherWatcher(AtomicInteger expectedDocs, AtomicReference<String> failingCoreNodeName) {
return new CollectionStateWatcher() {
@Override
- public boolean onStateChanged(Set<String> liveNodes, DocCollection collectionState) {
+ public boolean onStateChanged(ShardStateProvider ssp, Set<String> liveNodes, DocCollection collectionState) {
try {
String coreNodeName = coreNodeNameRef.get();
String coreName = coreNameRef.get();
@@ -255,7 +256,7 @@ public class TestCloudSearcherWarming extends SolrCloudTestCase {
Replica replica = collectionState.getReplica(coreNodeName);
if (replica == null) return false;
log.info("Collection state: {}", collectionState);
- if (replica.isActive(liveNodes)) {
+ if (ssp.isActive(replica)) {
log.info("Active replica: {}", coreNodeName);
for (int i = 0; i < cluster.getJettySolrRunners().size(); i++) {
JettySolrRunner jettySolrRunner = cluster.getJettySolrRunner(i);
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestPullReplica.java b/solr/core/src/test/org/apache/solr/cloud/TestPullReplica.java
index 6c72c5b..e8abffe 100644
--- a/solr/core/src/test/org/apache/solr/cloud/TestPullReplica.java
+++ b/solr/core/src/test/org/apache/solr/cloud/TestPullReplica.java
@@ -39,6 +39,7 @@ import org.apache.lucene.util.LuceneTestCase.AwaitsFix;
import org.apache.lucene.util.LuceneTestCase.Slow;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
@@ -326,7 +327,7 @@ public class TestPullReplica extends SolrCloudTestCase {
waitForState("Replica not added", collectionName, activeReplicaCount(1, 0, 0));
addDocs(500);
List<Replica.State> statesSeen = new ArrayList<>(3);
- cluster.getSolrClient().registerCollectionStateWatcher(collectionName, (liveNodes, collectionState) -> {
+ cluster.getSolrClient().registerCollectionStateWatcher(collectionName, (ssp, liveNodes, collectionState) -> {
Replica r = collectionState.getSlice("shard1").getReplica("core_node2");
log.info("CollectionStateWatcher state change: {}", r);
if (r == null) {
@@ -428,13 +429,14 @@ public class TestPullReplica extends SolrCloudTestCase {
}
docCollection = assertNumberOfReplicas(0, 0, 1, true, true);
+ ShardStateProvider ssp = cluster.getSolrClient().getClusterStateProvider().getShardStateProvider(collectionName);
// Check that there is no leader for the shard
- Replica leader = docCollection.getSlice("shard1").getLeader();
- assertTrue(leader == null || !leader.isActive(cluster.getSolrClient().getZkStateReader().getClusterState().getLiveNodes()));
+ Replica leader = ssp.getLeader(docCollection.getSlice("shard1"));
+ assertTrue(leader == null || !ssp.isActive(leader));
// Pull replica on the other hand should be active
Replica pullReplica = docCollection.getSlice("shard1").getReplicas(EnumSet.of(Replica.Type.PULL)).get(0);
- assertTrue(pullReplica.isActive(cluster.getSolrClient().getZkStateReader().getClusterState().getLiveNodes()));
+ assertTrue(ssp.isActive(pullReplica));
long highestTerm = 0L;
try (ZkShardTerms zkShardTerms = new ZkShardTerms(collectionName, "shard1", zkClient())) {
@@ -478,8 +480,8 @@ public class TestPullReplica extends SolrCloudTestCase {
// Validate that the new nrt replica is the leader now
cluster.getSolrClient().getZkStateReader().forceUpdateCollection(collectionName);
docCollection = getCollectionState(collectionName);
- leader = docCollection.getSlice("shard1").getLeader();
- assertTrue(leader != null && leader.isActive(cluster.getSolrClient().getZkStateReader().getClusterState().getLiveNodes()));
+ leader = ssp.getLeader(docCollection.getSlice("shard1"));
+ assertTrue(leader != null && ssp.isActive(leader));
// If jetty is restarted, the replication is not forced, and replica doesn't replicate from leader until new docs are added. Is this the correct behavior? Why should these two cases be different?
if (removeReplica) {
@@ -621,7 +623,7 @@ public class TestPullReplica extends SolrCloudTestCase {
return false;
for (Slice slice : collectionState) {
for (Replica replica : slice) {
- if (replica.isActive(liveNodes))
+ if (ssp.isActive(replica))
switch (replica.getType()) {
case TLOG:
tlogFound++;
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestPullReplicaErrorHandling.java b/solr/core/src/test/org/apache/solr/cloud/TestPullReplicaErrorHandling.java
index 74e7570..eb4e26a 100644
--- a/solr/core/src/test/org/apache/solr/cloud/TestPullReplicaErrorHandling.java
+++ b/solr/core/src/test/org/apache/solr/cloud/TestPullReplicaErrorHandling.java
@@ -331,7 +331,7 @@ public void testCantConnectToPullReplica() throws Exception {
return false;
for (Slice slice : collectionState) {
for (Replica replica : slice) {
- if (replica.isActive(liveNodes))
+ if (ssp.isActive(replica))
switch (replica.getType()) {
case TLOG:
activesFound++;
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestRebalanceLeaders.java b/solr/core/src/test/org/apache/solr/cloud/TestRebalanceLeaders.java
index 9708ba4..f8ff6d5 100644
--- a/solr/core/src/test/org/apache/solr/cloud/TestRebalanceLeaders.java
+++ b/solr/core/src/test/org/apache/solr/cloud/TestRebalanceLeaders.java
@@ -32,6 +32,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.lucene.util.LuceneTestCase;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.client.solrj.request.QueryRequest;
@@ -226,12 +227,12 @@ public class TestRebalanceLeaders extends SolrCloudTestCase {
private void checkElectionQueues() throws KeeperException, InterruptedException {
DocCollection docCollection = cluster.getSolrClient().getZkStateReader().getClusterState().getCollection(COLLECTION_NAME);
- Set<String> liveNodes = cluster.getSolrClient().getZkStateReader().getClusterState().getLiveNodes();
+ ShardStateProvider ssp = cluster.getSolrClient().getZkStateReader().getShardStateProvider(COLLECTION_NAME);
for (Slice slice : docCollection.getSlices()) {
Set<Replica> liveReplicas = new HashSet<>();
slice.getReplicas().forEach(replica -> {
- if (replica.isActive(liveNodes)) {
+ if (ssp.isActive(replica)) {
liveReplicas.add(replica);
}
});
@@ -533,7 +534,7 @@ public class TestRebalanceLeaders extends SolrCloudTestCase {
while (timeout.hasTimedOut() == false) {
forceUpdateCollectionStatus();
docCollection = cluster.getSolrClient().getZkStateReader().getClusterState().getCollection(COLLECTION_NAME);
- liveNodes = cluster.getSolrClient().getZkStateReader().getClusterState().getLiveNodes();
+ ShardStateProvider ssp = cluster.getSolrClient().getZkStateReader().getShardStateProvider(COLLECTION_NAME);
boolean expectedInactive = true;
for (Slice slice : docCollection.getSlices()) {
@@ -542,7 +543,7 @@ public class TestRebalanceLeaders extends SolrCloudTestCase {
continue; // We are on a live node
}
// A replica on an allegedly down node is reported as active.
- if (rep.isActive(liveNodes)) {
+ if (ssp.isActive(rep)) {
expectedInactive = false;
}
}
@@ -563,11 +564,11 @@ public class TestRebalanceLeaders extends SolrCloudTestCase {
while (timeout.hasTimedOut() == false) {
forceUpdateCollectionStatus();
DocCollection docCollection = cluster.getSolrClient().getZkStateReader().getClusterState().getCollection(COLLECTION_NAME);
- Set<String> liveNodes = cluster.getSolrClient().getZkStateReader().getClusterState().getLiveNodes();
+ ShardStateProvider ssp = cluster.getSolrClient().getZkStateReader().getShardStateProvider(COLLECTION_NAME);
boolean allActive = true;
for (Slice slice : docCollection.getSlices()) {
for (Replica rep : slice.getReplicas()) {
- if (rep.isActive(liveNodes) == false) {
+ if (ssp.isActive(rep) == false) {
allActive = false;
}
}
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestTlogReplica.java b/solr/core/src/test/org/apache/solr/cloud/TestTlogReplica.java
index 1ddb528..5e59b7c 100644
--- a/solr/core/src/test/org/apache/solr/cloud/TestTlogReplica.java
+++ b/solr/core/src/test/org/apache/solr/cloud/TestTlogReplica.java
@@ -43,6 +43,7 @@ import org.apache.lucene.util.LuceneTestCase.Slow;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
@@ -679,8 +680,8 @@ public class TestTlogReplica extends SolrCloudTestCase {
private void waitForLeaderChange(JettySolrRunner oldLeaderJetty, String shardName) {
waitForState("Expect new leader", collectionName,
(liveNodes, collectionState, ssp) -> {
- Replica leader = collectionState.getLeader(shardName);
- if (leader == null || !leader.isActive(cluster.getSolrClient().getZkStateReader().getClusterState().getLiveNodes())) {
+ Replica leader = ssp.getLeader(collectionState.getSlice(shardName));
+ if (leader == null || !ssp.isActive(leader)) {
return false;
}
return oldLeaderJetty == null || !leader.getNodeName().equals(oldLeaderJetty.getNodeName());
@@ -775,7 +776,8 @@ public class TestTlogReplica extends SolrCloudTestCase {
private void waitForNumDocsInAllReplicas(int numDocs, Collection<Replica> replicas, String query, int timeout) throws IOException, SolrServerException, InterruptedException {
TimeOut t = new TimeOut(timeout, TimeUnit.SECONDS, TimeSource.NANO_TIME);
for (Replica r:replicas) {
- if (!r.isActive(cluster.getSolrClient().getZkStateReader().getClusterState().getLiveNodes())) {
+ ShardStateProvider ssp = cluster.getSolrClient().getZkStateReader().getShardStateProvider(r.collection);
+ if (!ssp.isActive(r)) {
continue;
}
try (HttpSolrClient replicaClient = getHttpSolrClient(r.getCoreUrl())) {
@@ -855,7 +857,7 @@ public class TestTlogReplica extends SolrCloudTestCase {
return false;
for (Slice slice : collectionState) {
for (Replica replica : slice) {
- if (replica.isActive(liveNodes))
+ if (ssp.isActive(replica))
switch (replica.getType()) {
case TLOG:
tlogFound++;
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 d5e5f18..95273ea 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
@@ -167,7 +167,7 @@ public class ShardSplitTest extends BasicDistributedZkTest {
waitForRecoveriesToFinish(collectionName, true);
// let's wait to see parent shard become inactive
CountDownLatch latch = new CountDownLatch(1);
- client.getZkStateReader().registerCollectionStateWatcher(collectionName, (liveNodes, collectionState) -> {
+ client.getZkStateReader().registerCollectionStateWatcher(collectionName, (ssp, liveNodes, collectionState) -> {
Slice parent = collectionState.getSlice(SHARD1);
Slice slice10 = collectionState.getSlice(SHARD1_0);
Slice slice11 = collectionState.getSlice(SHARD1_1);
@@ -221,7 +221,7 @@ public class ShardSplitTest extends BasicDistributedZkTest {
if (state == RequestStatusState.COMPLETED) {
CountDownLatch newReplicaLatch = new CountDownLatch(1);
- client.getZkStateReader().registerCollectionStateWatcher(collectionName, (liveNodes, collectionState) -> {
+ client.getZkStateReader().registerCollectionStateWatcher(collectionName, (ssp, liveNodes, collectionState) -> {
if (liveNodes.size() != liveNodeCount) {
return false;
}
@@ -441,7 +441,7 @@ public class ShardSplitTest extends BasicDistributedZkTest {
AtomicBoolean killed = new AtomicBoolean(false);
Runnable monkey = () -> {
ZkStateReader zkStateReader = cloudClient.getZkStateReader();
- zkStateReader.registerCollectionStateWatcher(AbstractDistribZkTestBase.DEFAULT_COLLECTION, (liveNodes, collectionState) -> {
+ zkStateReader.registerCollectionStateWatcher(AbstractDistribZkTestBase.DEFAULT_COLLECTION, (ssp, liveNodes, collectionState) -> {
if (stop.get()) {
return true; // abort and remove the watch
}
@@ -507,7 +507,7 @@ public class ShardSplitTest extends BasicDistributedZkTest {
waitForRecoveriesToFinish(AbstractDistribZkTestBase.DEFAULT_COLLECTION, true);
// let's wait for the overseer to switch shard states
CountDownLatch latch = new CountDownLatch(1);
- cloudClient.getZkStateReader().registerCollectionStateWatcher(AbstractDistribZkTestBase.DEFAULT_COLLECTION, (liveNodes, collectionState) -> {
+ cloudClient.getZkStateReader().registerCollectionStateWatcher(AbstractDistribZkTestBase.DEFAULT_COLLECTION, (ssp, liveNodes, collectionState) -> {
Slice parent = collectionState.getSlice(SHARD1);
Slice slice10 = collectionState.getSlice(SHARD1_0);
Slice slice11 = collectionState.getSlice(SHARD1_1);
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/ComputePlanActionTest.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/ComputePlanActionTest.java
index 8417381..15ae212 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/ComputePlanActionTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/ComputePlanActionTest.java
@@ -361,7 +361,7 @@ public class ComputePlanActionTest extends SolrCloudTestCase {
create.process(solrClient);
waitForState("Timed out waiting for replicas of new collection to be active",
- "testNodeAdded", (liveNodes, collectionState, ssp) -> collectionState.getReplicas().stream().allMatch(replica -> replica.isActive(liveNodes)));
+ "testNodeAdded", (liveNodes, collectionState, ssp) -> collectionState.getReplicas().stream().allMatch(replica -> ssp.isActive(replica)));
// reset to the original policy which has only 1 replica per shard per node
setClusterPolicyCommand = "{" +
@@ -609,7 +609,7 @@ public class ComputePlanActionTest extends SolrCloudTestCase {
waitForState("Timed out waiting for replicas of new collection to be active",
collectionNamePrefix + "_0", (liveNodes, collectionState, ssp) ->
- collectionState.getReplicas().stream().allMatch(replica -> replica.isActive(liveNodes)));
+ collectionState.getReplicas().stream().allMatch(ssp::isActive));
JettySolrRunner newNode = cluster.startJettySolrRunner();
cluster.waitForAllNodes(30);
@@ -636,7 +636,7 @@ public class ComputePlanActionTest extends SolrCloudTestCase {
waitForState("Timed out waiting for replicas of new collection to be active",
collectionNamePrefix + "_" + i, (liveNodes, collectionState, ssp) ->
- collectionState.getReplicas().stream().allMatch(replica -> replica.isActive(liveNodes)));
+ collectionState.getReplicas().stream().allMatch(ssp::isActive));
}
reset();
@@ -707,7 +707,7 @@ public class ComputePlanActionTest extends SolrCloudTestCase {
waitForState("Timed out waiting for replicas of new collection to be active",
collectionNamePrefix + "_0", (liveNodes, collectionState, ssp) ->
- collectionState.getReplicas().stream().allMatch(replica -> replica.isActive(liveNodes)));
+ collectionState.getReplicas().stream().allMatch(ssp::isActive));
cluster.stopJettySolrRunner(newNode);
assertTrue(triggerFiredLatch.await(30, TimeUnit.SECONDS));
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/ScheduledMaintenanceTriggerTest.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/ScheduledMaintenanceTriggerTest.java
index 3c1be6b..7f702f3 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/ScheduledMaintenanceTriggerTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/ScheduledMaintenanceTriggerTest.java
@@ -309,7 +309,7 @@ public class ScheduledMaintenanceTriggerTest extends SolrCloudTestCase {
ClusterState state = cloudManager.getClusterStateProvider().getClusterState();
- CloudUtil.clusterShape(2, 1).matches(state.getLiveNodes(), state.getCollection(collection1), cloudManager.getClusterStateProvider().getReplicaStateProvider(collection1));
+ CloudUtil.clusterShape(2, 1).matches(state.getLiveNodes(), state.getCollection(collection1), cloudManager.getClusterStateProvider().getShardStateProvider(collection1));
}
public static CountDownLatch getTriggerFired() {
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/TestPolicyCloud.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/TestPolicyCloud.java
index 6801a23..0c71bf1 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/TestPolicyCloud.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/TestPolicyCloud.java
@@ -187,7 +187,7 @@ public class TestPolicyCloud extends SolrCloudTestCase {
int actualReplicaCount = 0;
for (Slice slice : collection) {
for (Replica replica : slice) {
- if ( ! (replica.isActive(liveNodes)
+ if ( ! (ssp.isActive(replica)
&& expectedNodeName.equals(replica.getNodeName())) ) {
return false;
}
@@ -278,7 +278,7 @@ public class TestPolicyCloud extends SolrCloudTestCase {
}
// make sure our replicas are fully live...
final List<Replica> liveReplicas = slice.getReplicas
- ((r) -> r.isActive(liveNodes));
+ (ssp::isActive);
if (2 != liveReplicas.size()) {
return false;
}
@@ -316,7 +316,7 @@ public class TestPolicyCloud extends SolrCloudTestCase {
}
// make sure our replicas are fully live...
final List<Replica> liveReplicas = slice.getReplicas
- ((r) -> r.isActive(liveNodes));
+ (ssp::isActive);
if (2 != liveReplicas.size()) {
return false;
}
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimComputePlanAction.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimComputePlanAction.java
index ffe09c5..bc56b28 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimComputePlanAction.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimComputePlanAction.java
@@ -302,7 +302,7 @@ public class TestSimComputePlanAction extends SimSolrCloudTestCase {
create.process(solrClient);
CloudUtil.waitForState(cluster, "Timed out waiting for replicas of new collection to be active",
- "testNodeAdded", (liveNodes, collectionState, ssp) -> collectionState.getReplicas().stream().allMatch(replica -> replica.isActive(liveNodes)));
+ "testNodeAdded", (liveNodes, collectionState, ssp) -> collectionState.getReplicas().stream().allMatch(ssp::isActive));
// reset to the original policy which has only 1 replica per shard per node
setClusterPolicyCommand = "{" +
diff --git a/solr/core/src/test/org/apache/solr/cloud/rule/RulesTest.java b/solr/core/src/test/org/apache/solr/cloud/rule/RulesTest.java
index 224dea2..0da5d89 100644
--- a/solr/core/src/test/org/apache/solr/cloud/rule/RulesTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/rule/RulesTest.java
@@ -123,7 +123,7 @@ public class RulesTest extends SolrCloudTestCase {
if (slice.getName().equals("shard1")) {
// for shard1, we should have 2 fully live replicas
final List<Replica> liveReplicas = slice.getReplicas
- ((r) -> r.isActive(liveNodes));
+ (ssp::isActive);
if (2 != liveReplicas.size()) {
return false;
}
@@ -131,8 +131,7 @@ public class RulesTest extends SolrCloudTestCase {
(Replica::getNodeName).collect(Collectors.toList()));
} else if (slice.getName().equals("shard2")) {
// for shard2, we should have 3 fully live replicas
- final List<Replica> liveReplicas = slice.getReplicas
- ((r) -> r.isActive(liveNodes));
+ final List<Replica> liveReplicas = slice.getReplicas(ssp::isActive);
if (3 != liveReplicas.size()) {
return false;
}
@@ -206,7 +205,7 @@ public class RulesTest extends SolrCloudTestCase {
// (and the contradictory policy was ignored)
return rulesCollection.getReplicas().stream().allMatch
(replica -> (replica.getNodeName().contains(port) &&
- replica.isActive(liveNodes)));
+ ssp.isActive(replica)));
});
}
@@ -249,7 +248,7 @@ public class RulesTest extends SolrCloudTestCase {
// now sanity check that the rules were *obeyed*
return rulesCollection.getReplicas().stream().allMatch
(replica -> (replica.getNodeName().contains(port) &&
- replica.isActive(liveNodes)));
+ ssp.isActive(replica)));
});
}
diff --git a/solr/core/src/test/org/apache/solr/handler/admin/AutoscalingHistoryHandlerTest.java b/solr/core/src/test/org/apache/solr/handler/admin/AutoscalingHistoryHandlerTest.java
index 935d2cd..4284d52 100644
--- a/solr/core/src/test/org/apache/solr/handler/admin/AutoscalingHistoryHandlerTest.java
+++ b/solr/core/src/test/org/apache/solr/handler/admin/AutoscalingHistoryHandlerTest.java
@@ -26,6 +26,7 @@ import java.util.stream.Collectors;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventProcessorStage;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
@@ -354,8 +355,9 @@ public class AutoscalingHistoryHandlerTest extends SolrCloudTestCase {
systemLeaderNodes = Collections.emptySet();
}
String nodeToKill = null;
+ ShardStateProvider ssp = cluster.getSolrClient().getZkStateReader().getShardStateProvider(coll.getName());
for (Replica r : coll.getReplicas()) {
- if (r.isActive(state.getLiveNodes()) &&
+ if (ssp.isActive(r)&&
!r.getNodeName().equals(overseerLeader)) {
if (systemLeaderNodes.contains(r.getNodeName())) {
log.info("--skipping .system leader replica {}", r);
@@ -420,6 +422,7 @@ public class AutoscalingHistoryHandlerTest extends SolrCloudTestCase {
boolean allActive = true;
boolean hasLeaders = true;
DocCollection collState = null;
+ ShardStateProvider ssp = solrClient.getZkStateReader().getShardStateProvider(collection);
for (int i = 0; i < 300; i++) {
ClusterState state = solrClient.getZkStateReader().getClusterState();
collState = getCollectionState(collection);
@@ -430,7 +433,7 @@ public class AutoscalingHistoryHandlerTest extends SolrCloudTestCase {
if (replicas != null && !replicas.isEmpty()) {
for (Replica r : replicas) {
if (state.getLiveNodes().contains(r.getNodeName())) {
- if (!r.isActive(state.getLiveNodes())) {
+ if (!ssp.isActive(r)) {
log.info("Not active: " + r);
allActive = false;
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/DirectShardState.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/DirectShardState.java
index 2a7333b..16b610f 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/DirectShardState.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/DirectShardState.java
@@ -50,4 +50,8 @@ public class DirectShardState implements ShardStateProvider {
isNodeLive.test(replica.getNodeName());
}
+ @Override
+ public boolean isActive(Slice slice) {
+ return slice.getState() == Slice.State.ACTIVE;
+ }
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/ShardStateProvider.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/ShardStateProvider.java
index 277381c..74d9d74 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/ShardStateProvider.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/ShardStateProvider.java
@@ -30,4 +30,6 @@ public interface ShardStateProvider {
Replica getLeader(Slice slice);
boolean isActive(Replica replica);
+
+ boolean isActive(Slice slice);
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/BaseCloudSolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/BaseCloudSolrClient.java
index 0461e67..2f6a48a 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/BaseCloudSolrClient.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/BaseCloudSolrClient.java
@@ -50,6 +50,7 @@ import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.V2RequestSupport;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.client.solrj.request.AbstractUpdateRequest;
import org.apache.solr.client.solrj.request.IsUpdateRequest;
import org.apache.solr.client.solrj.request.RequestWriter;
@@ -617,6 +618,7 @@ public abstract class BaseCloudSolrClient extends SolrClient {
}
private Map<String,List<String>> buildUrlMap(DocCollection col) {
+ ShardStateProvider ssp = getClusterStateProvider().getShardStateProvider(col.getName());
Map<String, List<String>> urlMap = new HashMap<>();
Slice[] slices = col.getActiveSlicesArr();
for (Slice slice : slices) {
@@ -625,7 +627,7 @@ public abstract class BaseCloudSolrClient extends SolrClient {
Replica leader = slice.getLeader();
if (directUpdatesToLeadersOnly && leader == null) {
for (Replica replica : slice.getReplicas(
- replica -> replica.isActive(getClusterStateProvider().getLiveNodes())
+ replica -> ssp.isActive(replica)
&& replica.getType() == Replica.Type.NRT)) {
leader = replica;
break;
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/BaseHttpClusterStateProvider.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/BaseHttpClusterStateProvider.java
index b170de6..d7d9b05 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/BaseHttpClusterStateProvider.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/BaseHttpClusterStateProvider.java
@@ -331,7 +331,7 @@ public abstract class BaseHttpClusterStateProvider implements ClusterStateProvid
private ShardStateProvider shardStateProvider = new DirectShardState(s -> getLiveNodes().contains(s));
@Override
- public ShardStateProvider getReplicaStateProvider(String coll) {
+ public ShardStateProvider getShardStateProvider(String coll) {
return shardStateProvider;
}
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ClusterStateProvider.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ClusterStateProvider.java
index 07dd058..58bf548 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ClusterStateProvider.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ClusterStateProvider.java
@@ -109,7 +109,7 @@ public interface ClusterStateProvider extends SolrCloseable {
*/
String getPolicyNameByCollection(String coll);
- default ShardStateProvider getReplicaStateProvider(String coll) {
+ default ShardStateProvider getShardStateProvider(String coll) {
return new DirectShardState(s -> getLiveNodes().contains(s)) ;
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkClientClusterStateProvider.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkClientClusterStateProvider.java
index e32f4fc..4b93d7c 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkClientClusterStateProvider.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkClientClusterStateProvider.java
@@ -244,7 +244,7 @@ public class ZkClientClusterStateProvider implements ClusterStateProvider {
}
@Override
- public ShardStateProvider getReplicaStateProvider(String coll) {
+ public ShardStateProvider getShardStateProvider(String coll) {
return zkStateReader.getShardStateProvider(coll);
}
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/CollectionStateWatcher.java b/solr/solrj/src/java/org/apache/solr/common/cloud/CollectionStateWatcher.java
index 63bfaf9..f37e245 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/CollectionStateWatcher.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/CollectionStateWatcher.java
@@ -19,6 +19,8 @@ package org.apache.solr.common.cloud;
import java.util.Set;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
+
/**
* Callback registered with {@link ZkStateReader#registerCollectionStateWatcher(String, CollectionStateWatcher)}
* and called whenever there is a change in the collection state <em>or</em> in the list of liveNodes.
@@ -41,6 +43,6 @@ public interface CollectionStateWatcher {
* deleted)
* @return true if the watcher should be removed
*/
- boolean onStateChanged(Set<String> liveNodes, DocCollection collectionState);
+ boolean onStateChanged(ShardStateProvider ssp, Set<String> liveNodes, DocCollection collectionState);
}
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/DocCollectionWatcher.java b/solr/solrj/src/java/org/apache/solr/common/cloud/DocCollectionWatcher.java
index 0d65cfe..ea504bd 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/DocCollectionWatcher.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/DocCollectionWatcher.java
@@ -17,6 +17,8 @@
package org.apache.solr.common.cloud;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
+
/**
* Callback registered with {@link ZkStateReader#registerDocCollectionWatcher(String, DocCollectionWatcher)}
* and called whenever the DocCollection changes.
@@ -35,6 +37,6 @@ public interface DocCollectionWatcher {
* @param collection the new collection state (may be null if the collection has been deleted)
* @return true if the watcher should be removed
*/
- boolean onStateChanged(DocCollection collection);
+ boolean onStateChanged(DocCollection collection, ShardStateProvider ssp);
}
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 bc57176..21d59fe 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
@@ -172,6 +172,7 @@ public class Replica extends ZkNodeProps {
return state;
}
+ @Deprecated
public boolean isActive(Set<String> liveNodes) {
return this.nodeName != null && liveNodes.contains(this.nodeName) && this.state == State.ACTIVE;
}
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 a765807..352a716 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
@@ -263,6 +263,10 @@ public class ZkStateReader implements SolrCloseable {
return new AutoScalingConfig(map);
}
+ public DocCollection getCollection(String coll) {
+ return clusterState.getCollectionOrNull(coll);
+ }
+
private static class CollectionWatch<T> {
int coreRefCount = 0;
@@ -1678,7 +1682,7 @@ public class ZkStateReader implements SolrCloseable {
registerLiveNodesListener(wrapper);
DocCollection state = clusterState.getCollectionOrNull(collection);
- if (stateWatcher.onStateChanged(liveNodes, state) == true) {
+ if (stateWatcher.onStateChanged(getShardStateProvider(collection), liveNodes, state) == true) {
removeCollectionStateWatcher(collection, stateWatcher);
}
}
@@ -1707,7 +1711,7 @@ public class ZkStateReader implements SolrCloseable {
}
DocCollection state = clusterState.getCollectionOrNull(collection);
- if (stateWatcher.onStateChanged(state) == true) {
+ if (stateWatcher.onStateChanged(state, getShardStateProvider(collection)) == true) {
removeDocCollectionWatcher(collection, stateWatcher);
}
}
@@ -1745,9 +1749,9 @@ public class ZkStateReader implements SolrCloseable {
final CountDownLatch latch = new CountDownLatch(1);
waitLatches.add(latch);
AtomicReference<DocCollection> docCollection = new AtomicReference<>();
- CollectionStateWatcher watcher = (n, c) -> {
+ CollectionStateWatcher watcher = (ssp, n, c) -> {
docCollection.set(c);
- boolean matches = predicate.matches(n, c, getShardStateProvider(collection));
+ boolean matches = predicate.matches(n, c, ssp);
if (matches)
latch.countDown();
@@ -1791,7 +1795,7 @@ public class ZkStateReader implements SolrCloseable {
final CountDownLatch latch = new CountDownLatch(1);
waitLatches.add(latch);
AtomicReference<DocCollection> docCollection = new AtomicReference<>();
- DocCollectionWatcher watcher = (c) -> {
+ DocCollectionWatcher watcher = (c, ssp) -> {
docCollection.set(c);
boolean matches = predicate.test(c);
if (matches)
@@ -2046,7 +2050,7 @@ public class ZkStateReader implements SolrCloseable {
});
for (DocCollectionWatcher watcher : watchers) {
try {
- if (watcher.onStateChanged(collectionState)) {
+ if (watcher.onStateChanged(collectionState, getShardStateProvider(collection))) {
removeDocCollectionWatcher(collection, watcher);
}
} catch (Exception exception) {
@@ -2313,8 +2317,8 @@ public class ZkStateReader implements SolrCloseable {
}
@Override
- public boolean onStateChanged(DocCollection collectionState) {
- final boolean result = delegate.onStateChanged(ZkStateReader.this.liveNodes,
+ public boolean onStateChanged(DocCollection collectionState, ShardStateProvider ssp) {
+ final boolean result = delegate.onStateChanged(ssp, ZkStateReader.this.liveNodes,
collectionState);
if (result) {
// it might be a while before live nodes changes, so proactively remove ourselves
@@ -2326,7 +2330,7 @@ public class ZkStateReader implements SolrCloseable {
@Override
public boolean onChange(SortedSet<String> oldLiveNodes, SortedSet<String> newLiveNodes) {
final DocCollection collection = ZkStateReader.this.clusterState.getCollectionOrNull(collectionName);
- final boolean result = delegate.onStateChanged(newLiveNodes, collection);
+ final boolean result = delegate.onStateChanged(getShardStateProvider(collectionName), newLiveNodes, collection);
if (result) {
// it might be a while before collection changes, so proactively remove ourselves
removeDocCollectionWatcher(collectionName, this);
@@ -2334,6 +2338,9 @@ public class ZkStateReader implements SolrCloseable {
return result;
}
}
+ public Set<String> getCollections(){
+ return clusterState.getCollectionStates().keySet();
+ }
public ShardStateProvider getShardStateProvider(String coll){
return directReplicaState;
}
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 d57524e..54db7d7 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
@@ -135,12 +135,12 @@ public class TestCollectionStateWatchers extends SolrCloudTestCase {
// shutdown a node and check that we get notified about the change
final CountDownLatch latch = new CountDownLatch(1);
- client.registerCollectionStateWatcher("testcollection", (liveNodes, collectionState) -> {
+ client.registerCollectionStateWatcher("testcollection", (ssp, liveNodes, collectionState) -> {
int nodesWithActiveReplicas = 0;
log.info("State changed: {}", collectionState);
for (Slice slice : collectionState) {
for (Replica replica : slice) {
- if (replica.isActive(liveNodes))
+ if (ssp.isActive(replica))
nodesWithActiveReplicas++;
}
}
@@ -172,7 +172,7 @@ public class TestCollectionStateWatchers extends SolrCloudTestCase {
.processAndWait(client, MAX_WAIT_TIMEOUT);
final CountDownLatch latch = new CountDownLatch(1);
- client.registerCollectionStateWatcher("currentstate", (n, c) -> {
+ client.registerCollectionStateWatcher("currentstate", (ssp,n, c) -> {
latch.countDown();
return false;
});
@@ -183,7 +183,7 @@ public class TestCollectionStateWatchers extends SolrCloudTestCase {
1, client.getZkStateReader().getStateWatchers("currentstate").size());
final CountDownLatch latch2 = new CountDownLatch(1);
- client.registerCollectionStateWatcher("currentstate", (n, c) -> {
+ client.registerCollectionStateWatcher("currentstate", (ssp,n, c) -> {
latch2.countDown();
return true;
});
diff --git a/solr/solrj/src/test/org/apache/solr/common/cloud/TestDocCollectionWatcher.java b/solr/solrj/src/test/org/apache/solr/common/cloud/TestDocCollectionWatcher.java
index 29ac814..141908d 100644
--- a/solr/solrj/src/test/org/apache/solr/common/cloud/TestDocCollectionWatcher.java
+++ b/solr/solrj/src/test/org/apache/solr/common/cloud/TestDocCollectionWatcher.java
@@ -113,7 +113,7 @@ public class TestDocCollectionWatcher extends SolrCloudTestCase {
.processAndWait(client, MAX_WAIT_TIMEOUT);
final CountDownLatch latch = new CountDownLatch(1);
- client.registerDocCollectionWatcher("currentstate", (c) -> {
+ client.registerDocCollectionWatcher("currentstate", (c, ssp) -> {
latch.countDown();
return false;
});
@@ -124,7 +124,7 @@ public class TestDocCollectionWatcher extends SolrCloudTestCase {
1, client.getZkStateReader().getStateWatchers("currentstate").size());
final CountDownLatch latch2 = new CountDownLatch(1);
- client.registerDocCollectionWatcher("currentstate", (c) -> {
+ client.registerDocCollectionWatcher("currentstate", (c, ssp) -> {
latch2.countDown();
return true;
});
diff --git a/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java b/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java
index 76fd7a1..ee8165e 100644
--- a/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java
+++ b/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java
@@ -46,6 +46,7 @@ import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.client.solrj.cloud.SocketProxy;
import org.apache.solr.client.solrj.embedded.JettyConfig;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
@@ -2224,10 +2225,11 @@ public abstract class AbstractFullDistribZkTestBase extends AbstractDistribZkTes
StringBuilder builder = new StringBuilder();
zkStateReader.forceUpdateCollection(collectionName);
DocCollection collection = zkStateReader.getClusterState().getCollection(collectionName);
+ ShardStateProvider ssp = zkStateReader.getShardStateProvider(collectionName);
for(Slice s:collection.getSlices()) {
Replica leader = s.getLeader();
for (Replica r:s.getReplicas()) {
- if (!r.isActive(zkStateReader.getClusterState().getLiveNodes())) {
+ if (!ssp.isActive(r)) {
builder.append(String.format(Locale.ROOT, "Replica %s not in liveNodes or is not active%s", r.getName(), System.lineSeparator()));
continue;
}
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 9a7952f..a30a9b5 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
@@ -38,6 +38,7 @@ import java.util.function.Predicate;
import org.apache.solr.SolrTestCaseJ4;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.cloud.ShardStateProvider;
import org.apache.solr.client.solrj.embedded.JettyConfig;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
@@ -332,7 +333,7 @@ public class SolrCloudTestCase extends SolrTestCaseJ4 {
return false;
if (collectionState.getSlices().size() != expectedShards)
return false;
- return compareActiveReplicaCountsForShards(expectedReplicas, liveNodes, collectionState);
+ return compareActiveReplicaCountsForShards(ssp, expectedReplicas, liveNodes, collectionState);
};
}
@@ -347,7 +348,7 @@ public class SolrCloudTestCase extends SolrTestCaseJ4 {
log.info("active slice count: " + collectionState.getActiveSlices().size() + " expected:" + expectedShards);
if (collectionState.getActiveSlices().size() != expectedShards)
return false;
- return compareActiveReplicaCountsForShards(expectedReplicas, liveNodes, collectionState);
+ return compareActiveReplicaCountsForShards(ssp, expectedReplicas, liveNodes, collectionState);
};
}
@@ -376,11 +377,11 @@ public class SolrCloudTestCase extends SolrTestCaseJ4 {
};
}
- private static boolean compareActiveReplicaCountsForShards(int expectedReplicas, Set<String> liveNodes, DocCollection collectionState) {
+ private static boolean compareActiveReplicaCountsForShards(ShardStateProvider ssp, int expectedReplicas, Set<String> liveNodes, DocCollection collectionState) {
int activeReplicas = 0;
for (Slice slice : collectionState) {
for (Replica replica : slice) {
- if (replica.isActive(liveNodes)) {
+ if (ssp.isActive(replica)) {
activeReplicas++;
}
}