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/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 {