You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@solr.apache.org by ho...@apache.org on 2024/05/14 19:27:11 UTC
(solr) branch branch_9x updated: SOLR-17049: Fix Replica Down on startup logic (#2432)
This is an automated email from the ASF dual-hosted git repository.
houston pushed a commit to branch branch_9x
in repository https://gitbox.apache.org/repos/asf/solr.git
The following commit(s) were added to refs/heads/branch_9x by this push:
new e04095c5bf4 SOLR-17049: Fix Replica Down on startup logic (#2432)
e04095c5bf4 is described below
commit e04095c5bf46bd668f2e42876b15f8f1e8025636
Author: Houston Putman <ho...@apache.org>
AuthorDate: Tue May 14 15:16:46 2024 -0400
SOLR-17049: Fix Replica Down on startup logic (#2432)
(cherry picked from commit 1b582e90de228bd5db3f0e884f0bcb7ec3de2868)
---
solr/CHANGES.txt | 3 +
.../java/org/apache/solr/cloud/ZkController.java | 66 ++++++-------
.../apache/solr/cloud/overseer/NodeMutator.java | 46 +++------
.../solr/servlet/CoordinatorHttpSolrCall.java | 8 +-
.../apache/solr/cloud/ClusterStateMockUtil.java | 15 ++-
.../org/apache/solr/cloud/ZkControllerTest.java | 106 +++++++++++++++++----
.../org/apache/solr/common/cloud/ClusterState.java | 46 ++++++---
.../java/org/apache/solr/common/cloud/Replica.java | 8 +-
8 files changed, 182 insertions(+), 116 deletions(-)
diff --git a/solr/CHANGES.txt b/solr/CHANGES.txt
index bc4fafcd922..4333199978c 100644
--- a/solr/CHANGES.txt
+++ b/solr/CHANGES.txt
@@ -34,6 +34,9 @@ Bug Fixes
* SOLR-17261: Remove unintended timeout of 60 seconds for core loading. (Houston Putman)
+* SOLR-17049: Actually mark all replicas down at startup and truly wait for them.
+ This includes replicas that might not exist anymore locally. (Houston Putman, Vincent Primault)
+
Dependency Upgrades
---------------------
(No changes)
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 09bd049bab4..d982df85e5d 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkController.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
@@ -43,6 +43,7 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
@@ -54,6 +55,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
+import java.util.stream.Collectors;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
import org.apache.solr.client.solrj.impl.CloudLegacySolrClient;
@@ -1102,15 +1104,12 @@ public class ZkController implements Closeable {
publishAndWaitForDownStates(WAIT_DOWN_STATES_TIMEOUT_SECONDS);
}
- public void publishAndWaitForDownStates(int timeoutSeconds)
- throws KeeperException, InterruptedException {
-
- publishNodeAsDown(getNodeName());
+ public void publishAndWaitForDownStates(int timeoutSeconds) throws InterruptedException {
+ final String nodeName = getNodeName();
- Set<String> collectionsWithLocalReplica = ConcurrentHashMap.newKeySet();
- for (CoreDescriptor descriptor : cc.getCoreDescriptors()) {
- collectionsWithLocalReplica.add(descriptor.getCloudDescriptor().getCollectionName());
- }
+ Collection<String> collectionsWithLocalReplica = publishNodeAsDown(nodeName);
+ Map<String, Boolean> collectionsAlreadyVerified =
+ new ConcurrentHashMap<>(collectionsWithLocalReplica.size());
CountDownLatch latch = new CountDownLatch(collectionsWithLocalReplica.size());
for (String collectionWithLocalReplica : collectionsWithLocalReplica) {
@@ -1118,25 +1117,17 @@ public class ZkController implements Closeable {
collectionWithLocalReplica,
(collectionState) -> {
if (collectionState == null) return false;
- boolean foundStates = true;
- for (CoreDescriptor coreDescriptor : cc.getCoreDescriptors()) {
- if (coreDescriptor
- .getCloudDescriptor()
- .getCollectionName()
- .equals(collectionWithLocalReplica)) {
- Replica replica =
- collectionState.getReplica(
- coreDescriptor.getCloudDescriptor().getCoreNodeName());
- if (replica == null || replica.getState() != Replica.State.DOWN) {
- foundStates = false;
- }
- }
- }
-
- if (foundStates && collectionsWithLocalReplica.remove(collectionWithLocalReplica)) {
+ boolean allStatesCorrect =
+ Optional.ofNullable(collectionState.getReplicas(nodeName)).stream()
+ .flatMap(List::stream)
+ .allMatch(replica -> replica.getState() == Replica.State.DOWN);
+
+ if (allStatesCorrect
+ && collectionsAlreadyVerified.putIfAbsent(collectionWithLocalReplica, true)
+ == null) {
latch.countDown();
}
- return foundStates;
+ return allStatesCorrect;
});
}
@@ -2892,9 +2883,14 @@ public class ZkController implements Closeable {
* Best effort to set DOWN state for all replicas on node.
*
* @param nodeName to operate on
+ * @return the names of the collections that have replicas on the given node
*/
- public void publishNodeAsDown(String nodeName) {
+ public Collection<String> publishNodeAsDown(String nodeName) {
log.info("Publish node={} as DOWN", nodeName);
+
+ ClusterState clusterState = getClusterState();
+ Map<String, List<Replica>> replicasPerCollectionOnNode =
+ clusterState.getReplicaNamesPerCollectionOnNode(nodeName);
if (distributedClusterStateUpdater.isDistributedStateUpdate()) {
// Note that with the current implementation, when distributed cluster state updates are
// enabled, we mark the node down synchronously from this thread, whereas the Overseer cluster
@@ -2905,24 +2901,15 @@ public class ZkController implements Closeable {
distributedClusterStateUpdater.executeNodeDownStateUpdate(nodeName, zkStateReader);
} else {
try {
- // Create a concurrently accessible set to avoid repeating collections
- Set<String> processedCollections = new HashSet<>();
- for (CoreDescriptor cd : cc.getCoreDescriptors()) {
- String collName = cd.getCollectionName();
+ for (String collName : replicasPerCollectionOnNode.keySet()) {
DocCollection coll;
if (collName != null
- && processedCollections.add(collName)
&& (coll = zkStateReader.getCollection(collName)) != null
&& coll.isPerReplicaState()) {
- final List<String> replicasToDown = new ArrayList<>(coll.getSlicesMap().size());
- coll.forEachReplica(
- (s, replica) -> {
- if (replica.getNodeName().equals(nodeName)) {
- replicasToDown.add(replica.getName());
- }
- });
PerReplicaStatesOps.downReplicas(
- replicasToDown,
+ replicasPerCollectionOnNode.get(collName).stream()
+ .map(Replica::getName)
+ .collect(Collectors.toList()),
PerReplicaStatesOps.fetch(
coll.getZNode(), zkClient, coll.getPerReplicaStates()))
.persist(coll.getZNode(), zkClient);
@@ -2947,6 +2934,7 @@ public class ZkController implements Closeable {
log.warn("Could not publish node as down: ", e);
}
}
+ return replicasPerCollectionOnNode.keySet();
}
/**
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 8afcd805233..f54d2569ef5 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
@@ -18,11 +18,9 @@ package org.apache.solr.cloud.overseer;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.Optional;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
import org.apache.solr.common.cloud.ClusterState;
@@ -82,39 +80,21 @@ public class NodeMutator {
String nodeName, String collectionName, DocCollection docCollection, SolrZkClient client) {
boolean needToUpdateCollection = false;
List<String> downedReplicas = new ArrayList<>();
- Map<String, Slice> slicesCopy = new LinkedHashMap<>(docCollection.getSlicesMap());
+ final Map<String, Slice> slicesCopy = new LinkedHashMap<>(docCollection.getSlicesMap());
- for (Entry<String, Slice> sliceEntry : slicesCopy.entrySet()) {
- Slice slice = sliceEntry.getValue();
- Map<String, Replica> newReplicas = slice.getReplicasCopy();
-
- Collection<Replica> replicas = slice.getReplicas();
- for (Replica replica : replicas) {
- String rNodeName = replica.getNodeName();
- if (rNodeName == null) {
- throw new RuntimeException("Replica without node name! " + replica);
- }
- if (rNodeName.equals(nodeName)) {
- log.debug("Update replica state for {} to {}", replica, Replica.State.DOWN);
- Map<String, Object> props = replica.shallowCopy();
- Replica newReplica =
- new Replica(
- replica.getName(),
- replica.node,
- replica.collection,
- slice.getName(),
- replica.core,
- Replica.State.DOWN,
- replica.type,
- props);
- newReplicas.put(replica.getName(), newReplica);
- needToUpdateCollection = true;
- downedReplicas.add(replica.getName());
- }
+ List<Replica> replicasOnNode = docCollection.getReplicas(nodeName);
+ if (replicasOnNode == null || replicasOnNode.isEmpty()) {
+ return Optional.empty();
+ }
+ for (Replica replica : replicasOnNode) {
+ if (replica.getState() != Replica.State.DOWN) {
+ log.debug("Update replica state for {} to {}", replica, Replica.State.DOWN);
+ needToUpdateCollection = true;
+ downedReplicas.add(replica.getName());
+ slicesCopy.computeIfPresent(
+ replica.getShard(),
+ (name, slice) -> slice.copyWith(replica.copyWith(Replica.State.DOWN)));
}
-
- Slice newSlice = new Slice(slice.getName(), newReplicas, slice.shallowCopy(), collectionName);
- sliceEntry.setValue(newSlice);
}
if (needToUpdateCollection) {
diff --git a/solr/core/src/java/org/apache/solr/servlet/CoordinatorHttpSolrCall.java b/solr/core/src/java/org/apache/solr/servlet/CoordinatorHttpSolrCall.java
index aa4a0e2fd75..ec00107b717 100644
--- a/solr/core/src/java/org/apache/solr/servlet/CoordinatorHttpSolrCall.java
+++ b/solr/core/src/java/org/apache/solr/servlet/CoordinatorHttpSolrCall.java
@@ -163,8 +163,12 @@ public class CoordinatorHttpSolrCall extends HttpSolrCall {
10,
TimeUnit.SECONDS,
docCollection -> {
- for (Replica nodeNameSyntheticReplica :
- docCollection.getReplicas(solrCall.cores.getZkController().getNodeName())) {
+ List<Replica> replicas =
+ docCollection.getReplicas(solrCall.cores.getZkController().getNodeName());
+ if (replicas == null || replicas.isEmpty()) {
+ return false;
+ }
+ for (Replica nodeNameSyntheticReplica : replicas) {
if (nodeNameSyntheticReplica.getState() == Replica.State.ACTIVE) {
return true;
}
diff --git a/solr/core/src/test/org/apache/solr/cloud/ClusterStateMockUtil.java b/solr/core/src/test/org/apache/solr/cloud/ClusterStateMockUtil.java
index 674fe60b8a6..51a3c3263f5 100644
--- a/solr/core/src/test/org/apache/solr/cloud/ClusterStateMockUtil.java
+++ b/solr/core/src/test/org/apache/solr/cloud/ClusterStateMockUtil.java
@@ -111,7 +111,6 @@ public class ClusterStateMockUtil {
public static ZkStateReader buildClusterState(
String clusterDescription, int replicationFactor, String... liveNodes) {
Map<String, Slice> slices = null;
- Map<String, Replica> replicas = null;
Map<String, Object> collectionProps = new HashMap<>();
collectionProps.put(ZkStateReader.REPLICATION_FACTOR, Integer.toString(replicationFactor));
Map<String, DocCollection> collectionStates = new HashMap<>();
@@ -138,9 +137,9 @@ public class ClusterStateMockUtil {
collectionStates.put(docCollection.getName(), docCollection);
break;
case "s":
- replicas = new HashMap<>();
if (collName == null) collName = "collection" + (collectionStates.size() + 1);
- slice = new Slice(sliceName = "slice" + (slices.size() + 1), replicas, null, collName);
+ slice =
+ new Slice(sliceName = "slice" + (slices.size() + 1), new HashMap<>(), null, collName);
slices.put(slice.getName(), slice);
// hack alert: the DocCollection constructor copies over active slices to its active slice
@@ -168,7 +167,7 @@ public class ClusterStateMockUtil {
// O(n^2) alert! but this is for mocks and testing so shouldn't be used for very large
// cluster states
boolean leaderFound = false;
- for (Map.Entry<String, Replica> entry : replicas.entrySet()) {
+ for (Map.Entry<String, Replica> entry : slice.getReplicasMap().entrySet()) {
Replica value = entry.getValue();
if ("true".equals(value.get(ReplicaStateProps.LEADER))) {
leaderFound = true;
@@ -178,15 +177,13 @@ public class ClusterStateMockUtil {
if (!leaderFound && !m.group(1).equals("p")) {
replicaPropMap.put(ReplicaStateProps.LEADER, "true");
}
- replica = new Replica(replicaName, replicaPropMap, collName, sliceName);
- replicas.put(replica.getName(), replica);
// hack alert: re-create slice with existing data and new replicas map so that it updates
// its internal leader attribute
- slice = new Slice(slice.getName(), replicas, null, collName);
+ slice = slice.copyWith(new Replica(replicaName, replicaPropMap, collName, sliceName));
slices.put(slice.getName(), slice);
- // we don't need to update doc collection again because we aren't adding a new slice or
- // changing its state
+ docCollection = docCollection.copyWithSlices(slices);
+ collectionStates.put(docCollection.getName(), docCollection);
break;
default:
break;
diff --git a/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java b/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java
index 0f0b4ef28cf..06ca5cb033e 100644
--- a/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/ZkControllerTest.java
@@ -23,8 +23,12 @@ import static org.mockito.Mockito.mock;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
+import java.time.Duration;
+import java.time.Instant;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
+import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
@@ -34,7 +38,9 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.solr.SolrTestCaseJ4;
import org.apache.solr.common.MapWriter;
import org.apache.solr.common.cloud.ClusterProperties;
+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.SolrZkClient;
import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
@@ -55,21 +61,15 @@ import org.apache.solr.update.UpdateShardHandlerConfig;
import org.apache.solr.util.LogLevel;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
+import org.hamcrest.Matchers;
import org.junit.Test;
@SolrTestCaseJ4.SuppressSSL
-public class ZkControllerTest extends SolrTestCaseJ4 {
+public class ZkControllerTest extends SolrCloudTestCase {
static final int TIMEOUT = 10000;
- @BeforeClass
- public static void beforeClass() {}
-
- @AfterClass
- public static void afterClass() {}
-
+ @Test
public void testNodeNameUrlConversion() throws Exception {
// nodeName from parts
@@ -186,6 +186,7 @@ public class ZkControllerTest extends SolrTestCaseJ4 {
}
}
+ @Test
public void testGetHostName() throws Exception {
Path zkDir = createTempDir("zkData");
@@ -215,6 +216,7 @@ public class ZkControllerTest extends SolrTestCaseJ4 {
}
@LogLevel(value = "org.apache.solr.cloud=DEBUG;org.apache.solr.cloud.overseer=DEBUG")
+ @Test
public void testPublishAndWaitForDownStates() throws Exception {
/*
@@ -232,9 +234,8 @@ public class ZkControllerTest extends SolrTestCaseJ4 {
String nodeName = "127.0.0.1:8983_solr";
- ZkTestServer server = new ZkTestServer(zkDir);
try {
- server.run();
+ cluster = configureCluster(1).configure();
AtomicReference<ZkController> zkControllerRef = new AtomicReference<>();
CoreContainer cc =
@@ -259,9 +260,15 @@ public class ZkControllerTest extends SolrTestCaseJ4 {
try {
CloudConfig cloudConfig =
- new CloudConfig.CloudConfigBuilder("127.0.0.1", 8983, "solr").build();
+ new CloudConfig.CloudConfigBuilder("127.0.0.1", 8983, "solr")
+ .setUseDistributedClusterStateUpdates(
+ Boolean.getBoolean("solr.distributedClusterStateUpdates"))
+ .setUseDistributedCollectionConfigSetExecution(
+ Boolean.getBoolean("solr.distributedCollectionConfigSetExecution"))
+ .build();
zkController =
- new ZkController(cc, server.getZkAddress(), TIMEOUT, cloudConfig, () -> null);
+ new ZkController(
+ cc, cluster.getZkServer().getZkAddress(), TIMEOUT, cloudConfig, () -> null);
zkControllerRef.set(zkController);
zkController
@@ -294,6 +301,7 @@ public class ZkControllerTest extends SolrTestCaseJ4 {
zkController.getOverseerJobQueue().offer(Utils.toJSON(m));
}
+ // Add an active replica that shares the same core name, but on a non existent host
MapWriter propMap =
ew ->
ew.put(Overseer.QUEUE_OPERATION, ADDREPLICA.toLower())
@@ -315,6 +323,7 @@ public class ZkControllerTest extends SolrTestCaseJ4 {
zkController.getOverseerJobQueue().offer(propMap);
}
+ // Add an down replica that shares the same core name, also on a non existent host
propMap =
ew ->
ew.put(Overseer.QUEUE_OPERATION, ADDREPLICA.toLower())
@@ -335,20 +344,77 @@ public class ZkControllerTest extends SolrTestCaseJ4 {
zkController.getOverseerJobQueue().offer(propMap);
}
- zkController.getZkStateReader().forciblyRefreshAllClusterStateSlow();
+ // Add an active replica on the existing host. This replica will exist in the cluster state
+ // but not
+ // on the disk. We are testing that this replica is also put to "DOWN" even though it
+ // doesn't exist locally.
+ propMap =
+ ew ->
+ ew.put(Overseer.QUEUE_OPERATION, ADDREPLICA.toLower())
+ .put(COLLECTION_PROP, collectionName)
+ .put(SHARD_ID_PROP, "shard1")
+ .put(ZkStateReader.NODE_NAME_PROP, nodeName)
+ .put(ZkStateReader.CORE_NAME_PROP, collectionName + "-not-on-disk")
+ .put(ZkStateReader.STATE_PROP, "active");
+ if (zkController.getDistributedClusterStateUpdater().isDistributedStateUpdate()) {
+ zkController
+ .getDistributedClusterStateUpdater()
+ .doSingleStateUpdate(
+ DistributedClusterStateUpdater.MutatingCommand.SliceAddReplica,
+ new ZkNodeProps(propMap),
+ zkController.getSolrCloudManager(),
+ zkController.getZkStateReader());
+ } else {
+ zkController.getOverseerJobQueue().offer(propMap);
+ }
- long now = System.nanoTime();
- long timeout = now + TimeUnit.NANOSECONDS.convert(5, TimeUnit.SECONDS);
+ // Wait for the overseer to process all the replica additions
+ if (!zkController.getDistributedClusterStateUpdater().isDistributedStateUpdate()) {
+ zkController
+ .getZkStateReader()
+ .waitForState(
+ collectionName,
+ 10,
+ TimeUnit.SECONDS,
+ ((liveNodes, collectionState) ->
+ Optional.ofNullable(collectionState)
+ .map(DocCollection::getReplicas)
+ .map(List::size)
+ .orElse(0)
+ == 3));
+ }
+
+ Instant now = Instant.now();
zkController.publishAndWaitForDownStates(5);
- assertTrue(
- "The ZkController.publishAndWaitForDownStates should have timed out but it didn't",
- System.nanoTime() >= timeout);
+ assertThat(
+ "The ZkController.publishAndWaitForDownStates should not have timed out but it did",
+ Duration.between(now, Instant.now()),
+ Matchers.lessThanOrEqualTo(Duration.ofSeconds(5)));
+
+ zkController.getZkStateReader().forciblyRefreshAllClusterStateSlow();
+ ClusterState clusterState = zkController.getClusterState();
+
+ Map<String, List<Replica>> replicasOnNode =
+ clusterState.getReplicaNamesPerCollectionOnNode(nodeName);
+ assertNotNull("There should be replicas on the existing node", replicasOnNode);
+ List<Replica> replicas = replicasOnNode.get(collectionName);
+ assertNotNull("There should be replicas for the collection on the existing node", replicas);
+ assertEquals(
+ "Wrong number of replicas for the collection on the existing node", 1, replicas.size());
+ for (Replica replica : replicas) {
+ assertEquals(
+ "Replica "
+ + replica.getName()
+ + " is not DOWN, even though it is on the node that should be DOWN",
+ Replica.State.DOWN,
+ replica.getState());
+ }
} finally {
if (zkController != null) zkController.close();
cc.shutdown();
}
} finally {
- server.shutdown();
+ cluster.shutdown();
}
}
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 83898f57631..5f7a23f2ed0 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
@@ -25,8 +25,11 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
@@ -176,34 +179,53 @@ public class ClusterState implements MapWriter {
return Collections.unmodifiableSet(liveNodes);
}
+ @Deprecated
public String getShardId(String nodeName, String coreName) {
return getShardId(null, nodeName, coreName);
}
+ @Deprecated
public String getShardId(String collectionName, String nodeName, String coreName) {
- Collection<CollectionRef> states = collectionStates.values();
+ if (coreName == null || nodeName == null) {
+ return null;
+ }
+ Collection<CollectionRef> states = Collections.emptyList();
if (collectionName != null) {
CollectionRef c = collectionStates.get(collectionName);
if (c != null) states = Collections.singletonList(c);
+ } else {
+ states = collectionStates.values();
}
for (CollectionRef ref : states) {
DocCollection coll = ref.get();
- if (coll == null) continue; // this collection go tremoved in between, skip
- for (Slice slice : coll.getSlices()) {
- for (Replica replica : slice.getReplicas()) {
- // TODO: for really large clusters, we could 'index' on this
- String rnodeName = replica.getStr(ReplicaStateProps.NODE_NAME);
- String rcore = replica.getStr(ReplicaStateProps.CORE_NAME);
- if (nodeName.equals(rnodeName) && coreName.equals(rcore)) {
- return slice.getName();
- }
- }
- }
+ if (coll == null) continue; // this collection got removed in between, skip
+ // TODO: for really large clusters, we could 'index' on this
+ return Optional.ofNullable(coll.getReplicas(nodeName)).stream()
+ .flatMap(List::stream)
+ .filter(r -> coreName.equals(r.getStr(ReplicaStateProps.CORE_NAME)))
+ .map(Replica::getShard)
+ .findAny()
+ .orElse(null);
}
return null;
}
+ public Map<String, List<Replica>> getReplicaNamesPerCollectionOnNode(final String nodeName) {
+ Map<String, List<Replica>> replicaNamesPerCollectionOnNode = new HashMap<>();
+ collectionStates.values().stream()
+ .map(CollectionRef::get)
+ .filter(Objects::nonNull)
+ .forEach(
+ col -> {
+ List<Replica> replicas = col.getReplicas(nodeName);
+ if (replicas != null && !replicas.isEmpty()) {
+ replicaNamesPerCollectionOnNode.put(col.getName(), replicas);
+ }
+ });
+ return replicaNamesPerCollectionOnNode;
+ }
+
/** Check if node is alive. */
public boolean liveNodesContain(String name) {
return liveNodes.contains(name);
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 72bb2879662..91c8ba45fdc 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
@@ -346,7 +346,7 @@ public class Replica extends ZkNodeProps implements MapWriter {
}
public Replica copyWith(PerReplicaStates.State state) {
- log.debug("A replica is updated with new state : {}", state);
+ log.debug("A replica is updated with new PRS state : {}", state);
Map<String, Object> props = new LinkedHashMap<>(propMap);
if (state == null) {
props.put(ReplicaStateProps.STATE, State.DOWN.toString());
@@ -359,6 +359,12 @@ public class Replica extends ZkNodeProps implements MapWriter {
return r;
}
+ public Replica copyWith(State state) {
+ Replica r = new Replica(name, propMap, collection, shard);
+ r.setState(state);
+ return r;
+ }
+
public PerReplicaStates.State getReplicaState() {
if (perReplicaStatesRef != null) {
return perReplicaStatesRef.get().get(name);