Modify FateStore to introduce a Seeder API (apache#5374)

This adds a new Seeder API that can be used to seed transactions with a
FateKey. The purpose of this new API is to allow for eventual batching
of multiple seeding attempts together into one conditional mutation to
improve performance. This change just updates the api and all calls to
the seeder to attempt seeding will be performed individually. A future
update will support the new functionality of batching and commiting all
the changes together when the Seeder is closed.

This change is for Part 1 of apache#5160

Author: cshannon

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

@@ -540,7 +540,9 @@ public FateId startTransaction() {
 
   public void seedTransaction(FateOperation fateOp, FateKey fateKey, Repo<T> repo,
       boolean autoCleanUp) {
-    store.seedTransaction(fateOp, fateKey, repo, autoCleanUp);
+    try (var seeder = store.beginSeeding()) {
+      var unused = seeder.attemptToSeedTransaction(fateOp, fateKey, repo, autoCleanUp);
+    }
   }
 
   // start work in the transaction.. it is safe to call this

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

@@ -27,7 +27,9 @@
 import java.util.Arrays;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
 
 import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
 import org.apache.hadoop.io.DataInputBuffer;
@@ -49,25 +51,42 @@ public interface FateStore<T> extends ReadOnlyFateStore<T> {
    */
   FateId create();
 
-  /**
-   * Seeds a transaction with the given repo if it does not exist. A fateId will be derived from the
-   * fateKey. If seeded, sets the following data for the fateId in the store.
-   *
-   * <ul>
-   * <li>Set the fate op</li>
-   * <li>Set the status to SUBMITTED</li>
-   * <li>Set the fate key</li>
-   * <li>Sets autocleanup only if true</li>
-   * <li>Sets the creation time</li>
-   * </ul>
-   *
-   * @return The return type is only intended for testing it may not be correct in the face of
-   *         failures. When there are no failures returns optional w/ the fate id set if seeded and
-   *         empty optional otherwise. If there was a failure this could return an empty optional
-   *         when it actually succeeded.
-   */
-  Optional<FateId> seedTransaction(Fate.FateOperation fateOp, FateKey fateKey, Repo<T> repo,
-      boolean autoCleanUp);
+  interface Seeder<T> extends AutoCloseable {
+
+    /**
+     * Attempts to seed a transaction with the given repo if it does not exist. A fateId will be
+     * derived from the fateKey. If seeded, sets the following data for the fateId in the store.
+     *
+     * TODO: Support completing futures later in close method The current version will always return
+     * with a CompleteableFuture that is already completed. Future version will process will
+     * complete in the close() method for the User store.
+     *
+     * <ul>
+     * <li>Set the fate op</li>
+     * <li>Set the status to SUBMITTED</li>
+     * <li>Set the fate key</li>
+     * <li>Sets autocleanup only if true</li>
+     * <li>Sets the creation time</li>
+     * </ul>
+     *
+     * @return The return type is only intended for testing it may not be correct in the face of
+     *         failures. When there are no failures returns optional w/ the fate id set if seeded
+     *         and empty optional otherwise. If there was a failure this could return an empty
+     *         optional when it actually succeeded.
+     */
+    CompletableFuture<Optional<FateId>> attemptToSeedTransaction(Fate.FateOperation fateOp,
+        FateKey fateKey, Repo<T> repo, boolean autoCleanUp);
+
+    // TODO: Right now all implementations do nothing
+    // Eventually this would check the status of all added conditional mutations,
+    // retry unknown, and then close the conditional writer.
+    @Override
+    void close();
+  }
+
+  // Creates a conditional writer for the user fate store. For Zookeeper all this code will probably
+  // do the same thing its currently doing as zookeeper does not support multi-node operations.
+  Seeder<T> beginSeeding();
 
   /**
    * Seeds a transaction with the given repo if its current status is NEW and it is currently

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

@@ -27,6 +27,7 @@
 import java.util.Optional;
 import java.util.SortedMap;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
@@ -41,6 +42,7 @@
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.fate.AbstractFateStore;
 import org.apache.accumulo.core.fate.Fate;
+import org.apache.accumulo.core.fate.Fate.FateOperation;
 import org.apache.accumulo.core.fate.Fate.TxInfo;
 import org.apache.accumulo.core.fate.FateId;
 import org.apache.accumulo.core.fate.FateInstanceType;
@@ -133,7 +135,28 @@ public FateId getFateId() {
   }
 
   @Override
-  public Optional<FateId> seedTransaction(Fate.FateOperation fateOp, FateKey fateKey, Repo<T> repo,
+  public Seeder<T> beginSeeding() {
+    // TODO: For now can handle seeding 1 transaction at a time so just process
+    // everything in attemptToSeedTransaction
+    // Part 2 of the changes in #5160 will allow multiple seeding attempts to be combined
+    // into one conditional mutation and we will need to track the pending operations
+    // and futures in a map
+    return new Seeder<T>() {
+      @Override
+      public CompletableFuture<Optional<FateId>> attemptToSeedTransaction(FateOperation fateOp,
+          FateKey fateKey, Repo<T> repo, boolean autoCleanUp) {
+        return CompletableFuture
+            .completedFuture(seedTransaction(fateOp, fateKey, repo, autoCleanUp));
+      }
+
+      @Override
+      public void close() {
+        // TODO: This will be used in Part 2 of #5160
+      }
+    };
+  }
+
+  private Optional<FateId> seedTransaction(Fate.FateOperation fateOp, FateKey fateKey, Repo<T> repo,
       boolean autoCleanUp) {
     final var fateId = fateIdGenerator.fromTypeAndKey(type(), fateKey);
     Supplier<FateMutator<T>> mutatorFactory = () -> newMutator(fateId).requireAbsent()

core/src/main/java/org/apache/accumulo/core/fate/zookeeper/MetaFateStore.java

@@ -40,6 +40,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 import java.util.function.UnaryOperator;
@@ -48,6 +49,7 @@
 import org.apache.accumulo.core.clientImpl.AcceptableThriftTableOperationException;
 import org.apache.accumulo.core.fate.AbstractFateStore;
 import org.apache.accumulo.core.fate.Fate;
+import org.apache.accumulo.core.fate.Fate.FateOperation;
 import org.apache.accumulo.core.fate.Fate.TxInfo;
 import org.apache.accumulo.core.fate.FateId;
 import org.apache.accumulo.core.fate.FateInstanceType;
@@ -181,16 +183,26 @@ private Optional<FateTxStore<T>> createAndReserve(FateKey fateKey) {
   }
 
   @Override
-  public Optional<FateId> seedTransaction(Fate.FateOperation fateOp, FateKey fateKey, Repo<T> repo,
-      boolean autoCleanUp) {
-    return createAndReserve(fateKey).map(txStore -> {
-      try {
-        seedTransaction(fateOp, repo, autoCleanUp, txStore);
-        return txStore.getID();
-      } finally {
-        txStore.unreserve(Duration.ZERO);
+  public Seeder<T> beginSeeding() {
+    return new Seeder<T>() {
+      @Override
+      public CompletableFuture<Optional<FateId>> attemptToSeedTransaction(FateOperation fateOp,
+          FateKey fateKey, Repo<T> repo, boolean autoCleanUp) {
+        return CompletableFuture.completedFuture(createAndReserve(fateKey).map(txStore -> {
+          try {
+            seedTransaction(fateOp, repo, autoCleanUp, txStore);
+            return txStore.getID();
+          } finally {
+            txStore.unreserve(Duration.ZERO);
+          }
+        }));
+      }
+
+      @Override
+      public void close() {
+        // Nothing to do for Meta fate store
       }
-    });
+    };
   }
 
   @Override

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

@@ -21,18 +21,22 @@
 import java.io.Serializable;
 import java.util.EnumSet;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Stream;
 
 import org.apache.accumulo.core.fate.Fate;
+import org.apache.accumulo.core.fate.Fate.FateOperation;
 import org.apache.accumulo.core.fate.FateId;
 import org.apache.accumulo.core.fate.FateInstanceType;
 import org.apache.accumulo.core.fate.FateKey;
 import org.apache.accumulo.core.fate.FateStore;
 import org.apache.accumulo.core.fate.FateStore.FateTxStore;
+import org.apache.accumulo.core.fate.FateStore.Seeder;
 import org.apache.accumulo.core.fate.ReadOnlyFateStore;
 import org.apache.accumulo.core.fate.Repo;
 import org.apache.accumulo.core.fate.StackOverflowException;
@@ -150,19 +154,8 @@ public FateId create() {
       }
 
       @Override
-      public Optional<FateId> seedTransaction(Fate.FateOperation fateOp, FateKey fateKey,
-          Repo<T> repo, boolean autoCleanUp) {
-        var optional = store.seedTransaction(fateOp, fateKey, repo, autoCleanUp);
-        if (storeLog.isTraceEnabled()) {
-          optional.ifPresentOrElse(fateId -> {
-            storeLog.trace("{} seeded {} {} {}", fateId, fateKey, toLogString.apply(repo),
-                autoCleanUp);
-          }, () -> {
-            storeLog.trace("Possibly unable to seed {} {} {}", fateKey, toLogString.apply(repo),
-                autoCleanUp);
-          });
-        }
-        return optional;
+      public Seeder<T> beginSeeding() {
+        return new SeederLogger<>(store, toLogString);
       }
 
       @Override
@@ -202,4 +195,41 @@ public void deleteDeadReservations() {
       }
     };
   }
+
+  public static class SeederLogger<T> implements Seeder<T> {
+    private final FateStore<T> store;
+    private final Seeder<T> seeder;
+    private final Function<Repo<T>,String> toLogString;
+
+    public SeederLogger(FateStore<T> store, Function<Repo<T>,String> toLogString) {
+      this.store = Objects.requireNonNull(store);
+      this.seeder = store.beginSeeding();
+      this.toLogString = Objects.requireNonNull(toLogString);
+    }
+
+    @Override
+    public CompletableFuture<Optional<FateId>> attemptToSeedTransaction(FateOperation fateOp,
+        FateKey fateKey, Repo<T> repo, boolean autoCleanUp) {
+      var future = this.seeder.attemptToSeedTransaction(fateOp, fateKey, repo, autoCleanUp);
+      return future.whenComplete((optional, throwable) -> {
+        if (storeLog.isTraceEnabled()) {
+          optional.ifPresentOrElse(fateId -> {
+            storeLog.trace("{} seeded {} {} {}", fateId, fateKey, toLogString.apply(repo),
+                autoCleanUp);
+          }, () -> {
+            storeLog.trace("Possibly unable to seed {} {} {}", fateKey, toLogString.apply(repo),
+                autoCleanUp);
+          });
+        }
+      });
+    }
+
+    @Override
+    public void close() {
+      seeder.close();
+      if (storeLog.isTraceEnabled()) {
+        storeLog.trace("attempted to close seeder for {}", store.type());
+      }
+    }
+  }
 }

core/src/test/java/org/apache/accumulo/core/fate/TestStore.java

@@ -29,10 +29,12 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 import java.util.stream.Stream;
 
+import org.apache.accumulo.core.fate.Fate.FateOperation;
 import org.apache.accumulo.core.util.Pair;
 
 /**
@@ -53,9 +55,17 @@ public FateId create() {
   }
 
   @Override
-  public Optional<FateId> seedTransaction(Fate.FateOperation fateOp, FateKey fateKey,
-      Repo<String> repo, boolean autoCleanUp) {
-    return Optional.empty();
+  public Seeder<String> beginSeeding() {
+    return new Seeder<>() {
+      @Override
+      public CompletableFuture<Optional<FateId>> attemptToSeedTransaction(FateOperation fateOp,
+          FateKey fateKey, Repo<String> repo, boolean autoCleanUp) {
+        return CompletableFuture.completedFuture(Optional.empty());
+      }
+
+      @Override
+      public void close() {}
+    };
   }
 
   @Override

test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_1_IT.java

@@ -32,6 +32,7 @@
 import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.row;
 import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.verify;
 import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.writeData;
+import static org.apache.accumulo.test.fate.FateStoreUtil.seedTransaction;
 import static org.apache.accumulo.test.util.FileMetadataUtil.countFencedFiles;
 import static org.apache.accumulo.test.util.FileMetadataUtil.splitFilesIntoRanges;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -381,7 +382,7 @@ private FateId createCompactionCommitAndDeadMetadata(AccumuloClient c,
     // should never run. Its purpose is to prevent the dead compaction detector
     // from deleting the id.
     Repo<Manager> repo = new FakeRepo();
-    var fateId = fateStore.seedTransaction(Fate.FateOperation.COMMIT_COMPACTION,
+    var fateId = seedTransaction(fateStore, Fate.FateOperation.COMMIT_COMPACTION,
         FateKey.forCompactionCommit(allCids.get(tableId).get(0)), repo, true).orElseThrow();
 
     // Read the tablet metadata