Remove GC candidates found to still be referenced

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

@@ -891,6 +891,11 @@ public enum Property {
       "The listening port for the garbage collector's monitor service", "1.3.5"),
   GC_DELETE_THREADS("gc.threads.delete", "16", PropertyType.COUNT,
       "The number of threads used to delete RFiles and write-ahead logs", "1.3.5"),
+  @Experimental
+  GC_REMOVE_IN_USE_CANDIDATES("gc.remove.in.use.candidates", "false", PropertyType.BOOLEAN,
+      "GC will remove deletion candidates that are in-use from the metadata location. "
+          + "This is expected to increase the speed of subsequent GC runs",
+      "2.1.3"),
   @Deprecated(since = "2.1.1", forRemoval = true)
   GC_TRASH_IGNORE("gc.trash.ignore", "false", PropertyType.BOOLEAN,
       "Do not use the Trash, even if it is configured.", "1.5.0"),

core/src/main/java/org/apache/accumulo/core/gc/GcCandidate.java

@@ -0,0 +1,72 @@
+/*
+ * 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.gc;
+
+import java.lang.Object;
+import java.util.Objects;
+
+public class GcCandidate implements Comparable<GcCandidate> {
+  private final long uid;
+  private final String path;
+
+  public GcCandidate(String path, long uid) {
+    this.path = path;
+    this.uid = uid;
+  }
+
+  public String getPath() {
+    return path;
+  }
+
+  public long getUid() {
+    return uid;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(path, uid);
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj) {
+      return true;
+    }
+    if (obj instanceof GcCandidate) {
+      GcCandidate candidate = (GcCandidate) obj;
+      return this.uid == candidate.getUid() && this.path.equals(candidate.getPath());
+    }
+    return false;
+  }
+
+  @Override
+  public int compareTo(GcCandidate candidate) {
+    var cmp = this.path.compareTo(candidate.getPath());
+    if (cmp == 0) {
+      return Long.compare(this.uid, candidate.getUid());
+    } else {
+      return cmp;
+    }
+  }
+
+  @Override
+  public String toString() {
+    return path + ", UUID: " + uid;
+  }
+}

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

@@ -27,6 +27,7 @@
 import org.apache.accumulo.core.data.TableId;
 import org.apache.accumulo.core.dataImpl.KeyExtent;
 import org.apache.accumulo.core.fate.zookeeper.ServiceLock;
+import org.apache.accumulo.core.gc.GcCandidate;
 import org.apache.accumulo.core.gc.ReferenceFile;
 import org.apache.accumulo.core.metadata.MetadataTable;
 import org.apache.accumulo.core.metadata.RootTable;
@@ -133,6 +134,24 @@ public enum ReadConsistency {
     EVENTUAL
   }
 
+  /**
+   * Enables status based processing of GcCandidates.
+   */
+  public enum GcCandidateType {
+    /**
+     * Candidates which have corresponding file references still present in tablet metadata.
+     */
+    INUSE,
+    /**
+     * Candidates that have no matching file references and can be removed from the system.
+     */
+    VALID,
+    /**
+     * Candidates that are malformed.
+     */
+    INVALID
+  }
+
   /**
    * Read a single tablets metadata. No checking is done for prev row, so it could differ. The
    * method will read the data using {@link ReadConsistency#IMMEDIATE}.
@@ -193,11 +212,15 @@ default void putGcFileAndDirCandidates(TableId tableId, Collection<ReferenceFile
     throw new UnsupportedOperationException();
   }
 
-  default void deleteGcCandidates(DataLevel level, Collection<String> paths) {
+  /**
+   * Enum added to support unique candidate deletions in 2.1
+   */
+  default void deleteGcCandidates(DataLevel level, Collection<GcCandidate> candidates,
+      GcCandidateType type) {
     throw new UnsupportedOperationException();
   }
 
-  default Iterator<String> getGcCandidates(DataLevel level) {
+  default Iterator<GcCandidate> getGcCandidates(DataLevel level) {
     throw new UnsupportedOperationException();
   }
 

server/base/src/main/java/org/apache/accumulo/server/metadata/RootGcCandidates.java

@@ -20,12 +20,14 @@
 
 import static com.google.common.base.Preconditions.checkArgument;
 
+import java.security.SecureRandom;
 import java.util.SortedMap;
 import java.util.SortedSet;
 import java.util.TreeMap;
 import java.util.TreeSet;
 import java.util.stream.Stream;
 
+import org.apache.accumulo.core.gc.GcCandidate;
 import org.apache.accumulo.core.metadata.StoredTabletFile;
 import org.apache.hadoop.fs.Path;
 
@@ -78,20 +80,23 @@ public void add(Stream<StoredTabletFile> refs) {
         .add(ref.getFileName()));
   }
 
