Modified Fate TransactionRunner to exit normally when shut down (#6021)

Modified the Fate TransactionRunner threads to allow them to
exit their run loop normally when an InterruptedException is
raised due to Fate being shut down.

Closes #6014

Author: dlmarion

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

@@ -74,6 +74,24 @@ public enum TxInfo {
 
   private class TransactionRunner implements Runnable {
 
+    private boolean isInterruptedException(Throwable e) {
+      if (e == null) {
+        return false;
+      }
+
+      if (e instanceof InterruptedException) {
+        return true;
+      }
+
+      for (Throwable suppressed : e.getSuppressed()) {
+        if (isInterruptedException(suppressed)) {
+          return true;
+        }
+      }
+
+      return isInterruptedException(e.getCause());
+    }
+
     @Override
     public void run() {
       while (keepRunning.get()) {
@@ -99,14 +117,31 @@ public void run() {
                   store.setStatus(tid, IN_PROGRESS);
                 }
                 op = executeCall(tid, op);
+                // It's possible that a Fate operation impl
+                // may not do the right thing with an
+                // InterruptedException.
+                if (Thread.currentThread().isInterrupted()) {
+                  throw new InterruptedException("Fate Transaction Runner thread interrupted");
+                }
               } else {
                 continue;
               }
-
             } catch (Exception e) {
-              blockIfHadoopShutdown(tid, e);
-              transitionToFailed(tid, e);
-              continue;
+              if (!isInterruptedException(e)) {
+                blockIfHadoopShutdown(tid, e);
+                transitionToFailed(tid, e);
+                continue;
+              } else {
+                if (keepRunning.get()) {
+                  throw e;
+                } else {
+                  // If we are shutting down then Fate.shutdown was called
+                  // and ExecutorService.shutdownNow was called resulting
+                  // in this exception. We will exit at the top of the loop.
+                  Thread.interrupted();
+                  continue;
+                }
+              }
             }
 
             if (op == null) {
@@ -130,7 +165,19 @@ public void run() {
             }
           }
         } catch (Exception e) {
-          runnerLog.error("Uncaught exception in FATE runner thread.", e);
+          if (isInterruptedException(e)) {
+            if (keepRunning.get()) {
+              runnerLog.error("Uncaught InterruptedException in FATE runner thread.", e);
+            } else {
+              // If we are shutting down then Fate.shutdown was called
+              // and ExecutorService.shutdownNow was called resulting
+              // in this exception. We will exit at the top of the loop,
+              // so continue this loop iteration normally.
+              Thread.interrupted();
+            }
+          } else {
+            runnerLog.error("Uncaught exception in FATE runner thread.", e);
+          }
         } finally {
           if (tid != null) {
             store.unreserve(tid, deferTime, TimeUnit.MILLISECONDS);
@@ -198,6 +245,7 @@ private void processFailed(long tid, Repo<T> op) {
     }
 
     private void doCleanUp(long tid) {
+      log.debug("Cleaning up {}", tid);
       Boolean autoClean = (Boolean) store.getTransactionInfo(tid, TxInfo.AUTO_CLEAN);
       if (autoClean != null && autoClean) {
         store.delete(tid);
@@ -261,6 +309,7 @@ public void startTransactionRunners(AccumuloConfiguration conf,
       ScheduledThreadPoolExecutor serverGeneralScheduledThreadPool) {
     final ThreadPoolExecutor pool = ThreadPools.getServerThreadPools().createExecutorService(conf,
         Property.MANAGER_FATE_THREADPOOL_SIZE, true);
+    log.debug("Starting Fate Transaction Runner pool with {} threads", pool.getCorePoolSize());
     ThreadPools
         .watchCriticalScheduledTask(serverGeneralScheduledThreadPool.scheduleWithFixedDelay(() -> {
           // resize the pool if the property changed
@@ -421,6 +470,9 @@ public Exception getException(long tid) {
    * Flags that FATE threadpool to clear out and end. Does not actively stop running FATE processes.
    */
   public void shutdown(boolean wait) {
+    log.info("Shutdown called on Fate, waiting: {}", wait);
+    // important this is set before shutdownNow is called as the background
+    // threads will check this to see if shutdown related errors should be ignored.
     keepRunning.set(false);
     if (executor == null) {
       return;

test/src/main/java/org/apache/accumulo/test/fate/zookeeper/FateIT.java

@@ -31,6 +31,7 @@
 import static org.easymock.EasyMock.replay;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
@@ -42,6 +43,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.accumulo.core.Constants;
 import org.apache.accumulo.core.clientImpl.thrift.TableOperation;
@@ -190,6 +192,7 @@ public Repo<Manager> call(long tid, Manager environment) throws Exception {
   private static CountDownLatch callStarted;
   private static CountDownLatch finishCall;
   private static CountDownLatch undoLatch;
+  private static AtomicReference<Throwable> interruptedException = new AtomicReference<>();
 
   Fate<Manager> fate;
 
@@ -211,6 +214,7 @@ public static void setup() throws Exception {
 
   @BeforeEach
   public void beforeEach() throws Exception {
+    interruptedException.set(null);
     final ZooStore<Manager> zooStore = new ZooStore<>(ZK_ROOT + Constants.ZFATE, zk, zc);
     final AgeOffStore<Manager> store = new AgeOffStore<>(zooStore, 3000, System::currentTimeMillis);
 
@@ -448,11 +452,73 @@ public void testRepoFails() throws Exception {
     assertTrue(fate.getException(txid).getMessage().contains("isReady() failed"));
   }
 
+  @Test
+  public void testShutdownDoesNotFailTx() throws Exception {
+    ConfigurationCopy config = new ConfigurationCopy();
+    config.set(Property.GENERAL_THREADPOOL_SIZE, "2");
+    config.set(Property.MANAGER_FATE_THREADPOOL_SIZE, "1");
+
+    // Wait for the transaction runner to be scheduled.
+    UtilWaitThread.sleep(3000);
+
+    callStarted = new CountDownLatch(1);
+    finishCall = new CountDownLatch(1);
+
+    long txid = fate.startTransaction();
+    assertEquals(TStatus.NEW, getTxStatus(zk, txid));
+    fate.seedTransaction("TestOperation", txid, new TestOperation(NS, TID), true, "Test Op");
+    assertEquals(TStatus.SUBMITTED, getTxStatus(zk, txid));
+
+    fate.startTransactionRunners(config, new ScheduledThreadPoolExecutor(2));
+
+    // wait for call() to be called
+    callStarted.await();
+    assertEquals(IN_PROGRESS, getTxStatus(zk, txid));
+
+    // shutdown fate
+    fate.shutdown(true);
+
+    // tell the op to exit the method
+    Wait.waitFor(() -> interruptedException.get() != null);
+    interruptedException.set(null);
+
+    // restart fate
+    beforeEach();
+    assertEquals(IN_PROGRESS, getTxStatus(zk, txid));
+
+    // Restarting the transaction runners will retry the in-progress
+    // transaction. Reset the CountDownLatch's to confirm.
+    callStarted = new CountDownLatch(1);
+    finishCall = new CountDownLatch(1);
+    fate.startTransactionRunners(config, new ScheduledThreadPoolExecutor(2));
+    callStarted.await();
+    assertEquals(IN_PROGRESS, getTxStatus(zk, txid));
+    finishCall.countDown();
+
+    // This should complete normally, cleaning up the tx and deleting it from ZK
+    while (true) {
+      try {
+        getTxStatus(zk, txid);
+        Thread.sleep(100);
+        continue;
+      } catch (KeeperException.NoNodeException e) {
+        break;
+      }
+    }
+    assertNull(interruptedException.get());
+  }
+
   private static void inCall() throws InterruptedException {
     // signal that call started
     callStarted.countDown();
-    // wait for the signal to exit the method
-    finishCall.await();
+    try {
+      // wait for the signal to exit the method
+      finishCall.await();
+    } catch (InterruptedException e) {
+      LOG.debug("InterruptedException occurred inCall.");
+      interruptedException.set(e);
+      throw e;
+    }
   }
 
   /*