Adds support for multiple managers running distributed fate (#6168)

Lays the foundation for multiple manager with the following changes.
The best place to start looking at these changes is in the Manager.run()
method which sets everything and ties it all together.

 * Each manager process acquires two zookeeper locks now, a primary lock
   and an assistant lock. Only one manager process can obtain the
   primary lock and when it does it assumes the role of primary manager. 
   All manager processes acquire an assistant lock, which is similar to a
   tserver or compactor lock.  The assistant lock advertises the manager
   process as being available to other Accumulo processes to handle
   assistant manager operations.
 * Manager processes have a single thrift server and thrift services
   hosted on that thrift server are categorized into primary manager and
   assistant manager services. When an assistant manager receives an RPC
   for a primary manager thrift service it will not execute the request and
   will throw an error or ignore the request.
 * The primary manager process delegates manager responsibility via RPCs
   to assistant managers.
 * Any management responsibility not delegated runs on the primary manager.

Using the changes above fate is now distributed across all manager
processes. In the future the changes above should make it easy to
delegate other responsibilities to assistant managers. The following
is an outline of the fate changes.

  * New FateWorker class.  This runs in every manager and handles
    request from the primary manager to adjust what range of the fate
    table its currently responsible for. FateWorker implements a new
    thrift service used to assign it ranges.
  * New FateManager class that is run by the primary manager and is
    responsible for partitioning fate processing across all assistant
    managers. As manager processes come and go this will repartition the
    fate table evenly across all available managers. The FateManager
    communicates with FateWorkers via thrift.
  * Some new RPCs for best effort notifications. Before these changes
    there were in memory notification systems that made the manager more
    responsive.  These would allow a fate operation to signal the Tablet
    Group Watcher to take action sooner.  FateWorkerEnv sends these
    notifications to the primary manger over a new RPC.  Does not matter if
    they are lost, things will still eventually happen.

Other than fate, the primary manager process does everything the
current manager does.  This change pulls from #3262 and #6139.


Co-authored-by: Dave Marion <[email protected]>

Author: keith-turner

core/src/main/java/org/apache/accumulo/core/Constants.java

@@ -49,6 +49,7 @@ public class Constants {
 
   public static final String ZMANAGERS = "/managers";
   public static final String ZMANAGER_LOCK = ZMANAGERS + "/lock";
+  public static final String ZMANAGER_ASSISTANT_LOCK = ZMANAGERS + "/assistants";
   public static final String ZMANAGER_GOAL_STATE = ZMANAGERS + "/goal_state";
   public static final String ZMANAGER_TICK = ZMANAGERS + "/tick";
 

core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java

@@ -1333,7 +1333,7 @@ private static Set<String> createPersistentWatcherPaths() {
         Constants.ZMANAGER_LOCK, Constants.ZMINI_LOCK, Constants.ZMONITOR_LOCK,
         Constants.ZNAMESPACES, Constants.ZRECOVERY, Constants.ZSSERVERS, Constants.ZTABLES,
         Constants.ZTSERVERS, Constants.ZUSERS, RootTable.ZROOT_TABLET, Constants.ZTEST_LOCK,
-        Constants.ZRESOURCEGROUPS)) {
+        Constants.ZMANAGER_ASSISTANT_LOCK, Constants.ZRESOURCEGROUPS)) {
       pathsToWatch.add(path);
     }
     return pathsToWatch;

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

