WIP

Author: keith-turner

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

@@ -224,7 +224,8 @@ public void run() {
               fe -> fe.getFateOps().equals(fateOps) && fe.getName().equals(fateExecutorName))) {
             log.debug("[{}] Adding FateExecutor for {} with {} threads", store.type(), fateOps,
                 poolSize);
-            var fateExecutor = new FateExecutor<>(Fate.this, environment, fateOps, poolSize, fateExecutorName);
+            var fateExecutor =
+                new FateExecutor<>(Fate.this, environment, fateOps, poolSize, fateExecutorName);
             fateExecutors.add(fateExecutor);
             fateExecutor.setPartitions(currentPartitions);
           }
@@ -575,11 +576,23 @@ public void close() {
     store.close();
   }
 
-  public void setPartitions(Set<FatePartition> partitions) {
+  public Set<FatePartition> getPartitions() {
+    synchronized (fateExecutors) {
+      return currentPartitions;
+    }
+  }
+
+  public boolean setPartitions(Set<FatePartition> expected, Set<FatePartition> partitions) {
+    Objects.requireNonNull(expected);
     Objects.requireNonNull(partitions);
     synchronized (fateExecutors) {
-      currentPartitions = Set.copyOf(partitions);
-      fateExecutors.forEach(fe -> fe.setPartitions(currentPartitions));
+      if (currentPartitions.equals(expected)) {
+        currentPartitions = Set.copyOf(partitions);
+        fateExecutors.forEach(fe -> fe.setPartitions(currentPartitions));
+        return true;
+      } else {
+        return false;
+      }
     }
   }
 

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

@@ -1285,7 +1285,7 @@ protected Fate<FateEnv> initializeFateInstance(ServerContext context, FateStore<
         .scheduleWithFixedDelay(fateCleaner::ageOff, 10, 4 * 60, MINUTES));
 
     if (store.type() == FateInstanceType.META) {
-      fateInstance.setPartitions(Set.of(FatePartition.all(FateInstanceType.META)));
+      fateInstance.setPartitions(Set.of(), Set.of(FatePartition.all(FateInstanceType.META)));
     } // else do not run user transactions for now in the manager... it will have an empty set of
       // partitions
 

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

@@ -81,7 +81,7 @@ public void managerWorkers() throws Exception {
         Set<FatePartition> partitions = entry.getValue();
         var curr = currentAssignments.getOrDefault(worker, Set.of());
         if (!Sets.difference(curr, partitions).isEmpty()) {
-          // This worker has extra partitions that are not desired, unload those
+          // This worker has extra partitions that are not desired
           var intersection = Sets.intersection(curr, partitions);
           if (!setWorkerPartitions(worker, curr, intersection)) {
             log.debug("Failed to set partitions for {} to {}", worker, intersection);

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

@@ -80,8 +80,11 @@ public List<TFatePartition> getPartitions(TInfo tinfo, TCredentials credentials)
           SecurityErrorCode.PERMISSION_DENIED).asThriftException();
     }
 
-    synchronized (currentPartitions) {
-      return currentPartitions.stream().map(FatePartition::toThrift).toList();
+    var localFate = fate;
+    if (localFate == null) {
+      return List.of();
+    } else {
+      return localFate.getPartitions().stream().map(FatePartition::toThrift).toList();
     }
   }
 
@@ -93,21 +96,17 @@ public boolean setPartitions(TInfo tinfo, TCredentials credentials, List<TFatePa
           SecurityErrorCode.PERMISSION_DENIED).asThriftException();
     }
 
-    var expectedSet = expected.stream().map(FatePartition::from).collect(Collectors.toSet());
-    synchronized (currentPartitions) {
-      if (currentPartitions.equals(expectedSet) && fate != null) {
-        expectedSet.forEach(p -> log.info("old partition {}", p));
-        currentPartitions.clear();
-        desired.stream().map(FatePartition::from).forEach(currentPartitions::add);
-        desired.stream().map(FatePartition::from).forEach(p -> log.info("new partition {}", p));
-        log.info("Changed partitions from {} to {}", expectedSet, currentPartitions);
-        fate.setPartitions(Set.copyOf(currentPartitions));
+    var localFate = fate;
+    if (localFate != null) {
+      var expectedSet = expected.stream().map(FatePartition::from).collect(Collectors.toSet());
+      var desiredSet = desired.stream().map(FatePartition::from).collect(Collectors.toSet());
+      if (localFate.setPartitions(expectedSet, desiredSet)) {
+        log.info("Changed partitions from {} to {}", expectedSet, desiredSet);
         return true;
-      } else {
-        log.info("Did not change partitions to {} because {} != {}", desired, expectedSet,
-            currentPartitions);
-        return false;
       }
     }
+
+    log.info("Did not change partitions to {}", desired);
+    return false;
   }
 }