WIP initial steps for distributing compaction coordination

core/src/main/java/org/apache/accumulo/core/Constants.java

@@ -54,6 +54,7 @@ public class Constants {
   public static final String ZMANAGER_TICK = ZMANAGERS + "/tick";
   public static final String ZMANAGER_ASSIGNMENTS = ZMANAGERS + "/assignments";
   public static final String ZMANAGER_FATE_ASSIGNMENTS = ZMANAGER_ASSIGNMENTS + "/fate";
+  public static final String ZMANAGER_COORDINATOR = ZMANAGER_ASSIGNMENTS + "/coordinators";
 
   public static final String ZGC = "/gc";
   public static final String ZGC_LOCK = ZGC + "/lock";

core/src/main/java/org/apache/accumulo/core/client/admin/InstanceOperations.java

@@ -247,6 +247,7 @@ Map<String,String> modifyProperties(Consumer<Map<String,String>> mapMutator)
    * @return set of servers of the supplied type
    * @since 4.0.0
    */
+  // TODO this needs to change for multiple managers
   Set<ServerId> getServers(ServerId.Type type);
 
   /**

core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java

@@ -25,6 +25,7 @@
 import java.util.function.Consumer;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
+import java.util.stream.Stream;
 
 import org.apache.accumulo.core.client.ConditionalWriter;
 import org.apache.accumulo.core.client.admin.TabletAvailability;
@@ -669,4 +670,24 @@ default void removeBulkLoadInProgressFlag(String path) {
   default ScanServerRefStore scanServerRefs() {
     throw new UnsupportedOperationException();
   }
+
+  record OrphanedCompaction(ExternalCompactionId id, TableId table, String dir) {
+  }
+
+  /**
+   * Tracks compactions that were removed from the metadata table but may still be running on
+   * compactors. The tmp files associated with these compactions can eventually be removed when the
+   * compaction is no longer running.
+   */
+  interface OrphanedCompactionStore {
+    Stream<OrphanedCompaction> list();
+
+    void add(Collection<OrphanedCompaction> orphanedCompactions);
+
+    void delete(Collection<OrphanedCompaction> orphanedCompactions);
+  }
+
+  default OrphanedCompactionStore orphanedCompactions() {
+    throw new UnsupportedOperationException();
+  }
 }

core/src/main/java/org/apache/accumulo/core/metadata/schema/MetadataSchema.java

@@ -547,4 +547,36 @@ public static String getRowPrefix() {
 
   }
 
+  /**
+   * Holds information about compactions that were deleted from tablets metadata by split or merge
+   * operations. These may still be running and may have tmp files that need to be cleaned up.
+   */
+  public static class OrphanedCompactionSection {
+    private static final Section section =
+        new Section(RESERVED_PREFIX + "ocomp", true, RESERVED_PREFIX + "ocomq", false);
+
+    public static Range getRange() {
+      return section.getRange();
+    }
+
+    public static String getRowPrefix() {
+      return section.getRowPrefix();
+    }
+
+    public static Ample.OrphanedCompaction decodeRow(String row) {
+      String[] fields = row.split("#");
+      Preconditions.checkArgument(fields.length == 4);
+      Preconditions.checkArgument(getRowPrefix().equals(fields[0]));
+      return new Ample.OrphanedCompaction(ExternalCompactionId.from(fields[1]),
+          TableId.of(fields[2]), fields[3]);
+    }
+
+    public static String encodeRow(Ample.OrphanedCompaction rc) {
+      // put the compaction id first in the row because its uuid will spread out nicely and avoid
+      // hot spotting
+      return getRowPrefix() + "#" + rc.id().canonical() + "#" + rc.table().canonical() + "#"
+          + rc.dir();
+
+    }
+  }
 }

core/src/main/java/org/apache/accumulo/core/util/compaction/ExternalCompactionUtil.java

@@ -24,7 +24,6 @@
 import static org.apache.accumulo.core.util.threads.ThreadPoolNames.COMPACTOR_RUNNING_COMPACTION_IDS_POOL;
 
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -262,7 +261,7 @@ public static List<ServerId> getCompactionsRunningOnCompactors(ClientContext con
     }
   }
 
-  public static Collection<ExternalCompactionId>
+  public static Set<ExternalCompactionId>
       getCompactionIdsRunningOnCompactors(ClientContext context) {
     final ExecutorService executor = ThreadPools.getServerThreadPools()
         .getPoolBuilder(COMPACTOR_RUNNING_COMPACTION_IDS_POOL).numCoreThreads(16).build();