Distributes compaction coordination across managers

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/conf/Property.java

@@ -1445,6 +1445,16 @@ start with the category prefix, followed by a scope (minc, majc, scan, \
           + " disables Compactor wake up. Enabling this feature will cause Compactors in a long wait state (see"
           + " COMPACTOR_MAX_JOB_WAIT_TIME) to check in with the Coordinator for work.",
       "2.1.7"),
+  COMPACTION_COORDINATOR_ASSIGNMENT_MIN("compaction.coordinator.assignment.sleep.min", "100ms",
+      PropertyType.TIMEDURATION,
+      "When there are multiple managers compaction resource groups are assigned across manager. This is the "
+          + "minimum time between checks to see if assignments need to be adjusted.",
+      "4.0.0"),
+  COMPACTION_COORDINATOR_ASSIGNMENT_MAX("compaction.coordinator.assignment.sleep.max", "3m",
+      PropertyType.TIMEDURATION,
+      "When there are multiple managers compaction resource groups are assigned across manager. This is the "
+          + "maximum time between checks to see if assignments need to be adjusted.",
+      "4.0.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");

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

@@ -37,6 +37,7 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
+import java.util.function.Predicate;
 
 import org.apache.accumulo.core.client.admin.servers.ServerId;
 import org.apache.accumulo.core.clientImpl.ClientContext;
@@ -223,6 +224,12 @@ public static void getCompactionsRunningOnCompactors(ClientContext context,
     }
   }
 
+  public static List<ServerId> getCompactionsRunningOnCompactors(ClientContext context,
+      ExecutorService executor, Consumer<TExternalCompaction> consumer)
+      throws InterruptedException {
+    return getCompactionsRunningOnCompactors(context, executor, rg -> true, consumer);
+  }
+
   /**
    * This method returns information from the Compactors about the job that is currently running.
    *
@@ -232,12 +239,12 @@ public static void getCompactionsRunningOnCompactors(ClientContext context,
    * @return list of compactor addresses where RPC failed
    */
   public static List<ServerId> getCompactionsRunningOnCompactors(ClientContext context,
-      ExecutorService executor, Consumer<TExternalCompaction> consumer)
-      throws InterruptedException {
+      ExecutorService executor, Predicate<ResourceGroupId> rgPredicate,
+      Consumer<TExternalCompaction> consumer) throws InterruptedException {
 
     final List<ServerId> failures = new ArrayList<>();
-    final Set<ServerId> compactors =
-        context.instanceOperations().getServers(ServerId.Type.COMPACTOR);
+    final Set<ServerId> compactors = context.instanceOperations()
+        .getServers(ServerId.Type.COMPACTOR, rgPredicate, (host, port) -> true);
     final List<RpcFuture<TExternalCompaction>> rcFutures = new ArrayList<>(compactors.size());
 
     try {