moved Zk cleaner threads into utility class

core/src/main/java/org/apache/accumulo/core/conf/Property.java

@@ -1423,7 +1423,12 @@ start with the category prefix, followed by a scope (minc, majc, scan, \
       "The interval at which to check for dead compactors.", "2.1.0"),
   GENERAL_AMPLE_CONDITIONAL_WRITER_THREADS_MAX("general.ample.conditional.writer.threads.max", "8",
       PropertyType.COUNT,
-      "The maximum number of threads for the shared ConditionalWriter used by Ample.", "4.0.0");
+      "The maximum number of threads for the shared ConditionalWriter used by Ample.", "4.0.0"),
+  GENERAL_EXPECTED_PROCESS_COUNTS("general.expected.process.counts", "", PropertyType.STRING,
+      "Declare the expected number of server processes per type and resource group. Used by the process-status utility to identify compactor and scan servers that are down."
+          + "Format: comma-separated list of <type>.<resourceGroup>=<count> entries. valid types are 'compactor' and 'sserver'."
+          + "Example: compactor.default=2, compactor.CTEST=3, sserver.default=1 ",
+      "4.0.0");
 
   private final String key;
   private final String defaultValue;

core/src/main/java/org/apache/accumulo/core/util/ExpectedProcessCounts.java

@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.core.util;
+
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.OptionalInt;
+
+import org.apache.accumulo.core.client.admin.servers.ServerId;
+import org.apache.accumulo.core.data.ResourceGroupId;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ExpectedProcessCounts {
+
+  private static final Logger log = LoggerFactory.getLogger(ExpectedProcessCounts.class);
+
+  private static final Map<String,ServerId.Type> SUPPORTED_TYPES =
+      Map.of("compactor", ServerId.Type.COMPACTOR, "sserver", ServerId.Type.SCAN_SERVER);
+
+  private final Map<ServerId.Type,Map<ResourceGroupId,Integer>> counts;
+
+  private ExpectedProcessCounts(Map<ServerId.Type,Map<ResourceGroupId,Integer>> counts) {
+    this.counts = counts;
+  }
+
+  public static ExpectedProcessCounts parse(String propertyValue) {
+    Map<ServerId.Type,Map<ResourceGroupId,Integer>> result = new EnumMap<>(ServerId.Type.class);
+
+    if (propertyValue == null || propertyValue.isBlank()) {
+      return new ExpectedProcessCounts(result);
+    }
+
+    for (String entry : propertyValue.split(",")) {
+      entry = entry.trim();
+      if (entry.isEmpty()) {
+        continue;
+      }
+
+      int eqIdx = entry.lastIndexOf('=');
+      if (eqIdx < 0) {
+        log.warn("Ignoring malformed entry in {} (missing '='): {}",
+            "general.expected.process.counts", entry);
+        continue;
+      }
+
+      String key = entry.substring(0, eqIdx).trim();
+      String valueStr = entry.substring(eqIdx + 1).trim();
+
+      int dotIdx = key.indexOf('.');
+      if (dotIdx < 0) {
+        log.warn("Ignoring malformed key in {} (expected '<type>.<resourceGroup>'): {}",
+            "general.expected.process.counts", key);
+        continue;
+      }
+
+      String typeName = key.substring(0, dotIdx).trim().toLowerCase();
+      String groupName = key.substring(dotIdx + 1).trim();
+
+      ServerId.Type serverType = SUPPORTED_TYPES.get(typeName);
+      if (serverType == null) {
+        log.warn("Ignoring unknown server type '{}' in general.expected.process.counts."
+            + " Supported types: {}", typeName, SUPPORTED_TYPES.keySet());
+        continue;
+      }
+
+      if (groupName.isEmpty()) {
+        log.warn("Ignoring entry with empty resource group name in"
+            + " general.expected.process.counts: {}", entry);
+        continue;
+      }
+
+      int count;
+      try {
+        count = Integer.parseInt(valueStr);
+        if (count < 0) {
+          throw new NumberFormatException("count must be non-negative");
+        }
+      } catch (NumberFormatException e) {
+        log.warn("Ignoring entry with invalid count '{}' in general.expected.process.counts: {}",
+            valueStr, entry);
+        continue;
+      }
+
+      result.computeIfAbsent(serverType, t -> new HashMap<>()).put(ResourceGroupId.of(groupName),
+          count);
+    }
+
+    return new ExpectedProcessCounts(Collections.unmodifiableMap(result));
+  }
+
+  public OptionalInt getExpectedCount(ServerId.Type type, ResourceGroupId group) {
+    var groupMap = counts.get(type);
+    if (groupMap == null) {
+      return OptionalInt.empty();
+    }
+    Integer count = groupMap.get(group);
+    return count == null ? OptionalInt.empty() : OptionalInt.of(count);
+  }
+
+  public Map<ServerId.Type,Map<ResourceGroupId,Integer>> all() {
+    return counts;
+  }
+
+  public boolean isEmpty() {
+    return counts.isEmpty();
+  }
+}

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

