SOLR-16116: Use apache curator to manage the Solr Zookeeper interactions

solr/core/src/java/org/apache/solr/cloud/CloudUtil.java

@@ -146,10 +146,10 @@ public static String unifiedResourcePath(SolrResourceLoader loader) {
   public static Map<String, byte[]> getTrustedKeys(SolrZkClient zk, String dir) {
     Map<String, byte[]> result = new HashMap<>();
     try {
-      List<String> children = zk.getChildren("/keys/" + dir, null, true);
+      List<String> children = zk.getChildren("/keys/" + dir, null);
       for (String key : children) {
         if (key.endsWith(".der"))
-          result.put(key, zk.getData("/keys/" + dir + "/" + key, null, null, true));
+          result.put(key, zk.getData("/keys/" + dir + "/" + key, null, null));
       }
     } catch (KeeperException.NoNodeException e) {
       log.info("Error fetching key names");

solr/core/src/java/org/apache/solr/cloud/DistributedApiAsyncTracker.java

@@ -311,8 +311,8 @@ enum State {
       this.rootNodePath = rootNodePath;
 
       try {
-        if (!zkClient.exists(rootNodePath, true)) {
-          zkClient.makePath(rootNodePath, new byte[0], CreateMode.PERSISTENT, true);
+        if (!zkClient.exists(rootNodePath)) {
+          zkClient.makePath(rootNodePath, new byte[0], CreateMode.PERSISTENT);
         }
       } catch (KeeperException.NodeExistsException nee) {
         // Some other thread (on this or another JVM) beat us to create the node, that's ok, the
@@ -329,27 +329,27 @@ void createNewInFlightTask(String asyncId) throws KeeperException, InterruptedEx
       zkClient.create(
           getPath(asyncId),
           State.SUBMITTED.shorthand.getBytes(StandardCharsets.UTF_8),
-          CreateMode.EPHEMERAL,
-          true);
+          CreateMode.EPHEMERAL
+      );
     }
 
     void setTaskRunning(String asyncId) throws KeeperException, InterruptedException {
       zkClient.setData(
-          getPath(asyncId), State.RUNNING.shorthand.getBytes(StandardCharsets.UTF_8), true);
+          getPath(asyncId), State.RUNNING.shorthand.getBytes(StandardCharsets.UTF_8));
     }
 
     void deleteInFlightTask(String asyncId) throws KeeperException, InterruptedException {
-      zkClient.delete(getPath(asyncId), -1, true);
+      zkClient.delete(getPath(asyncId), -1);
     }
 
     State getInFlightState(String asyncId) throws KeeperException, InterruptedException {
-      if (!zkClient.exists(getPath(asyncId), true)) {
+      if (!zkClient.exists(getPath(asyncId))) {
         return State.NOT_FOUND;
       }
 
       final byte[] bytes;
       try {
-        bytes = zkClient.getData(getPath(asyncId), null, null, true);
+        bytes = zkClient.getData(getPath(asyncId), null, null);
       } catch (KeeperException.NoNodeException nne) {
         // Unlikely race, but not impossible...
         if (log.isInfoEnabled()) {

solr/core/src/java/org/apache/solr/cloud/DistributedClusterStateUpdater.java

@@ -578,7 +578,7 @@ private void doStateDotJsonCasUpdate(ClusterState updatedState)
         if (updater.isCollectionCreation()) {
           // The state.json file does not exist yet (more precisely it is assumed not to exist)
           log.debug("going to create collection {}", jsonPath);
-          zkStateReader.getZkClient().create(jsonPath, stateJson, CreateMode.PERSISTENT, true);
+          zkStateReader.getZkClient().create(jsonPath, stateJson, CreateMode.PERSISTENT);
         } else {
           // We're updating an existing state.json
           if (log.isDebugEnabled()) {
@@ -589,7 +589,7 @@ private void doStateDotJsonCasUpdate(ClusterState updatedState)
           }
           zkStateReader
               .getZkClient()
-              .setData(jsonPath, stateJson, collection.getZNodeVersion(), true);
+              .setData(jsonPath, stateJson, collection.getZNodeVersion());
         }
       }
     }
@@ -603,7 +603,7 @@ private void doStateDotJsonCasUpdate(ClusterState updatedState)
     private ClusterState fetchStateForCollection() throws KeeperException, InterruptedException {
       String collectionStatePath = ZkStateReader.getCollectionPath(updater.getCollectionName());
       Stat stat = new Stat();
-      byte[] data = zkStateReader.getZkClient().getData(collectionStatePath, null, stat, true);
+      byte[] data = zkStateReader.getZkClient().getData(collectionStatePath, null, stat);
 
       // This factory method can detect a missing configName and supply it by reading it from the
       // old ZK location.
@@ -910,7 +910,7 @@ public static void executeNodeDownStateUpdate(String nodeName, ZkStateReader zkS
 
       try {
         final List<String> collectionNames =
-            zkStateReader.getZkClient().getChildren(COLLECTIONS_ZKNODE, null, true);
+            zkStateReader.getZkClient().getChildren(COLLECTIONS_ZKNODE, null);
 
         // Collections are totally independent of each other. Multiple threads could share the load
         // here (need a ZK connection for each though).

solr/core/src/java/org/apache/solr/cloud/DistributedMap.java

@@ -22,7 +22,6 @@
 import org.apache.solr.common.SolrException;
 import org.apache.solr.common.SolrException.ErrorCode;
 import org.apache.solr.common.cloud.SolrZkClient;
-import org.apache.solr.common.cloud.ZkCmdExecutor;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.KeeperException.NodeExistsException;
@@ -42,9 +41,8 @@ public class DistributedMap {
   public DistributedMap(SolrZkClient zookeeper, String dir) {
     this.dir = dir;
 
-    ZkCmdExecutor cmdExecutor = new ZkCmdExecutor(zookeeper.getZkClientTimeout());
     try {
-      cmdExecutor.ensureExists(dir, zookeeper);
+      zookeeper.ensureExists(dir);
     } catch (KeeperException e) {
       throw new SolrException(ErrorCode.SERVER_ERROR, e);
     } catch (InterruptedException e) {
@@ -57,7 +55,7 @@ public DistributedMap(SolrZkClient zookeeper, String dir) {
 
   public void put(String trackingId, byte[] data) throws KeeperException, InterruptedException {
     zookeeper.makePath(
-        dir + "/" + PREFIX + trackingId, data, CreateMode.PERSISTENT, null, false, true);
+        dir + "/" + PREFIX + trackingId, data, CreateMode.PERSISTENT, null, false);
   }
 
   /**
@@ -69,24 +67,24 @@ public boolean putIfAbsent(String trackingId, byte[] data)
       throws KeeperException, InterruptedException {
     try {
       zookeeper.makePath(
-          dir + "/" + PREFIX + trackingId, data, CreateMode.PERSISTENT, null, true, true);
+          dir + "/" + PREFIX + trackingId, data, CreateMode.PERSISTENT, null, true);
       return true;
     } catch (NodeExistsException e) {
       return false;
     }
   }
 
   public byte[] get(String trackingId) throws KeeperException, InterruptedException {
-    return zookeeper.getData(dir + "/" + PREFIX + trackingId, null, null, true);
+    return zookeeper.getData(dir + "/" + PREFIX + trackingId, null, null);
   }
 
   public boolean contains(String trackingId) throws KeeperException, InterruptedException {
-    return zookeeper.exists(dir + "/" + PREFIX + trackingId, true);
+    return zookeeper.exists(dir + "/" + PREFIX + trackingId);
   }
 
   public int size() throws KeeperException, InterruptedException {
     Stat stat = new Stat();
-    zookeeper.getData(dir, null, stat, true);
+    zookeeper.getData(dir, null, stat);
     return stat.getNumChildren();
   }
 
@@ -96,7 +94,7 @@ public int size() throws KeeperException, InterruptedException {
    */
   public boolean remove(String trackingId) throws KeeperException, InterruptedException {
     try {
-      zookeeper.delete(dir + "/" + PREFIX + trackingId, -1, true);
+      zookeeper.delete(dir + "/" + PREFIX + trackingId, -1);
     } catch (KeeperException.NoNodeException e) {
       return false;
     }
@@ -105,15 +103,15 @@ public boolean remove(String trackingId) throws KeeperException, InterruptedExce
 
   /** Helper method to clear all child nodes for a parent node. */
   public void clear() throws KeeperException, InterruptedException {
-    List<String> childNames = zookeeper.getChildren(dir, null, true);
+    List<String> childNames = zookeeper.getChildren(dir, null);
     for (String childName : childNames) {
-      zookeeper.delete(dir + "/" + childName, -1, true);
+      zookeeper.delete(dir + "/" + childName, -1);
     }
   }
 
   /** Returns the keys of all the elements in the map */
   public Collection<String> keys() throws KeeperException, InterruptedException {
-    List<String> childs = zookeeper.getChildren(dir, null, true);
+    List<String> childs = zookeeper.getChildren(dir, null);
     final List<String> ids = new ArrayList<>(childs.size());
     childs.stream().forEach((child) -> ids.add(child.substring(PREFIX.length())));
     return ids;

solr/core/src/java/org/apache/solr/cloud/ElectionContext.java

@@ -55,7 +55,7 @@ public void cancelElection() throws InterruptedException, KeeperException {
     if (leaderSeqPath != null) {
       try {
         log.debug("Canceling election {}", leaderSeqPath);
-        zkClient.delete(leaderSeqPath, -1, true);
+        zkClient.delete(leaderSeqPath, -1);
       } catch (NoNodeException e) {
         // fine
         log.debug("cancelElection did not find election node to remove {}", leaderSeqPath);

solr/core/src/java/org/apache/solr/cloud/LeaderElector.java

@@ -29,7 +29,6 @@
 import org.apache.solr.common.AlreadyClosedException;
 import org.apache.solr.common.SolrException;
 import org.apache.solr.common.cloud.SolrZkClient;
-import org.apache.solr.common.cloud.ZkCmdExecutor;
 import org.apache.solr.common.cloud.ZooKeeperException;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
@@ -62,8 +61,6 @@ public class LeaderElector {
 
   protected SolrZkClient zkClient;
 
-  private ZkCmdExecutor zkCmdExecutor;
-
   private volatile ElectionContext context;
 
   private ElectionWatcher watcher;
@@ -73,13 +70,11 @@ public class LeaderElector {
 
   public LeaderElector(SolrZkClient zkClient) {
     this.zkClient = zkClient;
-    zkCmdExecutor = new ZkCmdExecutor(zkClient.getZkClientTimeout());
   }
 
   public LeaderElector(
       SolrZkClient zkClient, ContextKey key, Map<ContextKey, ElectionContext> electionContexts) {
     this.zkClient = zkClient;
-    zkCmdExecutor = new ZkCmdExecutor(zkClient.getZkClientTimeout());
     this.electionContexts = electionContexts;
     this.contextKey = key;
   }
@@ -100,7 +95,7 @@ private void checkIfIamLeader(final ElectionContext context, boolean replacement
     context.checkIfIamLeaderFired();
     // get all other numbers...
     final String holdElectionPath = context.electionPath + ELECTION_NODE;
-    List<String> seqs = zkClient.getChildren(holdElectionPath, null, true);
+    List<String> seqs = zkClient.getChildren(holdElectionPath, null);
     sortSeqs(seqs);
 
     String leaderSeqNodeName =
@@ -112,15 +107,15 @@ private void checkIfIamLeader(final ElectionContext context, boolean replacement
 
     // If any double-registrations exist for me, remove all but this latest one!
     // TODO: can we even get into this state?
-    String prefix = zkClient.getSolrZooKeeper().getSessionId() + "-" + context.id + "-";
+    String prefix = zkClient.getZkSessionId() + "-" + context.id + "-";
     Iterator<String> it = seqs.iterator();
     while (it.hasNext()) {
       String elec = it.next();
       if (!elec.equals(leaderSeqNodeName) && elec.startsWith(prefix)) {
         try {
           String toDelete = holdElectionPath + "/" + elec;
           log.warn("Deleting duplicate registration: {}", toDelete);
-          zkClient.delete(toDelete, -1, true);
+          zkClient.delete(toDelete, -1);
         } catch (KeeperException.NoNodeException e) {
           // ignore
         }
@@ -154,8 +149,8 @@ private void checkIfIamLeader(final ElectionContext context, boolean replacement
             watcher =
                 new ElectionWatcher(
                     context.leaderSeqPath, watchedNode, getSeq(context.leaderSeqPath), context),
-            null,
-            true);
+            null
+        );
         log.debug("Watching path {} to know if I could be the leader", watchedNode);
       } catch (KeeperException.SessionExpiredException e) {
         throw e;
@@ -232,7 +227,7 @@ public int joinElection(ElectionContext context, boolean replacement, boolean jo
 
     final String shardsElectZkPath = context.electionPath + LeaderElector.ELECTION_NODE;
 
-    long sessionId = zkClient.getSolrZooKeeper().getSessionId();
+    long sessionId = zkClient.getZkSessionId();
     String id = sessionId + "-" + context.id;
     String leaderSeqPath = null;
     boolean cont = true;
@@ -248,8 +243,8 @@ public int joinElection(ElectionContext context, boolean replacement, boolean jo
                 zkClient.create(
                     shardsElectZkPath + "/" + id + "-n_",
                     null,
-                    CreateMode.EPHEMERAL_SEQUENTIAL,
-                    false);
+                    CreateMode.EPHEMERAL_SEQUENTIAL
+                );
           } else {
             String firstInLine = nodes.get(1);
             log.debug("The current head: {}", firstInLine);
@@ -258,23 +253,23 @@ public int joinElection(ElectionContext context, boolean replacement, boolean jo
               throw new IllegalStateException("Could not find regex match in:" + firstInLine);
             }
             leaderSeqPath = shardsElectZkPath + "/" + id + "-n_" + m.group(1);
-            zkClient.create(leaderSeqPath, null, CreateMode.EPHEMERAL, false);
+            zkClient.create(leaderSeqPath, null, CreateMode.EPHEMERAL);
           }
         } else {
           leaderSeqPath =
               zkClient.create(
                   shardsElectZkPath + "/" + id + "-n_",
                   null,
-                  CreateMode.EPHEMERAL_SEQUENTIAL,
-                  false);
+                  CreateMode.EPHEMERAL_SEQUENTIAL
+              );
         }
 
         log.debug("Joined leadership election with path: {}", leaderSeqPath);
         context.leaderSeqPath = leaderSeqPath;
         cont = false;
       } catch (ConnectionLossException e) {
         // we don't know if we made our node or not...
-        List<String> entries = zkClient.getChildren(shardsElectZkPath, null, true);
+        List<String> entries = zkClient.getChildren(shardsElectZkPath, null);
 
         boolean foundId = false;
         for (String entry : entries) {
@@ -342,7 +337,7 @@ public void process(WatchedEvent event) {
       if (canceled) {
         log.debug("This watcher is not active anymore {}", myNode);
         try {
-          zkClient.delete(myNode, -1, true);
+          zkClient.delete(myNode, -1);
         } catch (KeeperException.NoNodeException nne) {
           // expected . don't do anything
         } catch (Exception e) {
@@ -353,7 +348,7 @@ public void process(WatchedEvent event) {
       try {
         // am I the next leader?
         checkIfIamLeader(context, true);
-      } catch (AlreadyClosedException e) {
+      } catch (IllegalStateException e) {
 
       } catch (Exception e) {
         if (!zkClient.isClosed()) {
@@ -367,10 +362,10 @@ public void process(WatchedEvent event) {
   public void setup(final ElectionContext context) throws InterruptedException, KeeperException {
     String electZKPath = context.electionPath + LeaderElector.ELECTION_NODE;
     if (context instanceof OverseerElectionContext) {
-      zkCmdExecutor.ensureExists(electZKPath, zkClient);
+      zkClient.ensureExists(electZKPath, null, CreateMode.PERSISTENT);
     } else {
       // we use 2 param so that replica won't create /collection/{collection} if it doesn't exist
-      zkCmdExecutor.ensureExists(electZKPath, (byte[]) null, CreateMode.PERSISTENT, zkClient, 2);
+      zkClient.ensureExists(electZKPath, null, CreateMode.PERSISTENT, 2);
     }
 
     this.context = context;