Several misc Fate changes

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

@@ -310,7 +310,7 @@ protected boolean isReserved() {
     public TStatus waitForStatusChange(EnumSet<TStatus> expected) {
       Preconditions.checkState(!isReserved(),
           "Attempted to wait for status change while reserved: " + fateId);
-      verifyReserved(false);
+      verifyReservedAndNotDeleted(false);
 
       int currNumCallers = concurrentStatusChangeCallers.incrementAndGet();
 
@@ -375,24 +375,22 @@ public void unreserve(Duration deferTime) {
 
     protected abstract void unreserve();
 
-    protected void verifyReserved(boolean isWrite) {
-      if (!isReserved() && isWrite) {
-        throw new IllegalStateException(
-            "Attempted write on unreserved FATE transaction: " + fateId);
-      }
+    protected void verifyReservedAndNotDeleted(boolean isWrite) {
+      Preconditions.checkState(!isWrite || (isReserved() && !deleted),
+          "Attempted write on unreserved or deleted FATE transaction: " + fateId);
     }
 
     @Override
     public TStatus getStatus() {
-      verifyReserved(false);
+      verifyReservedAndNotDeleted(false);
       var status = _getStatus(fateId);
       observedStatus = status;
       return status;
     }
 
     @Override
     public Optional<FateKey> getKey() {
-      verifyReserved(false);
+      verifyReservedAndNotDeleted(false);
       return AbstractFateStore.this.getKey(fateId);
     }
 

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

@@ -383,8 +383,8 @@ public Fate(T environment, FateStore<T> store, boolean runDeadResCleaner,
       // reservations held by dead processes, if they exist.
       deadResCleanerExecutor = ThreadPools.getServerThreadPools().createScheduledExecutorService(1,
           store.type() + "-dead-reservation-cleaner-pool");
-      ScheduledFuture<?> deadReservationCleaner = deadResCleanerExecutor
-          .scheduleWithFixedDelay(new DeadReservationCleaner(), 3, 30, SECONDS);
+      ScheduledFuture<?> deadReservationCleaner = deadResCleanerExecutor.scheduleWithFixedDelay(
+          new DeadReservationCleaner(), 3, getDeadResCleanupDelay().toSeconds(), SECONDS);
       ThreadPools.watchCriticalScheduledTask(deadReservationCleaner);
     }
     this.deadResCleanerExecutor = deadResCleanerExecutor;
@@ -393,6 +393,10 @@ public Fate(T environment, FateStore<T> store, boolean runDeadResCleaner,
     this.workFinder.start();
   }
 
+  public Duration getDeadResCleanupDelay() {
+    return Duration.ofMinutes(3);
+  }
+
   // get a transaction id back to the requester before doing any work
   public FateId startTransaction() {
     return store.create();

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

@@ -28,6 +28,8 @@
 
 import org.apache.accumulo.core.dataImpl.KeyExtent;
 import org.apache.accumulo.core.metadata.schema.ExternalCompactionId;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
 import org.apache.hadoop.io.DataInputBuffer;
 
 public class FateKey {
@@ -168,4 +170,13 @@ private static Optional<ExternalCompactionId> deserializeCompactionId(FateKeyTyp
         throw new IllegalStateException("Unexpected FateInstanceType found " + type);
     }
   }
+
+  @Override
+  public String toString() {
+    var buf = new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE);
+    buf.append("FateKeyType", type);
+    keyExtent.ifPresentOrElse(keyExtent -> buf.append("KeyExtent", keyExtent),
+        () -> buf.append("ExternalCompactionID", compactionId.orElseThrow()));
+    return buf.toString();
+  }
 }

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

@@ -420,7 +420,7 @@ private FateTxStoreImpl(FateId fateId, FateReservation reservation) {
 
     @Override
     public Repo<T> top() {
-      verifyReserved(false);
+      verifyReservedAndNotDeleted(false);
 
       return scanTx(scanner -> {
         scanner.setRange(getRow(fateId));
@@ -436,7 +436,7 @@ public Repo<T> top() {
 
     @Override
     public List<ReadOnlyRepo<T>> getStack() {
-      verifyReserved(false);
+      verifyReservedAndNotDeleted(false);
 
       return scanTx(scanner -> {
         scanner.setRange(getRow(fateId));
@@ -451,7 +451,7 @@ public List<ReadOnlyRepo<T>> getStack() {
 
     @Override
     public Serializable getTransactionInfo(TxInfo txInfo) {
-      verifyReserved(false);
+      verifyReservedAndNotDeleted(false);
 
       try (Scanner scanner = context.createScanner(tableName, Authorizations.EMPTY)) {
         scanner.setRange(getRow(fateId));
@@ -487,7 +487,7 @@ public Serializable getTransactionInfo(TxInfo txInfo) {
 
     @Override
     public long timeCreated() {
-      verifyReserved(false);
+      verifyReservedAndNotDeleted(false);
 
       return scanTx(scanner -> {
         scanner.setRange(getRow(fateId));
@@ -499,7 +499,7 @@ public long timeCreated() {
 
     @Override
     public void push(Repo<T> repo) throws StackOverflowException {
-      verifyReserved(true);
+      verifyReservedAndNotDeleted(true);
 
       Optional<Integer> top = findTop();
 
@@ -514,7 +514,7 @@ public void push(Repo<T> repo) throws StackOverflowException {
 
     @Override
     public void pop() {
-      verifyReserved(true);
+      verifyReservedAndNotDeleted(true);
 
       Optional<Integer> top = findTop();
       top.ifPresent(t -> newMutator(fateId)
@@ -523,15 +523,15 @@ public void pop() {
 
     @Override
     public void setStatus(TStatus status) {
-      verifyReserved(true);
+      verifyReservedAndNotDeleted(true);
 
       newMutator(fateId).putStatus(status).mutate();
       observedStatus = status;
     }
 
     @Override
     public void setTransactionInfo(TxInfo txInfo, Serializable so) {
-      verifyReserved(true);
+      verifyReservedAndNotDeleted(true);
 
       final byte[] serialized = serializeTxInfo(so);
 
@@ -540,7 +540,7 @@ public void setTransactionInfo(TxInfo txInfo, Serializable so) {
 
     @Override
     public void delete() {
-      verifyReserved(true);
+      verifyReservedAndNotDeleted(true);
 
       var mutator = newMutator(fateId);
       mutator.requireStatus(TStatus.NEW, TStatus.SUBMITTED, TStatus.SUCCESSFUL, TStatus.FAILED);

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

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.accumulo.core.fate;
+package org.apache.accumulo.core.fate.zookeeper;
 
 import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
@@ -39,9 +39,15 @@
 import java.util.stream.Stream;
 
 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.TxInfo;
-import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
-import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
+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.ReadOnlyRepo;
+import org.apache.accumulo.core.fate.Repo;
+import org.apache.accumulo.core.fate.StackOverflowException;
 import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeExistsPolicy;
 import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeMissingPolicy;
 import org.apache.hadoop.io.DataInputBuffer;
@@ -239,7 +245,7 @@ private FateTxStoreImpl(FateId fateId, FateReservation reservation) {
 
     @Override
     public Repo<T> top() {
-      verifyReserved(false);
+      verifyReservedAndNotDeleted(false);
 
       for (int i = 0; i < RETRIES; i++) {
         String txpath = getTXPath(fateId);
@@ -291,7 +297,7 @@ private String findTop(String txpath) throws KeeperException, InterruptedExcepti
 
     @Override
     public void push(Repo<T> repo) throws StackOverflowException {
-      verifyReserved(true);
+      verifyReservedAndNotDeleted(true);
 
       String txpath = getTXPath(fateId);
       try {
@@ -310,7 +316,7 @@ public void push(Repo<T> repo) throws StackOverflowException {
 
     @Override
     public void pop() {
-      verifyReserved(true);
+      verifyReservedAndNotDeleted(true);
 
       try {
         String txpath = getTXPath(fateId);
@@ -326,7 +332,7 @@ public void pop() {
 
     @Override
     public void setStatus(TStatus status) {
-      verifyReserved(true);
+      verifyReservedAndNotDeleted(true);
 
       try {
         zk.mutateExisting(getTXPath(fateId), currSerializedData -> {
@@ -353,7 +359,7 @@ public void setStatus(TStatus status) {
 
     @Override
     public void delete() {
-      verifyReserved(true);
+      verifyReservedAndNotDeleted(true);
 
       try {
         zk.recursiveDelete(getTXPath(fateId), NodeMissingPolicy.SKIP);
@@ -365,7 +371,7 @@ public void delete() {
 
     @Override
     public void setTransactionInfo(Fate.TxInfo txInfo, Serializable so) {
-      verifyReserved(true);
+      verifyReservedAndNotDeleted(true);
 
       try {
         zk.putPersistentData(getTXPath(fateId) + "/" + txInfo, serializeTxInfo(so),
@@ -377,14 +383,14 @@ public void setTransactionInfo(Fate.TxInfo txInfo, Serializable so) {
 
     @Override
     public Serializable getTransactionInfo(Fate.TxInfo txInfo) {
-      verifyReserved(false);
+      verifyReservedAndNotDeleted(false);
 
       return MetaFateStore.this.getTransactionInfo(txInfo, fateId);
     }
 
     @Override
     public long timeCreated() {
-      verifyReserved(false);
+      verifyReservedAndNotDeleted(false);
 
       try {
         Stat stat = zk.getZooKeeper().exists(getTXPath(fateId), false);
@@ -396,7 +402,7 @@ public long timeCreated() {
 
     @Override
     public List<ReadOnlyRepo<T>> getStack() {
-      verifyReserved(false);
+      verifyReservedAndNotDeleted(false);
       String txpath = getTXPath(fateId);
 
       outer: while (true) {

server/base/src/main/java/org/apache/accumulo/server/util/Admin.java

@@ -71,9 +71,9 @@
 import org.apache.accumulo.core.fate.FateId;
 import org.apache.accumulo.core.fate.FateInstanceType;
 import org.apache.accumulo.core.fate.FateStore;
-import org.apache.accumulo.core.fate.MetaFateStore;
 import org.apache.accumulo.core.fate.ReadOnlyFateStore;
 import org.apache.accumulo.core.fate.user.UserFateStore;
+import org.apache.accumulo.core.fate.zookeeper.MetaFateStore;
 import org.apache.accumulo.core.fate.zookeeper.ZooCache;
 import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
 import org.apache.accumulo.core.lock.ServiceLock;

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

@@ -76,8 +76,8 @@
 import org.apache.accumulo.core.fate.FateId;
 import org.apache.accumulo.core.fate.FateInstanceType;
 import org.apache.accumulo.core.fate.FateStore;
-import org.apache.accumulo.core.fate.MetaFateStore;
 import org.apache.accumulo.core.fate.user.UserFateStore;
+import org.apache.accumulo.core.fate.zookeeper.MetaFateStore;
 import org.apache.accumulo.core.fate.zookeeper.ZooCache.ZcStat;
 import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
 import org.apache.accumulo.core.fate.zookeeper.ZooUtil;

server/manager/src/main/java/org/apache/accumulo/manager/metrics/fate/meta/MetaFateMetrics.java

@@ -26,8 +26,8 @@
 
 import org.apache.accumulo.core.Constants;
 import org.apache.accumulo.core.fate.AbstractFateStore;
-import org.apache.accumulo.core.fate.MetaFateStore;
 import org.apache.accumulo.core.fate.ReadOnlyFateStore;
+import org.apache.accumulo.core.fate.zookeeper.MetaFateStore;
 import org.apache.accumulo.manager.metrics.fate.FateMetrics;
 import org.apache.accumulo.server.ServerContext;
 import org.apache.zookeeper.KeeperException;

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

@@ -78,8 +78,8 @@
 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.MetaFateStore;
 import org.apache.accumulo.core.fate.user.UserFateStore;
+import org.apache.accumulo.core.fate.zookeeper.MetaFateStore;
 import org.apache.accumulo.core.iterators.DevNull;
 import org.apache.accumulo.core.iterators.Filter;
 import org.apache.accumulo.core.iterators.IteratorEnvironment;

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

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.test.fate;
+
+import java.time.Duration;
+import java.util.function.Function;
+
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.fate.Fate;
+import org.apache.accumulo.core.fate.FateStore;
+import org.apache.accumulo.core.fate.Repo;
+
+/**
+ * A FATE which performs the dead reservation cleanup with a much shorter delay between
+ */
+public class FastFate<T> extends Fate<T> {
+
+  public FastFate(T environment, FateStore<T> store, boolean runDeadResCleaner,
+      Function<Repo<T>,String> toLogStrFunc, AccumuloConfiguration conf) {
+    super(environment, store, runDeadResCleaner, toLogStrFunc, conf);
+  }
+
+  @Override
+  public Duration getDeadResCleanupDelay() {
+    return Duration.ofSeconds(15);
+  }
+}