@@ -162,16 +162,21 @@ public FateTxStore<T> reserve(FateId fateId) {
       EnumSet.of(TStatus.SUBMITTED, TStatus.FAILED_IN_PROGRESS);
 
   @Override
-  public void runnable(BooleanSupplier keepWaiting, Consumer<FateIdStatus> idConsumer) {
+  public void runnable(Set<FatePartition> partitions, BooleanSupplier keepWaiting,
+      Consumer<FateIdStatus> idConsumer) {
+
+    if (partitions.isEmpty()) {
+      return;
+    }
 
     AtomicLong seen = new AtomicLong(0);
 
     while (keepWaiting.getAsBoolean() && seen.get() == 0) {
       final long beforeCount = unreservedRunnableCount.getCount();
       final boolean beforeDeferredOverflow = deferredOverflow.get();
 
-      try (Stream<FateIdStatus> inProgress = getTransactions(IN_PROGRESS_SET);
-          Stream<FateIdStatus> other = getTransactions(OTHER_RUNNABLE_SET)) {
+      try (Stream<FateIdStatus> inProgress = getTransactions(partitions, IN_PROGRESS_SET);
+          Stream<FateIdStatus> other = getTransactions(partitions, OTHER_RUNNABLE_SET)) {
         // read the in progress transaction first and then everything else in order to process those
         // first
         var transactions = Stream.concat(inProgress, other);
@@ -200,6 +205,8 @@ public void runnable(BooleanSupplier keepWaiting, Consumer<FateIdStatus> idConsu
         if (beforeCount == unreservedRunnableCount.getCount()) {
           long waitTime = 5000;
           synchronized (deferred) {
+            deferred.keySet().removeIf(
+                fateId -> partitions.stream().noneMatch(partition -> partition.contains(fateId)));
             if (!deferred.isEmpty()) {
               waitTime = deferred.values().stream()
                   .mapToLong(countDownTimer -> countDownTimer.timeLeft(TimeUnit.MILLISECONDS)).min()
@@ -240,9 +247,11 @@ public ReadOnlyFateTxStore<T> read(FateId fateId) {
   }
 
   @Override
-  public Map<FateId,FateReservation> getActiveReservations() {
-    return list().filter(entry -> entry.getFateReservation().isPresent()).collect(Collectors
-        .toMap(FateIdStatus::getFateId, entry -> entry.getFateReservation().orElseThrow()));
+  public Map<FateId,FateReservation> getActiveReservations(Set<FatePartition> partitions) {
+    try (var stream = getTransactions(partitions, EnumSet.allOf(TStatus.class))) {
+      return stream.filter(entry -> entry.getFateReservation().isPresent()).collect(Collectors
+          .toMap(FateIdStatus::getFateId, entry -> entry.getFateReservation().orElseThrow()));
+    }
   }
 
   protected boolean isRunnable(TStatus status) {
@@ -289,6 +298,9 @@ protected void verifyLock(ZooUtil.LockID lockID, FateId fateId) {
 
   protected abstract Stream<FateIdStatus> getTransactions(EnumSet<TStatus> statuses);
 
+  protected abstract Stream<FateIdStatus> getTransactions(Set<FatePartition> partitions,
+      EnumSet<TStatus> statuses);
+
   protected abstract TStatus _getStatus(FateId fateId);
 
   protected abstract Optional<FateKey> getKey(FateId fateId);
@@ -418,7 +430,8 @@ public interface FateIdGenerator {
     FateId newRandomId(FateInstanceType instanceType);
   }
 
-  protected void seededTx() {
+  @Override
+  public void seeded() {
     unreservedRunnableCount.increment();
   }
 

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

@@ -30,6 +30,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.concurrent.ExecutorService;
@@ -51,6 +52,8 @@
 import org.slf4j.LoggerFactory;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Sets;
 import com.google.gson.JsonParser;
 
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
@@ -76,6 +79,7 @@ public class Fate<T> extends FateClient<T> {
   private final AtomicBoolean keepRunning = new AtomicBoolean(true);
   // Visible for FlakyFate test object
   protected final Set<FateExecutor<T>> fateExecutors = new HashSet<>();
+  private Set<FatePartition> currentPartitions = Set.of();
 
   public enum TxInfo {
     FATE_OP, AUTO_CLEAN, EXCEPTION, TX_AGEOFF, RETURN_VALUE
@@ -208,8 +212,10 @@ public void run() {
               fe -> fe.getFateOps().equals(fateOps) && fe.getName().equals(fateExecutorName))) {
             log.debug("[{}] Adding FateExecutor for {} with {} threads", store.type(), fateOps,
                 poolSize);
-            fateExecutors.add(
-                new FateExecutor<>(Fate.this, environment, fateOps, poolSize, fateExecutorName));
+            var fateExecutor =
+                new FateExecutor<>(Fate.this, environment, fateOps, poolSize, fateExecutorName);
+            fateExecutors.add(fateExecutor);
+            fateExecutor.setPartitions(currentPartitions);
           }
         }
       }
@@ -233,7 +239,11 @@ private class DeadReservationCleaner implements Runnable {
     @Override
     public void run() {
       if (keepRunning.get()) {
-        store.deleteDeadReservations();
+        Set<FatePartition> partitions;
+        synchronized (fateExecutors) {
+          partitions = currentPartitions;
+        }
+        store.deleteDeadReservations(partitions);
       }
     }
   }
@@ -369,6 +379,17 @@ public AtomicInteger getNeedMoreThreadsWarnCount() {
     return needMoreThreadsWarnCount;
   }
 
+  public void seeded(Set<FatePartition> partitions) {
+    synchronized (fateExecutors) {
+      if (Sets.intersection(currentPartitions, partitions).isEmpty()) {
+        return;
+      }
+    }
+
+    log.trace("Notified of seeding for {}", partitions);
+    store.seeded();
+  }
+
   /**
    * Initiates shutdown of background threads that run fate operations and cleanup fate data and
    * optionally waits on them. Leaves the fate object in a state where it can still update and read
@@ -432,6 +453,27 @@ public void close() {
     store.close();
   }
 
+  public Set<FatePartition> getPartitions() {
+    synchronized (fateExecutors) {
+      return currentPartitions;
+    }
+  }
+
+  public Set<FatePartition> setPartitions(Set<FatePartition> partitions) {
+    Objects.requireNonNull(partitions);
+    Preconditions.checkArgument(
+        partitions.stream().allMatch(
+            fp -> fp.start().getType() == store.type() && fp.end().getType() == store.type()),
+        "type mismatch type:%s partitions:%s", store.type(), partitions);
+
+    synchronized (fateExecutors) {
+      var old = currentPartitions;
+      currentPartitions = Set.copyOf(partitions);
+      fateExecutors.forEach(fe -> fe.setPartitions(currentPartitions));
+      return old;
+    }
+  }
+
   private boolean anyFateExecutorIsAlive() {
     synchronized (fateExecutors) {
       return fateExecutors.stream().anyMatch(FateExecutor::isAlive);

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

@@ -28,6 +28,9 @@
 import java.time.Duration;
 import java.util.EnumSet;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Stream;
 
@@ -46,6 +49,8 @@ public class FateClient<T> {
   private static final EnumSet<ReadOnlyFateStore.TStatus> FINISHED_STATES =
       EnumSet.of(FAILED, SUCCESSFUL, UNKNOWN);
 
+  private AtomicReference<Consumer<FateId>> seedingConsumer = new AtomicReference<>(fid -> {});
+
   public FateClient(FateStore<T> store, Function<Repo<T>,String> toLogStrFunc) {
     this.store = FateLogger.wrap(store, toLogStrFunc, false);
   }
@@ -56,15 +61,33 @@ public FateId startTransaction() {
   }
 
   public FateStore.Seeder<T> beginSeeding() {
-    return store.beginSeeding();
+    var seeder = store.beginSeeding();
+    return new FateStore.Seeder<T>() {
+      @Override
+      public CompletableFuture<Optional<FateId>> attemptToSeedTransaction(Fate.FateOperation fateOp,
+          FateKey fateKey, Repo<T> repo, boolean autoCleanUp) {
+        var cfuture = seeder.attemptToSeedTransaction(fateOp, fateKey, repo, autoCleanUp);
+        return cfuture.thenApply(optional -> {
+          optional.ifPresent(seedingConsumer.get());
+          return optional;
+        });
+      }
+
+      @Override
+      public void close() {
+        seeder.close();
+      }
+    };
   }
 
   public void seedTransaction(Fate.FateOperation fateOp, FateKey fateKey, Repo<T> repo,
       boolean autoCleanUp) {
+    CompletableFuture<Optional<FateId>> cfuture;
     try (var seeder = store.beginSeeding()) {
-      @SuppressWarnings("unused")
-      var unused = seeder.attemptToSeedTransaction(fateOp, fateKey, repo, autoCleanUp);
+      cfuture = seeder.attemptToSeedTransaction(fateOp, fateKey, repo, autoCleanUp);
     }
+    var optional = cfuture.join();
+    optional.ifPresent(seedingConsumer.get());
   }
 
   // start work in the transaction.. it is safe to call this
@@ -73,6 +96,7 @@ public void seedTransaction(Fate.FateOperation fateOp, FateId fateId, Repo<T> re
       boolean autoCleanUp, String goalMessage) {
     Fate.log.info("[{}] Seeding {} {} {}", store.type(), fateOp, fateId, goalMessage);
     store.seedTransaction(fateOp, fateId, repo, autoCleanUp);
+    seedingConsumer.get().accept(fateId);
   }
 
   // check on the transaction
@@ -176,4 +200,8 @@ public Exception getException(FateId fateId) {
   public Stream<FateKey> list(FateKey.FateKeyType type) {
     return store.list(type);
   }
+
+  public void setSeedingConsumer(Consumer<FateId> seedingConsumer) {
+    this.seedingConsumer.set(seedingConsumer);
+  }
 }

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

@@ -33,6 +33,7 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentLinkedQueue;
@@ -43,6 +44,8 @@
 import java.util.concurrent.TransferQueue;
 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;
@@ -81,6 +84,7 @@ public class FateExecutor<T> {
   private final Set<Fate.FateOperation> fateOps;
   private final ConcurrentLinkedQueue<Integer> idleCountHistory = new ConcurrentLinkedQueue<>();
   private final FateExecutorMetrics<T> fateExecutorMetrics;
+  private final AtomicReference<Set<FatePartition>> partitions = new AtomicReference<>(Set.of());
 
   public FateExecutor(Fate<T> fate, T environment, Set<Fate.FateOperation> fateOps, int poolSize,
       String name) {
@@ -298,6 +302,11 @@ protected ConcurrentLinkedQueue<Integer> getIdleCountHistory() {
     return idleCountHistory;
   }
 
+  public void setPartitions(Set<FatePartition> partitions) {
+    Objects.requireNonNull(partitions);
+    this.partitions.set(Set.copyOf(partitions));
+  }
+
   /**
    * A single thread that finds transactions to work on and queues them up. Do not want each worker
    * thread going to the store and looking for work as it would place more load on the store.
@@ -308,7 +317,12 @@ private class WorkFinder implements Runnable {
     public void run() {
       while (fate.getKeepRunning().get() && !isShutdown()) {
         try {
-          fate.getStore().runnable(() -> fate.getKeepRunning().get(), 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
@@ -319,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/FateExecutorMetrics.java

@@ -22,14 +22,13 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.accumulo.core.metrics.Metric;
-import org.apache.accumulo.core.metrics.MetricsProducer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.micrometer.core.instrument.Gauge;
 import io.micrometer.core.instrument.MeterRegistry;
 
-public class FateExecutorMetrics<T> implements MetricsProducer {
+public class FateExecutorMetrics<T> {
   private static final Logger log = LoggerFactory.getLogger(FateExecutorMetrics.class);
   private final FateInstanceType type;
   private final String poolName;
@@ -49,7 +48,6 @@ protected FateExecutorMetrics(FateInstanceType type, String poolName,
     this.idleWorkerCount = idleWorkerCount;
   }
 
-  @Override
   public void registerMetrics(MeterRegistry registry) {
     // noop if already registered or cleared
     if (state == State.UNREGISTERED) {