Single node META FATE data (apache#5127)

- Moved all the fate data for a single `META` transaction into a single ZK node
	- Pushed all the data into `NodeValue` (renamed to `FateData`)
	- Previously, `FateData` stored `TStatus`, `FateReservation`, and `FateKey`. Now additionally stores the `REPO` stack and `TxInfo`.
- Status enforcement added to `MetaFateStore` (previously only possible for `UserFateStore`).
- Moved `testFateInitialConfigCorrectness()` from `UserFateStoreIT` to `UserFateIT`
- Renamed `UserFateStoreIT` to `UserFateStatusEnforcementIT` (now extends a new class `FateStatusEnforcementIT`)
	- Now only tests status enforcement (previously status enforcement + `testFateInitialConfigCorrectness()`)
- Created `MetaFateStatusEnforcementIT` (extends `FateStatusEnforcementIT`)
	- Tests that the new status enforcement in `MetaFateStore` works
- Created `FateStoreUtil`, moving the `createFateTable()` util here, created `MetaFateZKSetup` inner class here (the counterpart to `createFateTable()` for `UserFateStore` but sets up ZooKeeper for use in `MetaFateStore`)
- Deleted `UserFateStoreIT`s (now `UserFateStatusEnforcementIT`) method `injectStatus` replacing with the existing `setStatus` which does the same thing
- Changed `StackOverflowException` to instead be a `RuntimeException` (previously `Exception`)
- Deleted unnecessary preexisting catch and immediate re-throw of a `StackOverflowException` in `MetaFateStore.FateTxStoreImpl.push(repo)`
- Cleaned up and refactored `MetaFateStore` methods which mutate existing FateData; now reuse same pattern across these methods: all call new method `MetaFateStore.mutate()`

Author: kevinrr888

core/src/main/java/org/apache/accumulo/core/fate/AbstractFateStore.java

@@ -34,6 +34,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -75,12 +76,22 @@ public FateId newRandomId(FateInstanceType instanceType) {
       return FateId.from(instanceType, UUID.randomUUID());
     }
   };
+  protected static final int MAX_REPOS = 100;
 
   // The ZooKeeper lock for the process that's running this store instance
   protected final ZooUtil.LockID lockID;
   protected final Predicate<ZooUtil.LockID> isLockHeld;
   protected final Map<FateId,CountDownTimer> deferred;
   protected final FateIdGenerator fateIdGenerator;
+  // the statuses required to perform operations
+  public static final Set<TStatus> REQ_PUSH_STATUS = Set.of(TStatus.IN_PROGRESS, TStatus.NEW);
+  public static final Set<TStatus> REQ_POP_STATUS =
+      Set.of(TStatus.FAILED_IN_PROGRESS, TStatus.SUCCESSFUL);
+  public static final Set<TStatus> REQ_DELETE_STATUS =
+      Set.of(TStatus.NEW, TStatus.SUBMITTED, TStatus.SUCCESSFUL, TStatus.FAILED);
+  // all but UNKNOWN
+  public static final Set<TStatus> REQ_FORCE_DELETE_STATUS = Set.of(TStatus.NEW, TStatus.SUBMITTED,
+      TStatus.SUCCESSFUL, TStatus.FAILED, TStatus.FAILED_IN_PROGRESS, TStatus.IN_PROGRESS);
   private final int maxDeferred;
   private final AtomicBoolean deferredOverflow = new AtomicBoolean();
 
@@ -415,7 +426,7 @@ protected void seededTx() {
     unreservedRunnableCount.increment();
   }
 
