Avoids using tablet metadata for compaction reservation

This change fixes apache#5188.  Unfortunately it touches a lot of code because
of cascading dependencies in the code.  It would be difficult to break
this into a smaller commit.  These changes do reduce some of those
dependencies though.

There are two major advantages after this change.  First tablet metadata
is no longer kept in memory for queued compactions.  Second the tablet
metadata is no longer read during compaction reservation.  Before this
change the following would happen.

 1. TGW would find a tablet to compact and enqueue compaction
    jobs+tablet metadata.
 2. Eventually when a compactor requested a job it would yank job+tablet
    metadata off the queue.
 3. To reserve the compaction a lot of complex analysis was done in the
    coordinator and then a conditional mutation was submitted.  The
    conditional mutation would require all data involved in the complex
    analysis to be the same.
 4. If the conditional mutation failed then the coordinator would
    reread the tablet metadata and go back to step 3.

After this change the following happens in the code.

 1. TGW would find a tablet to compact and enqueue a new class called
    ResolvedCompactionJob.  This new class takes the compaction job and
    tablet metadata and computes all information needed for the compaction
    later.  The TabletMetadata object is no longer refrenced by this class
    after the constructor returns.  So this class will use much less memory
    on the queue for the case when tablet have lots of files.
 2. Eventually when a compactor requested a job it would yank a
    ResolvedCompactionJob off the queue.
 3. All of the complex analysis to determine if a compaction can start
    is now done in the conditional mutation instead of in the
    coordinator.  To enable this, new functionality was added to Ample
    including the new TabletMetadataCheck interface,
    TabletMetadataCheckIterator, and the CompactionReservationCheck class.
    With these changes its now super easy to write a conditional check that
    will do arbitrary analysis of TabletMetadata prior to committing a
    mutation.
 4. Since the analysis is done in the conditional mutation there is no
    longer a need to retry.  If the mutation fails then we know the
    compaction can not run.

The following were some supporting changes that had to be made.

 * Took methods for encoding KeyExtent as base64 from
   TabletManagementParameters and moved the KeyExtent because this was
   needed in the new TabletMetadataCheckIterator.
 * Move TabletMetadata out of the compaction queues, which made those
   more independent but was a big change.  The main change here is that
   instead of adding `TabletMetadata, List<CompactionJob>` to the
   compaction queue now `KeyExtent, List<CompactionJob>` is added.  This
   required changing the test for these classes and the code that interacts
   with them in CompactionCoordinator creating lots of diff.
 * Moved CompactionCoordinatorTest.testCanReserve() to
   CompactionReservationCheckTest.testCanReserve() and changed the code
   to work with the new CompactionReservationCheck class.  These are test
   of the complex logic that used to run in the coordinator and now runs in
   the tablet server as part of a conditional mutation.
 * Removed conditional checks from Ample related to compactions that
   were no longer used after these changes.  There was code for
   requiring a set of files not not be compacting.

The new TabletMetadataCheck functionality of ample could be used to
simplify other conditional mutation checks of tablet metadata.  It made
the compaction reservation code much simpler and easier to understand.
This code could be further improved if apache#5244 were implemented.

core/src/main/java/org/apache/accumulo/core/dataImpl/KeyExtent.java

@@ -28,9 +28,11 @@
 import java.io.DataOutput;
 import java.io.DataOutputStream;
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
+import java.util.Arrays;
 import java.util.Base64;
 import java.util.Collections;
 import java.util.Comparator;
