(solr) 01/01: Removed "state" and "leader" attributes from replicas of PRS collection

[no...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

noble pushed a commit to branch jira/17272
in repository https://gitbox.apache.org/repos/asf/solr.git

commit 24b3708fadc1feee813331bdc18c5ca680c77111
Author: noblepaul <no...@gmail.com>
AuthorDate: Tue May 14 03:10:26 2024 +1000

    Removed "state" and "leader" attributes from replicas of PRS collection
---
 .../apache/solr/cloud/overseer/SliceMutator.java   |  4 ++
 .../java/org/apache/solr/common/cloud/Replica.java | 14 +++++--
 .../cloud/PerReplicaStatesIntegrationTest.java     | 48 ++++++++++++++++++++--
 3 files changed, 58 insertions(+), 8 deletions(-)

diff --git a/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java b/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
index c016fa5489b..d5170c66474 100644
--- a/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
+++ b/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
@@ -159,6 +159,10 @@ public class SliceMutator {
       log.error("Could not mark shard leader for non existing collection: {}", collectionName);
       return ZkStateWriter.NO_OP;
     }
+    if (coll.isPerReplicaState()) {
+      log.debug("Do not mark shard leader for PRS collection: {}", collectionName);
+      return ZkStateWriter.NO_OP;
+    }
 
     Map<String, Slice> slices = coll.getSlicesMap();
     Slice slice = slices.get(sliceName);
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 0d4cd3afffb..ede575f3189 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
@@ -25,6 +25,8 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+
 import org.apache.solr.common.MapWriter;
 import org.apache.solr.common.params.CollectionAdminParams;
 import org.apache.solr.common.util.StrUtils;
@@ -249,7 +251,9 @@ public class Replica extends ZkNodeProps implements MapWriter {
     Objects.requireNonNull(this.collection, "'collection' must not be null");
     Objects.requireNonNull(this.shard, "'shard' must not be null");
     Objects.requireNonNull(this.type, "'type' must not be null");
-    Objects.requireNonNull(this.state, "'state' must not be null");
+    if (perReplicaStatesRef == null) {// PRS collection
+      Objects.requireNonNull(this.state, "'state' must not be null");
+    }
     Objects.requireNonNull(this.node, "'node' must not be null");
 
     String baseUrl = (String) propMap.get(ReplicaStateProps.BASE_URL);
@@ -259,7 +263,9 @@ public class Replica extends ZkNodeProps implements MapWriter {
     propMap.put(ReplicaStateProps.NODE_NAME, node);
     propMap.put(ReplicaStateProps.CORE_NAME, core);
     propMap.put(ReplicaStateProps.TYPE, type.toString());
-    propMap.put(ReplicaStateProps.STATE, state.toString());
+    if (perReplicaStatesRef == null) {//PRS collection
+      propMap.put(ReplicaStateProps.STATE, state.toString());
+    }
   }
 
   public String getCollection() {
@@ -412,8 +418,8 @@ public class Replica extends ZkNodeProps implements MapWriter {
     ew.putIfNotNull(ReplicaStateProps.CORE_NAME, core)
         .putIfNotNull(ReplicaStateProps.NODE_NAME, node)
         .putIfNotNull(ReplicaStateProps.TYPE, type.toString())
-        .putIfNotNull(ReplicaStateProps.STATE, getState().toString())
-        .putIfNotNull(ReplicaStateProps.LEADER, () -> isLeader() ? "true" : null)
+        .putIfNotNull(ReplicaStateProps.STATE, () -> perReplicaStatesRef == null ? getState().toString() : null)
+        .putIfNotNull(ReplicaStateProps.LEADER, () -> perReplicaStatesRef != null || !isLeader() ? null : "true")
         .putIfNotNull(
             ReplicaStateProps.FORCE_SET_STATE, propMap.get(ReplicaStateProps.FORCE_SET_STATE))
         .putIfNotNull(ReplicaStateProps.BASE_URL, propMap.get(ReplicaStateProps.BASE_URL));
diff --git a/solr/solrj/src/test/org/apache/solr/common/cloud/PerReplicaStatesIntegrationTest.java b/solr/solrj/src/test/org/apache/solr/common/cloud/PerReplicaStatesIntegrationTest.java
index e75100f1a50..3512e74c6f3 100644
--- a/solr/solrj/src/test/org/apache/solr/common/cloud/PerReplicaStatesIntegrationTest.java
+++ b/solr/solrj/src/test/org/apache/solr/common/cloud/PerReplicaStatesIntegrationTest.java
@@ -40,6 +40,7 @@ import org.slf4j.LoggerFactory;
 /** This test would be faster if we simulated the zk state instead. */
 @LogLevel(
     "org.apache.solr.common.cloud.ZkStateReader=DEBUG;"
+        + "org.apache.solr.cloud.overseer.ZkStateWriter=DEBUG;"
         + "org.apache.solr.handler.admin.CollectionsHandler=DEBUG;"
         + "org.apache.solr.common.cloud.PerReplicaStatesOps=DEBUG;"
         + "org.apache.solr.cloud.Overseer=INFO;"
@@ -315,7 +316,8 @@ public class PerReplicaStatesIntegrationTest extends SolrCloudTestCase {
       CollectionAdminRequest.createCollection(PRS_COLL, "conf", 10, 1)
           .setPerReplicaState(Boolean.TRUE)
           .process(cluster.getSolrClient());
-      stat = cluster.getZkClient().exists(DocCollection.getCollectionPath(PRS_COLL), null, true);
+      String PRS_PATH = DocCollection.getCollectionPath(PRS_COLL);
+      stat = cluster.getZkClient().exists(PRS_PATH, null, true);
       // +1 after all replica are added with on state.json write to CreateCollectionCmd.setData()
       assertEquals(1, stat.getVersion());
       // For each replica:
@@ -330,7 +332,7 @@ public class PerReplicaStatesIntegrationTest extends SolrCloudTestCase {
           CollectionAdminRequest.addReplicaToShard(PRS_COLL, "shard1")
               .process(cluster.getSolrClient());
       cluster.waitForActiveCollection(PRS_COLL, 10, 11);
-      stat = cluster.getZkClient().exists(DocCollection.getCollectionPath(PRS_COLL), null, true);
+      stat = cluster.getZkClient().exists(PRS_PATH, null, true);
       // For the new replica:
       // +2 for state.json overseer writes, even though there's no longer PRS updates from
       // overseer, current code would still do a "TOUCH" on the PRS entry
@@ -350,7 +352,7 @@ public class PerReplicaStatesIntegrationTest extends SolrCloudTestCase {
       CollectionAdminRequest.deleteReplica(PRS_COLL, "shard1", addedReplica.getName())
           .process(cluster.getSolrClient());
       cluster.waitForActiveCollection(PRS_COLL, 10, 10);
-      stat = cluster.getZkClient().exists(DocCollection.getCollectionPath(PRS_COLL), null, true);
+      stat = cluster.getZkClient().exists(PRS_PATH, null, true);
       // For replica deletion
       // +1 for ZkController#unregister, which delete the PRS entry from data node
       // overseer, current code would still do a "TOUCH" on the PRS entry
@@ -359,11 +361,49 @@ public class PerReplicaStatesIntegrationTest extends SolrCloudTestCase {
       for (JettySolrRunner j : cluster.getJettySolrRunners()) {
         j.stop();
         j.start(true);
-        stat = cluster.getZkClient().exists(DocCollection.getCollectionPath(PRS_COLL), null, true);
+        stat = cluster.getZkClient().exists(PRS_PATH, null, true);
         // ensure restart does not update the state.json, after addReplica/deleteReplica, 2 more
         // updates hence at version 3 on state.json version
         assertEquals(3, stat.getVersion());
       }
+
+      // test for leader election
+      Replica leader =
+          cluster.getZkStateReader().clusterState.getCollection(PRS_COLL).getLeader("shard2");
+
+      JettySolrRunner j2 = cluster.startJettySolrRunner();
+      response =
+          CollectionAdminRequest.addReplicaToShard(PRS_COLL, "shard2")
+              .setNode(j2.getNodeName())
+              .process(cluster.getSolrClient());
+
+      // wait for the new replica to be active
+      cluster.waitForActiveCollection(PRS_COLL, 10, 11);
+      stat = cluster.getZkClient().exists(PRS_PATH, null, true);
+      // +1 for a new replica
+      assertEquals(4, stat.getVersion());
+      DocCollection c = cluster.getZkStateReader().getCollection(PRS_COLL);
+      Replica newreplica = c.getReplica((s, replica) -> replica.node.equals(j2.getNodeName()));
+
+      // let's stop the old leader
+      JettySolrRunner oldJetty = cluster.getReplicaJetty(leader);
+      oldJetty.stop();
+
+      cluster
+          .getZkStateReader()
+          .waitForState(
+              PRS_COLL,
+              10,
+              TimeUnit.SECONDS,
+              (liveNodes, collectionState) ->
+                  PerReplicaStatesOps.fetch(PRS_PATH, cluster.getZkClient(), null)
+                      .states
+                      .get(newreplica.name)
+                      .isLeader);
+      PerReplicaStates prs = PerReplicaStatesOps.fetch(PRS_PATH, cluster.getZkClient(), null);
+      stat = cluster.getZkClient().exists(PRS_PATH, null, true);
+      // the version should not have updated
+      assertEquals(4, stat.getVersion());
     } finally {
       cluster.shutdown();
     }