@@ -0,0 +1,216 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.server.util;
+
+import java.util.Arrays;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.cli.ServerOpts;
+import org.apache.accumulo.core.data.ResourceGroupId;
+import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
+import org.apache.accumulo.core.lock.ServiceLock;
+import org.apache.accumulo.core.lock.ServiceLockData;
+import org.apache.accumulo.core.lock.ServiceLockPaths.AddressSelector;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ResourceGroupPredicate;
+import org.apache.accumulo.core.lock.ServiceLockPaths.ServiceLockPath;
+import org.apache.accumulo.core.zookeeper.ZcStat;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.util.CleanZooKeeper.CleanOpts;
+import org.apache.accumulo.start.spi.CommandGroup;
+import org.apache.accumulo.start.spi.CommandGroups;
+import org.apache.accumulo.start.spi.KeywordExecutable;
+import org.apache.zookeeper.KeeperException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import com.google.auto.service.AutoService;
+
+@AutoService(KeywordExecutable.class)
+public class CleanZooKeeper extends ServerKeywordExecutable<CleanOpts> {
+
+  private static final Logger log = LoggerFactory.getLogger(CleanZooKeeper.class);
+
+  public CleanZooKeeper() {
+    super(new CleanOpts());
+  }
+
+  static class CleanOpts extends ServerOpts {
+    @Parameter(names = "--compactors",
+        description = "Remove ZooKeeper paths for compactors that are no longer running")
+    boolean cleanCompactors = false;
+
+    @Parameter(names = "--sservers",
+        description = "Remove ZooKeeper paths for scan servers that are no longer running")
+    boolean cleanScanServers = false;
+
+    @Parameter(names = "--include-groups",
+        description = "Comma-separated list of resource groups to include (default: all)")
+    String includeGroups;
+
+    @Parameter(names = "--dry-run",
+        description = "Print paths that would be removed without making any changes")
+    boolean dryRun = false;
+
+    @Parameter(names = "--verbose", description = "Print progress messages")
+    boolean verbose = false;
+  }
+
+  @Override
+  public String keyword() {
+    return "clean-zk";
+  }
+
+  @Override
+  public String description() {
+    return "Removes ZooKeeper paths for compactors and scan servers that are no longer running."
+        + " Run only when permanently changing deployment topology, not during normal operations.";
+  }
+
+  @Override
+  public CommandGroup commandGroup() {
+    return CommandGroups.ZOOKEEPER;
+  }
+
+  @Override
+  public void execute(JCommander cl, CleanOpts opts) throws Exception {
+    if (!opts.cleanCompactors && !opts.cleanScanServers) {
+      new JCommander(opts).usage();
+      return;
+    }
+
+    final ResourceGroupPredicate rgp;
+    if (opts.includeGroups != null) {
+      Set<ResourceGroupId> groups = Arrays.stream(opts.includeGroups.split(",")).map(String::trim)
+          .map(ResourceGroupId::of).collect(Collectors.toSet());
+      rgp = groups::contains;
+    } else {
+      rgp = ResourceGroupPredicate.ANY;
+    }
+
+    var context = getServerContext();
+    var zrw = context.getZooSession().asReaderWriter();
+
+    if (opts.cleanCompactors) {
+      cleanCompactors(context, zrw, rgp, opts);
+    }
+    if (opts.cleanScanServers) {
+      cleanScanServers(context, zrw, rgp, opts);
+    }
+  }
+
+  private void cleanCompactors(ServerContext context, ZooReaderWriter zrw,
+      ResourceGroupPredicate rgp, CleanOpts opts) {
+
+    // Remove individual compactor nodes with no active lock
+    Set<ServiceLockPath> compactorPaths =
+        context.getServerPaths().getCompactor(rgp, AddressSelector.all(), false);
+
+    for (ServiceLockPath path : compactorPaths) {
+      ZcStat stat = new ZcStat();
+      Optional<ServiceLockData> lockData =
+          ServiceLock.getLockData(context.getZooCache(), path, stat);
+      if (lockData.isEmpty()) {
+        message("Removing empty compactor ZK path: " + path, opts);
+        if (!opts.dryRun) {
+          try {
+            zrw.delete(path.toString());
+          } catch (KeeperException.NotEmptyException e) {
+            log.debug("Failed to delete compactor ZK node {} because it is not empty,"
+                + " likely an expected race condition.", path);
+          } catch (KeeperException | InterruptedException e) {
+            if (e instanceof InterruptedException) {
+              Thread.currentThread().interrupt();
+            }
+            log.warn("Failed to delete compactor ZK node {}", path, e);
+          }
+        }
+      }
+    }
+
+    // Remove empty resource group parent nodes under ZCOMPACTORS
+    try {
+      var groups = zrw.getChildren(Constants.ZCOMPACTORS);
+      for (String group : groups) {
+        ResourceGroupId rgid = ResourceGroupId.of(group);
+        if (!rgp.test(rgid)) {
+          continue;
+        }
+        String groupPath = Constants.ZCOMPACTORS + "/" + group;
+        var children = zrw.getChildren(groupPath);
+        if (children.isEmpty()) {
+          message("Removing empty compactor group ZK path: " + groupPath, opts);
+          if (!opts.dryRun) {
+            try {
+              zrw.delete(groupPath);
+            } catch (KeeperException.NotEmptyException e) {
+              log.debug("Failed to delete compactor group ZK node {} because it is not empty.",
+                  groupPath);
+            }
+          }
+        }
+      }
+    } catch (KeeperException | InterruptedException e) {
+      if (e instanceof InterruptedException) {
+        Thread.currentThread().interrupt();
+      }
+      log.warn("Failed to clean up empty compactor group paths", e);
+    }
+  }
+
+  private void cleanScanServers(org.apache.accumulo.server.ServerContext context,
+      ZooReaderWriter zrw, ResourceGroupPredicate rgp, CleanOpts opts) {
+    try {
+      Set<ServiceLockPath> scanServerPaths =
+          context.getServerPaths().getScanServer(rgp, AddressSelector.all(), false);
+
+      for (ServiceLockPath path : scanServerPaths) {
+        ZcStat stat = new ZcStat();
+        Optional<ServiceLockData> lockData =
+            ServiceLock.getLockData(context.getZooCache(), path, stat);
+        if (lockData.isEmpty()) {
+          message("Removing empty scan server ZK path: " + path, opts);
+          if (!opts.dryRun) {
+            try {
+              zrw.delete(path.toString());
+            } catch (KeeperException.NotEmptyException e) {
+              log.debug("Failed to delete scan server ZK node {} because it is not empty,"
+                  + " likely an expected race condition.", path);
+            }
+          }
+        }
+      }
+    } catch (KeeperException e) {
+      log.error("Exception trying to delete empty scan server ZK paths", e);
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      log.error("Interrupted trying to delete empty scan server ZK paths", e);
+    }
+  }
+
+  private static void message(String msg, CleanOpts opts) {
+    if (opts.verbose || opts.dryRun) {
+      System.out.println(msg);
+    }
+  }
+}