Added ability to wake Compactors that might be in a long wait

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

@@ -1698,6 +1698,13 @@ public enum Property {
   COMPACTION_COORDINATOR_TSERVER_COMPACTION_CHECK_INTERVAL(
       "compaction.coordinator.tserver.check.interval", "1m", PropertyType.TIMEDURATION,
       "The interval at which to check the tservers for external compactions.", "2.1.0"),
+  @Experimental
+  COMPACTION_COORDINATOR_COMPACTOR_WAKEUP_THREADS("compaction.coordinator.compactor.wakeup.threads",
+      "0", PropertyType.COUNT,
+      "The number of threads the Coordinator should use to wake Compactors that are in a wait state. A value of zero"
+          + " 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"),
   // deprecated properties grouped at the end to reference property that replaces them
   @Deprecated(since = "1.6.0")
   @ReplacedBy(property = INSTANCE_VOLUMES)

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

@@ -19,6 +19,7 @@
 package org.apache.accumulo.core.util.compaction;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.accumulo.core.util.threads.ThreadPoolNames.COMPACTION_COORDINATOR_COMPACTOR_WAKE_POOL;
 import static org.apache.accumulo.core.util.threads.ThreadPoolNames.COMPACTOR_RUNNING_COMPACTIONS_POOL;
 import static org.apache.accumulo.core.util.threads.ThreadPoolNames.COMPACTOR_RUNNING_COMPACTION_IDS_POOL;
 
@@ -326,4 +327,22 @@ public static void cancelCompaction(ClientContext context, HostAndPort compactor
       ThriftUtil.returnClient(client, context);
     }
   }
+
+  public static void wakeCompactors(ClientContext context, List<HostAndPort> compactors,
+      int threads) {
+    final ExecutorService executor = ThreadPools.getServerThreadPools()
+        .getPoolBuilder(COMPACTION_COORDINATOR_COMPACTOR_WAKE_POOL).numCoreThreads(threads).build();
+    compactors.forEach(c -> {
+      CompactorService.Client client = null;
+      try {
+        client = ThriftUtil.getClient(ThriftClientTypes.COMPACTOR, c, context);
+        client.wake(TraceUtil.traceInfo(), context.rpcCreds());
+      } catch (TException e) {
+        LOG.debug("Failed to wake compactor {}", c);
+      } finally {
+        ThriftUtil.returnClient(client, context);
+      }
+    });
+    executor.shutdown();
+  }
 }

core/src/main/java/org/apache/accumulo/core/util/threads/ThreadPoolNames.java

@@ -28,6 +28,7 @@ public enum ThreadPoolNames {
   BULK_IMPORT_CLIENT_BULK_THREADS_POOL("accumulo.pool.bulk.import.client.bulk.threads"),
   BULK_IMPORT_DIR_MOVE_POOL("accumulo.pool.bulk.dir.move"),
   COMPACTION_COORDINATOR_SUMMARY_POOL("accumulo.pool.compaction.summary.gatherer"),
+  COMPACTION_COORDINATOR_COMPACTOR_WAKE_POOL("accumulo.pool.compaction.compactor.wake"),
   COMPACTION_SERVICE_COMPACTION_PLANNER_POOL("accumulo.pool.compaction.service.compaction.planner"),
   COMPACTOR_RUNNING_COMPACTIONS_POOL("accumulo.pool.compactor.running.compactions"),
   COMPACTOR_RUNNING_COMPACTION_IDS_POOL("accumulo.pool.compactor.running.compaction.ids"),