Avoids holding tablet metadata for queued compactions

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()