-  protected byte[] serializeTxInfo(Serializable so) {
+  protected static byte[] serializeTxInfo(Serializable so) {
     if (so instanceof String) {
       return ("S " + so).getBytes(UTF_8);
     } else {
@@ -428,7 +439,7 @@ protected byte[] serializeTxInfo(Serializable so) {
     }
   }
 
-  protected Serializable deserializeTxInfo(TxInfo txInfo, byte[] data) {
+  protected static Serializable deserializeTxInfo(TxInfo txInfo, byte[] data) {
     if (data[0] == 'O') {
       byte[] sera = new byte[data.length - 2];
       System.arraycopy(data, 2, sera, 0, sera.length);

core/src/main/java/org/apache/accumulo/core/fate/StackOverflowException.java

@@ -18,7 +18,7 @@
  */
 package org.apache.accumulo.core.fate;
 
-public class StackOverflowException extends Exception {
+public class StackOverflowException extends RuntimeException {
 
   public StackOverflowException(String msg) {
     super(msg);

core/src/main/java/org/apache/accumulo/core/fate/user/UserFateStore.java

@@ -71,9 +71,8 @@ public class UserFateStore<T> extends AbstractFateStore<T> {
   private final String tableName;
 
   private static final FateInstanceType fateInstanceType = FateInstanceType.USER;
-  private static final int maxRepos = 100;
   private static final com.google.common.collect.Range<Integer> REPO_RANGE =
-      com.google.common.collect.Range.closed(1, maxRepos);
+      com.google.common.collect.Range.closed(1, MAX_REPOS);
 
   public UserFateStore(ClientContext context, String tableName, ZooUtil.LockID lockID,
       Predicate<ZooUtil.LockID> isLockHeld) {
@@ -457,12 +456,12 @@ public void push(Repo<T> repo) throws StackOverflowException {
 
       Optional<Integer> top = findTop();
 
-      if (top.filter(t -> t >= maxRepos).isPresent()) {
+      if (top.filter(t -> t >= MAX_REPOS).isPresent()) {
         throw new StackOverflowException("Repo stack size too large");
       }
 
       FateMutator<T> fateMutator =
-          newMutator(fateId).requireStatus(TStatus.IN_PROGRESS, TStatus.NEW);
+          newMutator(fateId).requireStatus(REQ_PUSH_STATUS.toArray(TStatus[]::new));
       fateMutator.putRepo(top.map(t -> t + 1).orElse(1), repo).mutate();
     }
 
@@ -471,8 +470,8 @@ public void pop() {
       verifyReservedAndNotDeleted(true);
 
       Optional<Integer> top = findTop();
-      top.ifPresent(t -> newMutator(fateId)
-          .requireStatus(TStatus.FAILED_IN_PROGRESS, TStatus.SUCCESSFUL).deleteRepo(t).mutate());
+      top.ifPresent(t -> newMutator(fateId).requireStatus(REQ_POP_STATUS.toArray(TStatus[]::new))
+          .deleteRepo(t).mutate());
     }
 
     @Override
@@ -497,7 +496,7 @@ public void delete() {
       verifyReservedAndNotDeleted(true);
 
       var mutator = newMutator(fateId);
-      mutator.requireStatus(TStatus.NEW, TStatus.SUBMITTED, TStatus.SUCCESSFUL, TStatus.FAILED);
+      mutator.requireStatus(REQ_DELETE_STATUS.toArray(TStatus[]::new));
       mutator.delete().mutate();
       this.deleted = true;
     }
@@ -507,9 +506,7 @@ public void forceDelete() {
       verifyReservedAndNotDeleted(true);
 
       var mutator = newMutator(fateId);
-      // allow deletion of all txns other than UNKNOWN
-      mutator.requireStatus(TStatus.NEW, TStatus.SUBMITTED, TStatus.SUCCESSFUL, TStatus.FAILED,
-          TStatus.FAILED_IN_PROGRESS, TStatus.IN_PROGRESS);
+      mutator.requireStatus(REQ_FORCE_DELETE_STATUS.toArray(TStatus[]::new));
       mutator.delete().mutate();
       this.deleted = true;
     }
@@ -537,14 +534,14 @@ protected void unreserve() {
 
   static Text invertRepo(int position) {
     Preconditions.checkArgument(REPO_RANGE.contains(position),
-        "Position %s is not in the valid range of [0,%s]", position, maxRepos);
-    return new Text(String.format("%02d", maxRepos - position));
+        "Position %s is not in the valid range of [0,%s]", position, MAX_REPOS);
+    return new Text(String.format("%02d", MAX_REPOS - position));
   }
 
   static Integer restoreRepo(Text invertedPosition) {
-    int position = maxRepos - Integer.parseInt(invertedPosition.toString());
+    int position = MAX_REPOS - Integer.parseInt(invertedPosition.toString());
     Preconditions.checkArgument(REPO_RANGE.contains(position),
-        "Position %s is not in the valid range of [0,%s]", position, maxRepos);
+        "Position %s is not in the valid range of [0,%s]", position, MAX_REPOS);
     return position;
   }
 }