WIP

Author: keith-turner

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

@@ -40,6 +40,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
@@ -161,7 +162,7 @@ public FateTxStore<T> reserve(FateId fateId) {
       EnumSet.of(TStatus.SUBMITTED, TStatus.FAILED_IN_PROGRESS);
 
   @Override
-  public void runnable(Set<FatePartition> partitions, AtomicBoolean keepWaiting,
+  public void runnable(Set<FatePartition> partitions, BooleanSupplier keepWaiting,
       Consumer<FateIdStatus> idConsumer) {
 
     if (partitions.isEmpty()) {
@@ -170,7 +171,7 @@ public void runnable(Set<FatePartition> partitions, AtomicBoolean keepWaiting,
 
     AtomicLong seen = new AtomicLong(0);
 
-    while (keepWaiting.get() && seen.get() == 0) {
+    while (keepWaiting.getAsBoolean() && seen.get() == 0) {
       final long beforeCount = unreservedRunnableCount.getCount();
       final boolean beforeDeferredOverflow = deferredOverflow.get();
 
@@ -212,8 +213,7 @@ public void runnable(Set<FatePartition> partitions, AtomicBoolean keepWaiting,
           }
 
           if (waitTime > 0) {
-            unreservedRunnableCount.waitFor(count -> count != beforeCount, waitTime,
-                keepWaiting::get);
+            unreservedRunnableCount.waitFor(count -> count != beforeCount, waitTime, keepWaiting);
           }
         }
       }

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

@@ -45,6 +45,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BooleanSupplier;
 
 import org.apache.accumulo.core.clientImpl.AcceptableThriftTableOperationException;
 import org.apache.accumulo.core.conf.Property;
@@ -316,8 +317,12 @@ private class WorkFinder implements Runnable {
     public void run() {
       while (fate.getKeepRunning().get() && !isShutdown()) {
         try {
-          // TODO
-          fate.getStore().runnable(partitions.get(), fate.getKeepRunning(), fateIdStatus -> {
+          var localPartitions = partitions.get();
+          // if the set of partitions changes, we should stop looking for work w/ the old set of
+          // partitions
+          BooleanSupplier keepRunning =
+              () -> fate.getKeepRunning().get() && localPartitions == partitions.get();
+          fate.getStore().runnable(localPartitions, keepRunning, fateIdStatus -> {
             // The FateId with the fate operation 'fateOp' is workable by this FateExecutor if
             // 1) This FateExecutor is assigned to work on 'fateOp' ('fateOp' is in 'fateOps')
             // 2) The transaction was cancelled while NEW. This is an edge case that needs to be
@@ -328,7 +333,7 @@ public void run() {
             var fateOp = fateIdStatus.getFateOperation().orElse(null);
             if ((fateOp != null && fateOps.contains(fateOp))
                 || txCancelledWhileNew(status, fateOp)) {
-              while (fate.getKeepRunning().get() && !isShutdown()) {
+              while (keepRunning.getAsBoolean() && !isShutdown()) {
                 try {
                   // The reason for calling transfer instead of queueing is avoid rescanning the
                   // storage layer and adding the same thing over and over. For example if all

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

@@ -24,7 +24,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.stream.Stream;
 
@@ -164,7 +164,7 @@ interface FateIdStatus {
    * is found or until the keepWaiting parameter is false. It will return once all runnable ids
    * found were passed to the consumer.
    */
-  void runnable(Set<FatePartition> partitions, AtomicBoolean keepWaiting,
+  void runnable(Set<FatePartition> partitions, BooleanSupplier keepWaiting,
       Consumer<FateIdStatus> idConsumer);
 
   /**

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

@@ -25,7 +25,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Stream;
@@ -142,7 +142,7 @@ public Stream<FateKey> list(FateKey.FateKeyType type) {
       }
 
       @Override
-      public void runnable(Set<FatePartition> partitions, AtomicBoolean keepWaiting,
+      public void runnable(Set<FatePartition> partitions, BooleanSupplier keepWaiting,
           Consumer<FateIdStatus> idConsumer) {
         store.runnable(partitions, keepWaiting, idConsumer);
       }

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

@@ -30,7 +30,7 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.stream.Stream;
 
@@ -273,7 +273,7 @@ public Stream<FateKey> list(FateKey.FateKeyType type) {
   }
 
   @Override
-  public void runnable(Set<FatePartition> partitions, AtomicBoolean keepWaiting,
+  public void runnable(Set<FatePartition> partitions, BooleanSupplier keepWaiting,
       Consumer<FateIdStatus> idConsumer) {
     throw new UnsupportedOperationException();
   }

server/manager/src/main/java/org/apache/accumulo/manager/fate/FateManager.java

@@ -47,7 +47,8 @@
 
 /**
  * Partitions fate across manager assistant processes. This is done by assigning ranges of the fate
- * uuid key space to different processes.
+ * uuid key space to different processes. The partitions are logical and do not correspond to the
+ * physical partitioning of the fate table.
  */
 public class FateManager {
 
@@ -163,6 +164,10 @@ private Map<HostAndPort,Set<FatePartition>> computeDesiredAssignments(
     return desiredAssignments;
   }
 
+  /**
+   * Computes a single partition for each worker such that the partition cover all possible UUIDs
+   * and evenly divide the UUIDs.
+   */
   private Set<FatePartition> getDesiredPartitions(int numWorkers) {
     Preconditions.checkArgument(numWorkers >= 0);
 
@@ -184,9 +189,9 @@ private Set<FatePartition> getDesiredPartitions(int numWorkers) {
           FateId.from(FateInstanceType.USER, endUuid)));
     }
 
-    // last one is
     long start = ((numWorkers - 1) * jump) << 4;
     UUID startUuid = new UUID(start, 0);
+    // last partition has a special end uuid that is all f nibbles.
     UUID endUuid = new UUID(-1, -1);
     desired.add(new FatePartition(FateId.from(FateInstanceType.USER, startUuid),
         FateId.from(FateInstanceType.USER, endUuid)));

test/src/main/java/org/apache/accumulo/test/MultipleManagerIT.java

@@ -51,19 +51,22 @@ protected void configure(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSit
   public void test() throws Exception {
 
     List<Process> managerWorkers = new ArrayList<>();
+    // start two fate workers initially
     for (int i = 0; i < 2; i++) {
       managerWorkers.add(exec(ManagerWorker.class));
     }
 
     var executor = Executors.newCachedThreadPool();
 
+    // This assigns fate partitions to fate worker processes, run it in a background thread.
     var fateMgr = new FateManager(getServerContext());
     var future = executor.submit(() -> {
       fateMgr.managerWorkers();
       return null;
     });
 
     Thread.sleep(30_000);
+    // start more fate workers, should see the partitions be shuffled eventually
     for (int i = 0; i < 3; i++) {
       managerWorkers.add(exec(ManagerWorker.class));
     }
@@ -76,9 +79,11 @@ public void test() throws Exception {
         var table = "t" + i;
         // TODO seeing in the logs that fate operations for the same table are running on different
         // processes, however there is a 5 second delay because there is no notification mechanism
+        // currently.
 
         // TODO its hard to find everything related to a table id in the logs, especially when the
-        // table id is like "b"
+        // table id is like "b". Was tring to follow a single table across multiple manager workers
+        // processes.
         var tableOpsFuture = executor.submit(() -> {
           client.tableOperations().create(table);
           log.info("Created table {}", table);

test/src/main/java/org/apache/accumulo/test/fate/FateStoreITBase.java

@@ -201,7 +201,7 @@ protected void testDeferredOverflow(FateStore<TestEnv> store, ServerContext sctx
       // Run and verify all 10 transactions still exist and were not
       // run because of the deferral time of all the transactions
       future = executor.submit(() -> store.runnable(Set.of(FatePartition.all(store.type())),
-          keepRunning, fateIdStatus -> transactions.remove(fateIdStatus.getFateId())));
+          keepRunning::get, fateIdStatus -> transactions.remove(fateIdStatus.getFateId())));
       Thread.sleep(2000);
       assertEquals(10, transactions.size());
       // Setting this flag to false should terminate the task if sleeping
@@ -227,7 +227,7 @@ protected void testDeferredOverflow(FateStore<TestEnv> store, ServerContext sctx
       // and removed from the store
       keepRunning.set(true);
       future = executor.submit(() -> store.runnable(Set.of(FatePartition.all(store.type())),
-          keepRunning, fateIdStatus -> transactions.remove(fateIdStatus.getFateId())));
+          keepRunning::get, fateIdStatus -> transactions.remove(fateIdStatus.getFateId())));
       Wait.waitFor(transactions::isEmpty);
       // Setting this flag to false should terminate the task if sleeping
       keepRunning.set(false);