Moves the compaction commit process into FATE (#4109)

The custom refresh tracking code was removed and compaction commit was
moved to being a FATE operation with the following 4 steps.

 1. Rename file done in RenameCompactionFile class
 2. Update the metadata table via a conditional mutation done in
   CommitCompaction class
 3. Write the gc candidates done in PutGcCandidates class
 4. Optionally send a RPC refresh request if the tablet was hosted
    done in RefreshTablet class

There is some follow on work that still needs to be done to improve
how this change works with detecting dead compactions.  After that is
done these changes should address problems outlined #3811 and #3802
that were related to process death before adding GC candidates.  Now
that GC candidates are written in FATE, if it dies it will run again
later.

This is currently storing the compaction commit FATE operations in
zookeeper.  This would not be suitable for a cluster because per tablet
information should never be stored in zookeeper.  However its fine as a
temporary situation in the elasticity branch until FATE storage is
availabe in an accumulo table, see #4049 and #3559

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

@@ -43,10 +43,6 @@ public class RootTable {
    * ZK path relative to the zookeeper node where the root tablet gc candidates are stored.
    */
   public static final String ZROOT_TABLET_GC_CANDIDATES = ZROOT_TABLET + "/gc_candidates";
-  /*
-   * ZK path relative to the zookeeper node where the root tablet refresh entries are stored.
-   */
-  public static final String ZROOT_TABLET_REFRESHES = ZROOT_TABLET + "/refreshes";
 
   public static final KeyExtent EXTENT = new KeyExtent(ID, null, null);
   public static final KeyExtent OLD_EXTENT =

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

@@ -21,7 +21,6 @@
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Set;
 import java.util.UUID;
 import java.util.function.Consumer;
@@ -675,46 +674,4 @@ default void addBulkLoadInProgressFlag(String path, long fateTxid) {
   default void removeBulkLoadInProgressFlag(String path) {
     throw new UnsupportedOperationException();
   }
-
-  interface Refreshes {
-    static class RefreshEntry {
-      private final ExternalCompactionId ecid;
-
-      private final KeyExtent extent;
-      private final TServerInstance tserver;
-
-      public RefreshEntry(ExternalCompactionId ecid, KeyExtent extent, TServerInstance tserver) {
-        this.ecid = Objects.requireNonNull(ecid);
-        this.extent = Objects.requireNonNull(extent);
-        this.tserver = Objects.requireNonNull(tserver);
-      }
-
-      public ExternalCompactionId getEcid() {
-        return ecid;
-      }
-
-      public KeyExtent getExtent() {
-        return extent;
-      }
-
-      public TServerInstance getTserver() {
-        return tserver;
-      }
-    }
-
-    void add(Collection<RefreshEntry> entries);
-
-    void delete(Collection<RefreshEntry> entries);
-
-    Stream<RefreshEntry> stream();
-  }
-
-  /**
-   * Refresh entries in the metadata table are used to track hosted tablets that need to have their
-   * metadata refreshed after a compaction. These entries ensure the refresh happens even in the
-   * case of process death.
-   */
-  default Refreshes refreshes(DataLevel dataLevel) {
-    throw new UnsupportedOperationException();
-  }
 }

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

@@ -533,17 +533,4 @@ public static String getRowPrefix() {
       return section.getRowPrefix();
     }
   }
-
-  public static class RefreshSection {
-    private static final Section section =
-        new Section(RESERVED_PREFIX + "refresh", true, RESERVED_PREFIX + "refresi", false);
-
-    public static Range getRange() {
-      return section.getRange();
-    }
-
-    public static String getRowPrefix() {
-      return section.getRowPrefix();
-    }
-  }
 }

server/base/src/main/java/org/apache/accumulo/server/init/ZooKeeperInitializer.java

@@ -47,7 +47,6 @@
 import org.apache.accumulo.server.conf.codec.VersionedProperties;
 import org.apache.accumulo.server.conf.store.SystemPropKey;
 import org.apache.accumulo.server.log.WalStateManager;
-import org.apache.accumulo.server.metadata.RefreshesImpl;
 import org.apache.accumulo.server.metadata.RootGcCandidates;
 import org.apache.accumulo.server.tables.TableManager;
 import org.apache.zookeeper.KeeperException;
@@ -137,8 +136,6 @@ void initialize(final ServerContext context, final boolean clearInstanceName,
         ZooUtil.NodeExistsPolicy.FAIL);
     zoo.putPersistentData(zkInstanceRoot + RootTable.ZROOT_TABLET_GC_CANDIDATES,
         new RootGcCandidates().toJson().getBytes(UTF_8), ZooUtil.NodeExistsPolicy.FAIL);
-    zoo.putPersistentData(zkInstanceRoot + RootTable.ZROOT_TABLET_REFRESHES,
-        RefreshesImpl.getInitialJson().getBytes(UTF_8), ZooUtil.NodeExistsPolicy.FAIL);
     zoo.putPersistentData(zkInstanceRoot + Constants.ZMANAGERS, EMPTY_BYTE_ARRAY,
         ZooUtil.NodeExistsPolicy.FAIL);
     zoo.putPersistentData(zkInstanceRoot + Constants.ZMANAGER_LOCK, EMPTY_BYTE_ARRAY,

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

@@ -338,10 +338,4 @@ public void deleteScanServerFileReferences(Collection<ScanServerRefTabletFile> r
       throw new IllegalStateException(e);
     }
   }
-
-  @Override
-  public Refreshes refreshes(DataLevel dataLevel) {
-    return new RefreshesImpl(context, dataLevel);
-  }
-
 }

server/manager/src/main/java/org/apache/accumulo/manager/Manager.java

@@ -941,7 +941,8 @@ public void run() {
     // Start the Manager's Fate Service
     fateServiceHandler = new FateServiceHandler(this);
     managerClientHandler = new ManagerClientServiceHandler(this);
-    compactionCoordinator = new CompactionCoordinator(context, tserverSet, security, nextEvent);
+    compactionCoordinator =
+        new CompactionCoordinator(context, tserverSet, security, nextEvent, fateRefs);
     // Start the Manager's Client service
     // Ensure that calls before the manager gets the lock fail
     ManagerClientService.Iface haProxy =