-  public void remove(Stream<String> refs) {
-    refs.map(Path::new).forEach(
-        path -> data.candidates.computeIfPresent(path.getParent().toString(), (key, values) -> {
-          values.remove(path.getName());
-          return values.isEmpty() ? null : values;
-        }));
+  public void remove(Stream<GcCandidate> refs) {
+    refs.map(GcCandidate::getPath).map(Path::new).forEach(path -> {
+      data.candidates.computeIfPresent(path.getParent().toString(), (key, values) -> {
+        values.remove(path.getName());
+        return values.isEmpty() ? null : values;
+      });
+    });
   }
 
-  public Stream<String> sortedStream() {
+  public Stream<GcCandidate> sortedStream() {
+    var uidGen = new SecureRandom();
     return data.candidates.entrySet().stream().flatMap(entry -> {
       String parent = entry.getKey();
       SortedSet<String> names = entry.getValue();
-      return names.stream().map(name -> new Path(parent, name));
-    }).map(Path::toString).sorted();
+      return names.stream()
+          .map(name -> new GcCandidate(parent + Path.SEPARATOR + name, uidGen.nextLong()));
+    }).sorted();
   }
 
   public String toJson() {

server/base/src/main/java/org/apache/accumulo/server/metadata/ServerAmpleImpl.java

@@ -45,6 +45,7 @@
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.dataImpl.KeyExtent;
 import org.apache.accumulo.core.fate.FateTxId;
+import org.apache.accumulo.core.gc.GcCandidate;
 import org.apache.accumulo.core.gc.ReferenceFile;
 import org.apache.accumulo.core.metadata.MetadataTable;
 import org.apache.accumulo.core.metadata.RootTable;
@@ -210,17 +211,22 @@ public void removeBulkLoadEntries(TableId tableId, long tid, Text firstSplit, Te
   }
 
   @Override
-  public void deleteGcCandidates(DataLevel level, Collection<String> paths) {
+  public void deleteGcCandidates(DataLevel level, Collection<GcCandidate> candidates,
+      GcCandidateType type) {
 
     if (level == DataLevel.ROOT) {
-      mutateRootGcCandidates(rgcc -> rgcc.remove(paths.stream()));
+      if (type == GcCandidateType.INUSE) {
+        // Deletion of INUSE candidates is not supported in 2.1.x.
+        return;
+      }
+      mutateRootGcCandidates(rgcc -> rgcc.remove(candidates.stream()));
       return;
     }
 
     try (BatchWriter writer = context.createBatchWriter(level.metaTable())) {
-      for (String path : paths) {
-        Mutation m = new Mutation(DeletesSection.encodeRow(path));
-        m.putDelete(EMPTY_TEXT, EMPTY_TEXT);
+      for (GcCandidate candidate : candidates) {
+        Mutation m = new Mutation(DeletesSection.encodeRow(candidate.getPath()));
+        m.putDelete(EMPTY_TEXT, EMPTY_TEXT, candidate.getUid());
         writer.addMutation(m);
       }
     } catch (MutationsRejectedException | TableNotFoundException e) {
@@ -229,7 +235,7 @@ public void deleteGcCandidates(DataLevel level, Collection<String> paths) {
   }
 
   @Override
-  public Iterator<String> getGcCandidates(DataLevel level) {
+  public Iterator<GcCandidate> getGcCandidates(DataLevel level) {
     if (level == DataLevel.ROOT) {
       var zooReader = context.getZooReader();
       byte[] jsonBytes;
@@ -251,7 +257,10 @@ public Iterator<String> getGcCandidates(DataLevel level) {
       }
       scanner.setRange(range);
       return scanner.stream().filter(entry -> entry.getValue().equals(SkewedKeyValue.NAME))
-          .map(entry -> DeletesSection.decodeRow(entry.getKey().getRow().toString())).iterator();
+          .map(
+              entry -> new GcCandidate(DeletesSection.decodeRow(entry.getKey().getRow().toString()),
+                  entry.getKey().getTimestamp()))
+          .iterator();
     } else {
       throw new IllegalArgumentException();
     }

server/base/src/main/java/org/apache/accumulo/server/util/ListVolumesUsed.java

@@ -22,6 +22,7 @@
 import java.util.TreeSet;
 
 import org.apache.accumulo.core.conf.SiteConfiguration;
+import org.apache.accumulo.core.gc.GcCandidate;
 import org.apache.accumulo.core.metadata.schema.Ample;
 import org.apache.accumulo.core.metadata.schema.TabletMetadata;
 import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
@@ -78,9 +79,9 @@ private static void listTable(Ample.DataLevel level, ServerContext context) thro
         + " deletes section (volume replacement occurs at deletion time)");
     volumes.clear();
 
-    Iterator<String> delPaths = context.getAmple().getGcCandidates(level);
+    Iterator<GcCandidate> delPaths = context.getAmple().getGcCandidates(level);
     while (delPaths.hasNext()) {
-      volumes.add(getTableURI(delPaths.next()));
+      volumes.add(getTableURI(delPaths.next().getPath()));
     }
     for (String volume : volumes) {
       System.out.println("\tVolume : " + volume);