@@ -55,6 +57,8 @@
 import org.apache.accumulo.core.util.Pair;
 import org.apache.accumulo.core.util.TextUtil;
 import org.apache.hadoop.io.BinaryComparable;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.io.DataOutputBuffer;
 import org.apache.hadoop.io.Text;
 
 /**
@@ -560,4 +564,26 @@ public String obscured() {
     return Base64.getEncoder().encodeToString(digester.digest());
   }
 
+  public String toBase64() {
+    DataOutputBuffer buffer = new DataOutputBuffer();
+    try {
+      writeTo(buffer);
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+
+    return Base64.getEncoder().encodeToString(Arrays.copyOf(buffer.getData(), buffer.getLength()));
+  }
+
+  public static KeyExtent fromBase64(String encoded) {
+    byte[] data = Base64.getDecoder().decode(encoded);
+    DataInputBuffer buffer = new DataInputBuffer();
+    buffer.reset(data, data.length);
+    try {
+      return KeyExtent.readFrom(buffer);
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
 }

core/src/main/java/org/apache/accumulo/core/logging/TabletLogger.java

@@ -36,7 +36,6 @@
 import org.apache.accumulo.core.metadata.TServerInstance;
 import org.apache.accumulo.core.metadata.TabletFile;
 import org.apache.accumulo.core.metadata.schema.ExternalCompactionId;
-import org.apache.accumulo.core.metadata.schema.TabletMetadata;
 import org.apache.accumulo.core.spi.compaction.CompactionJob;
 import org.apache.accumulo.core.spi.compaction.CompactionKind;
 import org.apache.accumulo.core.tabletserver.log.LogEntry;
@@ -147,20 +146,18 @@ public static void selected(FateId fateId, KeyExtent extent,
         Collections2.transform(inputs, StoredTabletFile::toMinimalString));
   }
 
-  public static void compacting(TabletMetadata tabletMetadata, ExternalCompactionId cid,
+  public static void compacting(KeyExtent extent, FateId selectedFateId, ExternalCompactionId cid,
       String compactorAddress, CompactionJob job) {
     if (fileLog.isDebugEnabled()) {
       if (job.getKind() == CompactionKind.USER) {
-        var fateId = tabletMetadata.getSelectedFiles().getFateId();
         fileLog.debug(
             "Compacting {} driver:{} id:{} group:{} compactor:{} priority:{} size:{} kind:{} files:{}",
-            tabletMetadata.getExtent(), fateId, cid, job.getGroup(), compactorAddress,
-            job.getPriority(), getSize(job.getFiles()), job.getKind(),
-            asMinimalString(job.getFiles()));
+            extent, selectedFateId, cid, job.getGroup(), compactorAddress, job.getPriority(),
+            getSize(job.getFiles()), job.getKind(), asMinimalString(job.getFiles()));
       } else {
         fileLog.debug(
             "Compacting {} id:{} group:{} compactor:{} priority:{} size:{} kind:{} files:{}",
-            tabletMetadata.getExtent(), cid, job.getGroup(), compactorAddress, job.getPriority(),
+            extent, cid, job.getGroup(), compactorAddress, job.getPriority(),
             getSize(job.getFiles()), job.getKind(), asMinimalString(job.getFiles()));
       }
     }

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

@@ -534,9 +534,10 @@ ConditionalTabletMutator requireSame(TabletMetadata tabletMetadata, ColumnType t
     ConditionalTabletMutator requireAbsentLoaded(Set<ReferencedTabletFile> files);
 
     /**
-     * Requires the given set of files are not currently involved in any running compactions.
+     * This check will run atomically on the server side and must pass in order for the tablet to be
+     * updated.
      */
-    ConditionalTabletMutator requireNotCompacting(Set<StoredTabletFile> files);
+    ConditionalTabletMutator requireCheckSuccess(TabletMetadataCheck check);
 
     /**
      * <p>

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

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.core.metadata.schema;
+
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.Set;
+
+/**
+ * This interface facilitates atomic checks of tablet metadata prior to updating tablet metadata.
+ * The way it is intended to be used is the following.
+ * <ol>
+ * <li>On the client side a TabletMetadataCheck object is created and passed to Ample</li>
+ * <li>Ample uses Gson to serialize the object, so it must not reference anything that can not
+ * serialize. Also it should not reference big things like server context or the Manager, it should
+ * only reference a small amount of data needed for the check.
+ * <li>On the tablet server side, as part of conditional mutation processing, this class is
+ * recreated and the {@link #canUpdate(TabletMetadata)} method is called and if it returns true the
+ * conditional mutation will go through.</li>
+ * </ol>
+ *
+ * <p>
+ * Implementations are expected to have a no arg constructor.
+ * </p>
+ *
+ */
+public interface TabletMetadataCheck {
+
+  Set<TabletMetadata.ColumnType> ALL_COLUMNS =
+      Collections.unmodifiableSet(EnumSet.noneOf(TabletMetadata.ColumnType.class));
+
+  boolean canUpdate(TabletMetadata tabletMetadata);
+
+  /**
+   * Determines what tablet metadata columns are read on the server side. Return the empty set to
+   * read all tablet metadata columns.
+   */
+  default Set<TabletMetadata.ColumnType> columnsToRead() {
+    return ALL_COLUMNS;
+  }
+}

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

@@ -20,7 +20,6 @@
 
 import java.util.Collection;
 import java.util.Objects;
-import java.util.Optional;
 import java.util.Set;
 
 import org.apache.accumulo.core.client.admin.compaction.CompactableFile;
