Added ZooZap and ServiceLock methods to delete ScanServer locks

assemble/bin/accumulo-cluster

@@ -796,7 +796,7 @@ function prune() {
       read -r -a groups <<<"$ARG_SSERVER_GROUP"
     else
       # find all groups known in zookeeper, this will allow pruning entire groups that do not even exist in cluster.yaml
-      readarray -t groups < <(jq -r ".summaries.S_SERVER.resourceGroups | .[] " "$service_json")
+      readarray -t groups < <(jq -r ".summaries.S_SERVER.resourceGroups | keys | .[]?" "$service_json")
     fi
 
     for group in "${groups[@]}"; do
@@ -815,7 +815,7 @@ function prune() {
       read -r -a groups <<<"$ARG_COMPACTOR_GROUP"
     else
       # find all groups known in zookeeper, this will allow pruning entire groups that do not even exist in cluster.yaml
-      readarray -t groups < <(jq -r ".summaries.COMPACTOR.resourceGroups | .[] " "$service_json")
+      readarray -t groups < <(jq -r ".summaries.COMPACTOR.resourceGroups | keys | .[]?" "$service_json")
     fi
 
     for group in "${groups[@]}"; do

core/src/main/java/org/apache/accumulo/core/fate/zookeeper/ServiceLock.java

@@ -18,6 +18,7 @@
  */
 package org.apache.accumulo.core.fate.zookeeper;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.Objects.requireNonNull;
 
 import java.util.ArrayList;
@@ -754,6 +755,43 @@ public static void deleteLocks(ZooReaderWriter zk, String zPath,
     }
   }
 
+  public static void deleteScanServerLocks(ZooReaderWriter zk, String zPath,
+      Predicate<HostAndPort> hostPortPredicate, Predicate<String> groupPredicate,
+      Consumer<String> messageOutput, Boolean dryRun) throws KeeperException, InterruptedException {
+
+    Objects.requireNonNull(zPath, "Lock path cannot be null");
+    Objects.requireNonNull(groupPredicate, "group predicate cannot be null");
+    if (!zk.exists(zPath)) {
+      throw new IllegalStateException("Path " + zPath + " does not exist");
+    }
+
+    List<String> servers = zk.getChildren(zPath);
+    if (servers.isEmpty()) {
+      throw new IllegalStateException("No server locks are held at " + zPath);
+    }
+
+    ZooKeeper z = zk.getZooKeeper();
+    for (String server : servers) {
+      if (hostPortPredicate.test(HostAndPort.fromString(server))) {
+        final String serverPath = zPath + "/" + server;
+        byte[] lockData = ServiceLock.getLockData(z, path(serverPath));
+        if (lockData == null) {
+          messageOutput.accept("Skipping server " + server + " as it's lock content is empty.");
+          continue;
+        }
+        String lockContent = new String(lockData, UTF_8);
+        String[] parts = lockContent.split(",");
+        if (parts.length == 2 && groupPredicate.test(parts[1])) {
+          messageOutput.accept("Deleting " + serverPath + " from zookeeper");
+          if (!dryRun) {
+            LOG.debug("Deleting all locks at path {} due to lock deletion", serverPath);
+            zk.recursiveDelete(serverPath, NodeMissingPolicy.SKIP);
+          }
+        }
+      }
+    }
+  }
+
   /**
    * This method will delete the top server lock for a given lock path
    *

server/base/src/main/java/org/apache/accumulo/server/util/Admin.java

@@ -560,6 +560,7 @@ private static void stopServer(final ClientContext context, final boolean tablet
         client -> client.shutdown(TraceUtil.traceInfo(), context.rpcCreds(), tabletServersToo));
   }
 
+  @SuppressWarnings("deprecation")
   private static void stopServers(final ServerContext context, List<String> servers,
       final boolean force)
       throws AccumuloException, AccumuloSecurityException, InterruptedException, KeeperException {
@@ -590,9 +591,11 @@ private static void stopServers(final ServerContext context, List<String> server
         String tserversPath = Constants.ZROOT + "/" + iid + Constants.ZTSERVERS;
         ZooZap.removeLocks(zk, tserversPath, hostAndPort::contains, opts);
         String compactorsBasepath = Constants.ZROOT + "/" + iid + Constants.ZCOMPACTORS;
-        ZooZap.removeGroupedLocks(zk, compactorsBasepath, rg -> true, hostAndPort::contains, opts);
+        ZooZap.removeCompactorGroupedLocks(zk, compactorsBasepath, rg -> true,
+            hostAndPort::contains, opts);
         String sserversPath = Constants.ZROOT + "/" + iid + Constants.ZSSERVERS;
-        ZooZap.removeGroupedLocks(zk, sserversPath, rg -> true, hostAndPort::contains, opts);
+        ZooZap.removeScanServerGroupLocks(zk, sserversPath, hostAndPort::contains, rg -> true,
+            opts);
 
         String managerLockPath = Constants.ZROOT + "/" + iid + Constants.ZMANAGER_LOCK;
         ZooZap.removeSingletonLock(zk, managerLockPath, hostAndPort::contains, opts);

server/base/src/main/java/org/apache/accumulo/server/util/ZooZap.java

@@ -235,7 +235,8 @@ public void zap(SiteConfiguration siteConf, String... args) {
     if (opts.zapCompactors) {
       String compactorsBasepath = Constants.ZROOT + "/" + iid + Constants.ZCOMPACTORS;
       try {
-        removeGroupedLocks(zoo, compactorsBasepath, groupPredicate, hostPortPredicate, opts);
+        removeCompactorGroupedLocks(zoo, compactorsBasepath, groupPredicate, hostPortPredicate,
+            opts);
       } catch (KeeperException | InterruptedException e) {
         log.error("Error deleting compactors from zookeeper", e);
       }
@@ -245,7 +246,11 @@ public void zap(SiteConfiguration siteConf, String... args) {
     if (opts.zapScanServers) {
       String sserversPath = Constants.ZROOT + "/" + iid + Constants.ZSSERVERS;
       try {
-        removeGroupedLocks(zoo, sserversPath, groupPredicate, hostPortPredicate, opts);
+        if (opts.includeGroups == null) {
+          removeLocks(zoo, sserversPath, hostPortPredicate, opts);
+        } else {
+          removeScanServerGroupLocks(zoo, sserversPath, hostPortPredicate, groupPredicate, opts);
+        }
       } catch (KeeperException | InterruptedException e) {
         log.error("Error deleting scan server locks", e);
       }
@@ -263,8 +268,8 @@ private static void zapDirectory(ZooReaderWriter zoo, String path, Opts opts)
     }
   }
 
-  static void removeGroupedLocks(ZooReaderWriter zoo, String path, Predicate<String> groupPredicate,
-      Predicate<HostAndPort> hostPortPredicate, Opts opts)
+  static void removeCompactorGroupedLocks(ZooReaderWriter zoo, String path,
+      Predicate<String> groupPredicate, Predicate<HostAndPort> hostPortPredicate, Opts opts)
       throws KeeperException, InterruptedException {
     if (zoo.exists(path)) {
       List<String> groups = zoo.getChildren(path);
@@ -283,6 +288,14 @@ static void removeLocks(ZooReaderWriter zoo, String path,
     ServiceLock.deleteLocks(zoo, path, hostPortPredicate, m -> message(m, opts), opts.dryRun);
   }
 
+  @Deprecated(since = "2.1.5")
+  static void removeScanServerGroupLocks(ZooReaderWriter zoo, String path,
+      Predicate<HostAndPort> hostPortPredicate, Predicate<String> groupPredicate, Opts opts)
+      throws KeeperException, InterruptedException {
+    ServiceLock.deleteScanServerLocks(zoo, path, hostPortPredicate, groupPredicate,
+        m -> message(m, opts), opts.dryRun);
+  }
+
   static void removeSingletonLock(ZooReaderWriter zoo, String path,
       Predicate<HostAndPort> hostPortPredicate, Opts ops)
       throws KeeperException, InterruptedException {

server/base/src/test/java/org/apache/accumulo/server/util/AdminTest.java

@@ -18,16 +18,21 @@
  */
 package org.apache.accumulo.server.util;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 import java.util.Collections;
+import java.util.List;
 import java.util.UUID;
 
 import org.apache.accumulo.core.Constants;
 import org.apache.accumulo.core.clientImpl.ClientContext;
 import org.apache.accumulo.core.data.InstanceId;
 import org.apache.accumulo.core.fate.zookeeper.ZooCache;
 import org.apache.accumulo.core.fate.zookeeper.ZooCache.ZcStat;
+import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
+import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeMissingPolicy;
+import org.apache.zookeeper.ZooKeeper;
 import org.easymock.EasyMock;
 import org.junit.jupiter.api.Test;
 
@@ -93,4 +98,69 @@ public void testCannotQualifySessionId() {
     EasyMock.verify(zc);
   }
 
+  /**
+   * SServer group filter should use lock data (UUID,group).
+   */
+  @SuppressWarnings("deprecation")
+  @Test
+  public void testSserverGroupFilterUsesLockData() throws Exception {
+
+    ZooReaderWriter zoo = EasyMock.createMock(ZooReaderWriter.class);
+    ZooKeeper zk = EasyMock.createMock(ZooKeeper.class);
+
+    String basePath = "/accumulo/iid/sservers";
+    String hostDefault = "host1:10000";
+    String hostOther = "host2:10001";
+    String zlock1 = "zlock#00000000-0000-0000-0000-aaaaaaaaaaaa#0000000001";
+    String zlock2 = "zlock#00000000-0000-0000-0000-bbbbbbbbbbbb#0000000001";
+
+    EasyMock.expect(zoo.exists(basePath)).andReturn(true);
+    EasyMock.expect(zoo.getChildren(basePath)).andReturn(List.of(hostDefault, hostOther));
+    EasyMock.expect(zoo.getZooKeeper()).andReturn(zk);
+    EasyMock.expect(zk.getChildren(basePath + "/" + hostDefault, null)).andReturn(List.of(zlock1));
+    EasyMock.expect(zk.getData(basePath + "/" + hostDefault + "/" + zlock1, false, null))
+        .andReturn((UUID.randomUUID().toString() + ",default").getBytes(UTF_8));
+    EasyMock.expect(zk.getChildren(basePath + "/" + hostOther, null)).andReturn(List.of(zlock2));
+    EasyMock.expect(zk.getData(basePath + "/" + hostOther + "/" + zlock2, false, null))
+        .andReturn((UUID.randomUUID().toString() + ",rg1").getBytes(UTF_8));
+
+    zoo.recursiveDelete(basePath + "/" + hostDefault, NodeMissingPolicy.SKIP);
+    EasyMock.expectLastCall();
+
+    EasyMock.replay(zoo, zk);
+
+    ZooZap.Opts opts = new ZooZap.Opts();
+    ZooZap.removeScanServerGroupLocks(zoo, basePath, hp -> true, "default"::equals, opts);
+
+    EasyMock.verify(zoo, zk);
+
+  }
+
+  /**
+   * SServer cleanup without group filter should delete all host nodes.
+   */
+  @Test
+  public void testSserverDeleteAllNoGroupFilter() throws Exception {
+    ZooReaderWriter zoo = EasyMock.createMock(ZooReaderWriter.class);
+
+    String basePath = "/accumulo/iid/sservers";
+    String host1 = "host1:10000";
+    String host2 = "host2:10001";
+
+    EasyMock.expect(zoo.exists(basePath)).andReturn(true);
+    EasyMock.expect(zoo.getChildren(basePath)).andReturn(List.of(host1, host2));
+
+    zoo.recursiveDelete(basePath + "/" + host1, NodeMissingPolicy.SKIP);
+    EasyMock.expectLastCall();
+
+    zoo.recursiveDelete(basePath + "/" + host2, NodeMissingPolicy.SKIP);
+    EasyMock.expectLastCall();
+
+    EasyMock.replay(zoo);
+
+    ZooZap.Opts opts = new ZooZap.Opts();
+    ZooZap.removeLocks(zoo, basePath, hp -> true, opts);
+
+    EasyMock.verify(zoo);
+  }
 }

server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java

@@ -328,7 +328,7 @@ public String getClientAddressString() {
   }
 
   /**
-   * Set up nodes and locks in ZooKeeper for this Compactor
+   * Set up nodes and locks in ZooKeeper for this ScanServer
    */
   private ServiceLock announceExistence() {
     ZooReaderWriter zoo = getContext().getZooReaderWriter();