moved fate thread creation out of the constructor

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

@@ -74,9 +74,10 @@ public class Fate<T> {
   private static final Logger log = LoggerFactory.getLogger(Fate.class);
 
   private final FateStore<T> store;
-  private final ScheduledFuture<?> fatePoolsWatcherFuture;
+  private ScheduledFuture<?> fatePoolsWatcherFuture;
   private final AtomicInteger needMoreThreadsWarnCount = new AtomicInteger(0);
-  private final ExecutorService deadResCleanerExecutor;
+  private ExecutorService deadResCleanerExecutor;
+  private final FateConfiguration<T> fateConfig;
 
   private static final EnumSet<TStatus> FINISHED_STATES = EnumSet.of(FAILED, SUCCESSFUL, UNKNOWN);
   public static final Duration INITIAL_DELAY = Duration.ofSeconds(3);
@@ -247,6 +248,10 @@ public void run() {
     }
   }
 
+  private record FateConfiguration<T>(T environment, boolean runDeadResCleaner,
+      AccumuloConfiguration conf, ScheduledThreadPoolExecutor genSchedExecutor) {
+  }
+
   /**
    * Creates a Fault-tolerant executor for the given store type.
    *
@@ -258,6 +263,18 @@ public Fate(T environment, FateStore<T> store, boolean runDeadResCleaner,
       Function<Repo<T>,String> toLogStrFunc, AccumuloConfiguration conf,
       ScheduledThreadPoolExecutor genSchedExecutor) {
     this.store = FateLogger.wrap(store, toLogStrFunc, false);
+    this.fateConfig =
+        new FateConfiguration<>(environment, runDeadResCleaner, conf, genSchedExecutor);
+
+  }
+
+  public void start() {
+    log.info("Start {} FATE", store.type());
+    keepRunning.set(true);
+    final var environment = fateConfig.environment;
+    final var runDeadResCleaner = fateConfig.runDeadResCleaner;
+    final var conf = fateConfig.conf;
+    final var genSchedExecutor = fateConfig.genSchedExecutor;
 
     fatePoolsWatcherFuture =
         genSchedExecutor.scheduleWithFixedDelay(new FatePoolsWatcher(environment, conf),

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

@@ -1267,6 +1267,7 @@ protected Fate<Manager> initializeFateInstance(ServerContext context, FateStore<
 
     final Fate<Manager> fateInstance = new Fate<>(this, store, true, TraceRepo::toLogString,
         getConfiguration(), context.getScheduledExecutor());
+    fateInstance.start();
 
     var fateCleaner = new FateCleaner<>(store, Duration.ofHours(8), this::getSteadyTime);
     ThreadPools.watchCriticalScheduledTask(context.getScheduledExecutor()

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

@@ -40,6 +40,7 @@ public FastFate(T environment, FateStore<T> store, boolean runDeadResCleaner,
       Function<Repo<T>,String> toLogStrFunc, AccumuloConfiguration conf) {
     super(environment, store, runDeadResCleaner, toLogStrFunc, conf,
         new ScheduledThreadPoolExecutor(2));
+    start();
   }
 
   @Override

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

@@ -194,9 +194,12 @@ private void waitFor(FateStore<FeoTestEnv> store, FateId txid) throws Exception
   }
 
   protected Fate<FeoTestEnv> initializeFate(AccumuloClient client, FateStore<FeoTestEnv> store) {
-    return new Fate<>(new FeoTestEnv(client), store, false, r -> r + "",
+    Fate<FeoTestEnv> fate = new Fate<>(new FeoTestEnv(client), store, false, r -> r + "",
         FateTestUtil.updateFateConfig(new ConfigurationCopy(), 1, "AllFateOps"),
         new ScheduledThreadPoolExecutor(2));
+
+    fate.start();
+    return fate;
   }
 
   private static Entry<FateId,String> toIdStep(Entry<Key,Value> e) {

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

@@ -554,9 +554,12 @@ private void submitDeferred(Fate<TestEnv> fate, ServerContext sctx, Set<FateId>
   }
 
   protected Fate<TestEnv> initializeFate(FateStore<TestEnv> store) {
-    return new Fate<>(new TestEnv(), store, false, r -> r + "",
+    Fate<TestEnv> fate = new Fate<>(new TestEnv(), store, false, r -> r + "",
         FateTestUtil.updateFateConfig(new ConfigurationCopy(), 1, "AllFateOps"),
         new ScheduledThreadPoolExecutor(2));
+
+    fate.start();
+    return fate;
   }
 
   protected abstract TStatus getTxStatus(ServerContext sctx, FateId fateId);

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

@@ -917,8 +917,10 @@ protected FastFate<LatchTestEnv> initFateWithDeadResCleaner(FateStore<LatchTestE
   }
 
   protected Fate<LatchTestEnv> initFateNoDeadResCleaner(FateStore<LatchTestEnv> store) {
-    return new Fate<>(new LatchTestEnv(), store, false, Object::toString,
+    Fate<LatchTestEnv> fate = new Fate<>(new LatchTestEnv(), store, false, Object::toString,
         DefaultConfiguration.getInstance(), new ScheduledThreadPoolExecutor(2));
+    fate.start();
+    return fate;
   }
 
   private boolean wordIsTStatus(String word) {

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

@@ -43,6 +43,7 @@ public FlakyFate(T environment, FateStore<T> store, Function<Repo<T>,String> toL
       fateExecutors.add(new FlakyFateExecutor<>(this, environment, poolConfig.getKey(),
           poolConfig.getValue().getValue(), poolConfig.getValue().getKey()));
     }
+    start();
   }
 
   private static class FlakyFateExecutor<T> extends FateExecutor<T> {

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

@@ -254,8 +254,10 @@ private void testMultipleFateInstances(TestStoreFactory<SleepingTestEnv> testSto
 
       Fate<SleepingTestEnv> fate1 = new Fate<>(testEnv1, store1, true, Object::toString,
           DefaultConfiguration.getInstance(), new ScheduledThreadPoolExecutor(2));
+      fate1.start();
       Fate<SleepingTestEnv> fate2 = new Fate<>(testEnv2, store2, false, Object::toString,
           DefaultConfiguration.getInstance(), new ScheduledThreadPoolExecutor(2));
+      fate2.start();
 
       try {
         for (int i = 0; i < numFateIds; i++) {
@@ -361,6 +363,7 @@ private void testDeadReservationsCleanup(TestStoreFactory<LatchTestEnv> testStor
           // fate1.
           fate2 = new Fate<>(testEnv2, store2, false, Object::toString, config,
               new ScheduledThreadPoolExecutor(2));
+          fate2.start();
 
           // Wait for the "dead" reservations to be deleted and picked up again (reserved using
           // fate2/store2/lock2 now).

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

@@ -58,6 +58,7 @@ public SlowFateSplit(T environment, FateStore<T> store, Function<Repo<T>,String>
       fateExecutors.add(new SlowFateSplitExecutor(this, environment, poolConfig.getKey(),
           poolConfig.getValue().getValue(), poolConfig.getValue().getKey()));
     }
+    start();
   }
 
   private class SlowFateSplitExecutor extends FateExecutor<T> {