@@ -40,22 +39,13 @@ public class CompactionJobImpl implements CompactionJob {
   private final CompactorGroupId group;
   private final Set<CompactableFile> files;
   private final CompactionKind kind;
-  // Tracks if a job selected all the tablet's files that existed at the time the job was created.
-  private final Optional<Boolean> jobSelectedAll;
 
-  /**
-   *
-   * @param jobSelectedAll This parameters only needs to be non-empty for job objects that are used
-   *        to start compaction. After a job is running, its not used. So when a job object is
-   *        recreated for a running external compaction this parameter can be empty.
-   */
   public CompactionJobImpl(short priority, CompactorGroupId group,
-      Collection<CompactableFile> files, CompactionKind kind, Optional<Boolean> jobSelectedAll) {
+      Collection<CompactableFile> files, CompactionKind kind) {
     this.priority = priority;
     this.group = Objects.requireNonNull(group);
     this.files = Set.copyOf(files);
     this.kind = Objects.requireNonNull(kind);
-    this.jobSelectedAll = Objects.requireNonNull(jobSelectedAll);
   }
 
   @Override
@@ -92,10 +82,6 @@ public int hashCode() {
     return Objects.hash(priority, group, files, kind);
   }
 
-  public boolean selectedAll() {
-    return jobSelectedAll.orElseThrow();
-  }
-
   @Override
   public boolean equals(Object o) {
     if (o instanceof CompactionJobImpl) {

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

@@ -23,7 +23,6 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Optional;
 import java.util.Set;
 
 import org.apache.accumulo.core.client.admin.compaction.CompactableFile;
@@ -80,8 +79,7 @@ public Builder addJob(short priority, CompactorGroupId group,
 
       seenFiles.addAll(filesSet);
 
-      jobs.add(new CompactionJobImpl(priority, group, filesSet, kind,
-          Optional.of(filesSet.equals(allFiles))));
+      jobs.add(new CompactionJobImpl(priority, group, filesSet, kind));
       return this;
     }
 

core/src/test/java/org/apache/accumulo/core/spi/compaction/RatioBasedCompactionPlannerTest.java

@@ -34,7 +34,6 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.Set;
 import java.util.stream.IntStream;
 
@@ -632,7 +631,7 @@ public void testMaxTabletFilesNoCompaction() {
     // that a compaction is not planned
     all = createCFs(1_000, 2, 2, 2, 2, 2, 2, 2);
     var job = new CompactionJobImpl((short) 1, CompactorGroupId.of("ee1"), createCFs("F1", "1000"),
-        CompactionKind.SYSTEM, Optional.of(false));
+        CompactionKind.SYSTEM);
     params = createPlanningParams(all, all, Set.of(job), 3, CompactionKind.SYSTEM, conf);
     plan = planner.makePlan(params);
 

core/src/test/java/org/apache/accumulo/core/util/compaction/CompactionPrioritizerTest.java

@@ -37,7 +37,6 @@
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
-import java.util.Optional;
 
 import org.apache.accumulo.core.client.admin.compaction.CompactableFile;
 import org.apache.accumulo.core.clientImpl.Namespace;
@@ -60,10 +59,9 @@ public CompactionJob createJob(CompactionKind kind, String tablet, int numFiles,
       files.add(CompactableFile
           .create(URI.create("hdfs://foonn/accumulo/tables/5/" + tablet + "/" + i + ".rf"), 4, 4));
     }
-    return new CompactionJobImpl(
-        CompactionJobPrioritizer.createPriority(Namespace.DEFAULT.id(), TableId.of("5"), kind,
-            totalFiles, numFiles, totalFiles * 2),
-        CompactorGroupId.of("test"), files, kind, Optional.of(false));
+    return new CompactionJobImpl(CompactionJobPrioritizer.createPriority(Namespace.DEFAULT.id(),
+        TableId.of("5"), kind, totalFiles, numFiles, totalFiles * 2), CompactorGroupId.of("test"),
+        files, kind);
   }
 
   @Test

server/base/src/main/java/org/apache/accumulo/server/compaction/CompactionJobGenerator.java

@@ -25,7 +25,6 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
-import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -285,7 +284,7 @@ public Collection<CompactionJob> getRunningCompactions() {
           Collection<CompactableFile> files = ecMeta.getJobFiles().stream()
               .map(f -> new CompactableFileImpl(f, allFiles2.get(f))).collect(Collectors.toList());
           CompactionJob job = new CompactionJobImpl(ecMeta.getPriority(),
-              ecMeta.getCompactionGroupId(), files, ecMeta.getKind(), Optional.empty());
+              ecMeta.getCompactionGroupId(), files, ecMeta.getKind());
           return job;
         }).collect(Collectors.toUnmodifiableList());
       }

server/base/src/main/java/org/apache/accumulo/server/compaction/CompactionPluginUtils.java

