Partially moves compactions into the manager

assemble/bin/accumulo-cluster

@@ -459,16 +459,17 @@ gc:
 tserver:
   - localhost
 
+compaction:
+  compactor:
+    - user-small:
+        - localhost
+    - user-large:
+        - localhost
+
 #sserver:
 #  - default:
 #    - localhost
 #
-#compaction:
-#  compactor:
-#    - q1:
-#        - localhost
-#    - q2:
-#        - localhost
 #
 
 #

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

@@ -606,7 +606,7 @@ public enum Property {
       "The maximum number of files a compaction will open", "2.1.0"),
   TSERV_COMPACTION_SERVICE_DEFAULT_EXECUTORS(
       "tserver.compaction.major.service.default.planner.opts.executors",
-      "[{'name':'small','type':'internal','maxSize':'32M','numThreads':2},{'name':'medium','type':'internal','maxSize':'128M','numThreads':2},{'name':'large','type':'internal','numThreads':2}]"
+      ("[{'name':'small','type':'external','maxSize':'128M','queue':'user-small'}, {'name':'large','type':'external','queue':'user-large'}]")
           .replaceAll("'", "\""),
       PropertyType.STRING,
       "See {% jlink -f org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner %} ",

core/src/main/java/org/apache/accumulo/core/manager/state/TabletManagement.java

@@ -45,7 +45,7 @@ public class TabletManagement {
   public static final EnumSet<ColumnType> CONFIGURED_COLUMNS =
       EnumSet.of(ColumnType.PREV_ROW, ColumnType.LOCATION, ColumnType.SUSPEND, ColumnType.LOGS,
           ColumnType.CHOPPED, ColumnType.HOSTING_GOAL, ColumnType.HOSTING_REQUESTED,
-          ColumnType.FILES, ColumnType.LAST, ColumnType.OPID);
+          ColumnType.FILES, ColumnType.LAST, ColumnType.OPID, ColumnType.ECOMP, ColumnType.DIR);
 
   private static final Text REASONS_COLUMN_NAME = new Text("REASONS");
 

core/src/main/java/org/apache/accumulo/core/metadata/CompactableFileImpl.java

@@ -30,6 +30,8 @@ public class CompactableFileImpl implements CompactableFile {
   private final DataFileValue dataFileValue;
 
   public CompactableFileImpl(URI uri, long size, long entries) {
+    // TODO this normalizes the path passing it through URI defeating the purpose of
+    // StoredTabletFile
     this.storedTabletFile = new StoredTabletFile(uri.toString());
     this.dataFileValue = new DataFileValue(size, entries);
   }

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

@@ -443,6 +443,16 @@ interface ConditionalTabletMutator extends TabletUpdates<ConditionalTabletMutato
      */
     ConditionalTabletMutator requireHostingGoal(TabletHostingGoal tabletHostingGoal);
 
+    /**
+     * Requires the tablet to have no external compactions.
+     */
+    ConditionalTabletMutator requireAbsentCompactions();
+
+    /**
+     * Requires the specified external compaction to exists
+     */
+    ConditionalTabletMutator requireCompaction(ExternalCompactionId ecid);
+
     /**
      * <p>
      * Ample provides the following features on top of the conditional writer to help automate

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

@@ -28,6 +28,7 @@
 import com.google.common.base.Preconditions;
 import com.google.gson.Gson;
 
+// ELASTICITY_TODO remove this class, remove it from ample, add upgrade code to remove it from metadata table
 public class ExternalCompactionFinalState {
 
   private static final Gson GSON = new Gson();

core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactionExecutorId.java

@@ -28,6 +28,7 @@
  * @see org.apache.accumulo.core.spi.compaction
  */
 public class CompactionExecutorId extends AbstractId<CompactionExecutorId> {
+  // ELASTICITY_TODO make this cache ids like TableId. This will help save manager memory.
   private static final long serialVersionUID = 1L;
 
   protected CompactionExecutorId(String canonical) {

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

@@ -21,7 +21,9 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
+import java.util.function.Predicate;
 
+import org.apache.accumulo.core.client.PluginEnvironment;
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
 import org.apache.accumulo.core.conf.ConfigurationTypeHelper;
 import org.apache.accumulo.core.conf.Property;
@@ -49,13 +51,21 @@ private long getDefaultThroughput() {
         .getMemoryAsBytes(Property.TSERV_COMPACTION_SERVICE_DEFAULT_RATE_LIMIT.getDefaultValue());
   }
 
-  private Map<String,String> getConfiguration(AccumuloConfiguration aconf) {
+  private static Map<String,String> getConfiguration(AccumuloConfiguration aconf) {
     return aconf.getAllPropertiesWithPrefix(Property.TSERV_COMPACTION_SERVICE_PREFIX);
   }
 
+  public CompactionServicesConfig(PluginEnvironment.Configuration conf) {
+    // TODO will probably not need rate limit eventually and the 2nd param predicate can go away
+    this(conf.getWithPrefix(Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey()),
+        property -> conf.isSet(property.getKey()));
+  }
+
   public CompactionServicesConfig(AccumuloConfiguration aconf) {
-    Map<String,String> configs = getConfiguration(aconf);
+    this(getConfiguration(aconf), aconf::isPropertySet);
+  }
 
+  private CompactionServicesConfig(Map<String,String> configs, Predicate<Property> isSetPredicate) {
     configs.forEach((prop, val) -> {
 
       var suffix = prop.substring(Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey().length());
@@ -66,7 +76,7 @@ public CompactionServicesConfig(AccumuloConfiguration aconf) {
         planners.put(tokens[0], val);
       } else if (tokens.length == 3 && tokens[1].equals("rate") && tokens[2].equals("limit")) {
         var eprop = Property.getPropertyByKey(prop);
-        if (eprop == null || aconf.isPropertySet(eprop)) {
+        if (eprop == null || isSetPredicate.test(eprop)) {
           rateLimits.put(tokens[0], ConfigurationTypeHelper.getFixedMemoryAsBytes(val));
         }
       } else {
@@ -82,7 +92,6 @@ public CompactionServicesConfig(AccumuloConfiguration aconf) {
       throw new IllegalArgumentException(
           "Incomplete compaction service definitions, missing planner class " + diff);
     }
-
   }
 
   public long getRateLimit(String serviceName) {