@@ -27,6 +27,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 import org.apache.accumulo.core.classloader.ClassLoaderUtil;
@@ -211,7 +212,7 @@ public Optional<SortedKeyValueIterator<Key,Value>> getSample(CompactableFile cf,
 
   public static Map<String,String> computeOverrides(Optional<CompactionConfig> compactionConfig,
       ServerContext context, KeyExtent extent, Set<CompactableFile> inputFiles,
-      Set<CompactableFile> selectedFiles) {
+      Supplier<Set<CompactableFile>> selectedFiles) {
 
     if (compactionConfig.isPresent()
         && !UserCompactionUtils.isDefault(compactionConfig.orElseThrow().getConfigurer())) {
@@ -234,7 +235,8 @@ public static Map<String,String> computeOverrides(Optional<CompactionConfig> com
   }
 
   public static Map<String,String> computeOverrides(ServerContext context, KeyExtent extent,
-      Set<CompactableFile> inputFiles, Set<CompactableFile> selectedFiles, PluginConfig cfg) {
+      Set<CompactableFile> inputFiles, Supplier<Set<CompactableFile>> selectedFiles,
+      PluginConfig cfg) {
 
     CompactionConfigurer configurer = newInstance(context.getTableConfiguration(extent.tableId()),
         cfg.getClassName(), CompactionConfigurer.class);
@@ -266,7 +268,7 @@ public Collection<CompactableFile> getInputFiles() {
 
       @Override
       public Set<CompactableFile> getSelectedFiles() {
-        return selectedFiles;
+        return selectedFiles.get();
       }
 
       @Override

server/base/src/main/java/org/apache/accumulo/server/manager/state/TabletManagementParameters.java

@@ -24,11 +24,7 @@
 import static java.util.stream.Collectors.toUnmodifiableMap;
 import static java.util.stream.Collectors.toUnmodifiableSet;
 
-import java.io.IOException;
-import java.io.UncheckedIOException;
 import java.net.URI;
-import java.util.Arrays;
-import java.util.Base64;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
@@ -49,8 +45,6 @@
 import org.apache.accumulo.core.util.time.SteadyTime;
 import org.apache.accumulo.server.manager.LiveTServerSet;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.DataInputBuffer;
-import org.apache.hadoop.io.DataOutputBuffer;
 
 import com.google.common.base.Suppliers;
 import com.google.gson.Gson;
@@ -120,7 +114,7 @@ private TabletManagementParameters(JsonData jdata) {
     this.serversToShutdown =
         jdata.serversToShutdown.stream().map(TServerInstance::new).collect(toUnmodifiableSet());
     this.migrations = jdata.migrations.entrySet().stream()
-        .collect(toUnmodifiableMap(entry -> JsonData.strToExtent(entry.getKey()),
+        .collect(toUnmodifiableMap(entry -> KeyExtent.fromBase64(entry.getKey()),
             entry -> new TServerInstance(entry.getValue())));
     this.level = jdata.level;
     this.compactionHints = makeImmutable(jdata.compactionHints.entrySet().stream()
@@ -226,30 +220,6 @@ private static class JsonData {
     Map<URI,URI> volumeReplacements;
     long steadyTime;
 
-    private static String toString(KeyExtent extent) {
-      DataOutputBuffer buffer = new DataOutputBuffer();
-      try {
-        extent.writeTo(buffer);
-      } catch (IOException e) {
-        throw new UncheckedIOException(e);
-      }
-
-      return Base64.getEncoder()
-          .encodeToString(Arrays.copyOf(buffer.getData(), buffer.getLength()));
-
-    }
-
-    private static KeyExtent strToExtent(String kes) {
-      byte[] data = Base64.getDecoder().decode(kes);
-      DataInputBuffer buffer = new DataInputBuffer();
-      buffer.reset(data, data.length);
-      try {
-        return KeyExtent.readFrom(buffer);
-      } catch (IOException e) {
-        throw new UncheckedIOException(e);
-      }
-    }
-
     // Gson requires private constructor
     @SuppressWarnings("unused")
     private JsonData() {}
@@ -262,8 +232,9 @@ private JsonData() {}
           .collect(toList());
       serversToShutdown = params.serversToShutdown.stream().map(TServerInstance::getHostPortSession)
           .collect(toList());
-      migrations = params.migrations.entrySet().stream().collect(
-          toMap(entry -> toString(entry.getKey()), entry -> entry.getValue().getHostPortSession()));
+      migrations =
+          params.migrations.entrySet().stream().collect(toMap(entry -> entry.getKey().toBase64(),
+              entry -> entry.getValue().getHostPortSession()));
       level = params.level;
       tserverGroups = params.getGroupedTServers().entrySet().stream()
           .collect(toMap(Map.Entry::getKey, entry -> entry.getValue().stream()