FateIT Repo Error Test (apache#4282)

Author: kevinrr888

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

@@ -36,6 +36,8 @@
 import static org.junit.jupiter.api.Assertions.fail;
 
 import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 
@@ -115,6 +117,59 @@ public Repo<Manager> call(long tid, Manager manager) throws Exception {
 
   }
 
+  public static class TestOperationFails extends ManagerRepo {
+    private static final long serialVersionUID = 1L;
+    private static final Logger LOG = LoggerFactory.getLogger(TestOperationFails.class);
+    private static List<String> undoOrder = new ArrayList<>();
+    private static final int TOTAL_NUM_OPS = 3;
+    private int opNum;
+    private final String opName;
+    private final ExceptionLocation location;
+
+    public TestOperationFails(int opNum, ExceptionLocation location) {
+      this.opNum = opNum;
+      this.opName = "OP" + opNum;
+      this.location = location;
+    }
+
+    @Override
+    public long isReady(long tid, Manager environment) throws Exception {
+      LOG.debug("{} {} Entered isReady()", opName, FateTxId.formatTid(tid));
+      if (location == ExceptionLocation.IS_READY) {
+        if (opNum < TOTAL_NUM_OPS) {
+          return 0;
+        } else {
+          throw new Exception(
+              opName + " " + FateTxId.formatTid(tid) + " isReady() failed - this is expected");
+        }
+      } else {
+        return 0;
+      }
+    }
+
+    @Override
+    public void undo(long tid, Manager environment) throws Exception {
+      LOG.debug("{} {} Entered undo()", opName, FateTxId.formatTid(tid));
+      undoOrder.add(opName);
+      undoLatch.countDown();
+    }
+
+    @Override
+    public Repo<Manager> call(long tid, Manager environment) throws Exception {
+      LOG.debug("{} {} Entered call()", opName, FateTxId.formatTid(tid));
+      if (location == ExceptionLocation.CALL) {
+        if (opNum < TOTAL_NUM_OPS) {
+          return new TestOperationFails(++opNum, location);
+        } else {
+          throw new Exception(
+              opName + " " + FateTxId.formatTid(tid) + " call() failed - this is expected");
+        }
+      } else {
+        return new TestOperationFails(++opNum, location);
+      }
+    }
+  }
+
   private static final Logger LOG = LoggerFactory.getLogger(FateIT.class);
 
   @TempDir
@@ -128,6 +183,11 @@ public Repo<Manager> call(long tid, Manager manager) throws Exception {
 
   private static CountDownLatch callStarted;
   private static CountDownLatch finishCall;
+  private static CountDownLatch undoLatch;
+
+  private enum ExceptionLocation {
+    CALL, IS_READY
+  };
 
   @BeforeAll
   public static void setup() throws Exception {
@@ -165,10 +225,6 @@ public void testTransactionStatus() throws Exception {
       ConfigurationCopy config = new ConfigurationCopy();
       config.set(Property.GENERAL_THREADPOOL_SIZE, "2");
       config.set(Property.MANAGER_FATE_THREADPOOL_SIZE, "1");
-      fate.startTransactionRunners(config);
-
-      // Wait for the transaction runner to be scheduled.
-      UtilWaitThread.sleep(3000);
 
       callStarted = new CountDownLatch(1);
       finishCall = new CountDownLatch(1);
@@ -177,6 +233,11 @@ public void testTransactionStatus() throws Exception {
       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);
+      // Wait for the transaction runner to be scheduled.
+      UtilWaitThread.sleep(3000);
+
       // wait for call() to be called
       callStarted.await();
       assertEquals(IN_PROGRESS, getTxStatus(zk, txid));
@@ -346,10 +407,6 @@ public void testCancelWhileInCall() throws Exception {
       ConfigurationCopy config = new ConfigurationCopy();
       config.set(Property.GENERAL_THREADPOOL_SIZE, "2");
       config.set(Property.MANAGER_FATE_THREADPOOL_SIZE, "1");
-      fate.startTransactionRunners(config);
-
-      // Wait for the transaction runner to be scheduled.
-      UtilWaitThread.sleep(3000);
 
       callStarted = new CountDownLatch(1);
       finishCall = new CountDownLatch(1);
@@ -359,6 +416,11 @@ public void testCancelWhileInCall() throws Exception {
       assertEquals(NEW, getTxStatus(zk, txid));
       fate.seedTransaction("TestOperation", txid, new TestOperation(NS, TID), true, "Test Op");
       assertEquals(SUBMITTED, getTxStatus(zk, txid));
+
+      fate.startTransactionRunners(config);
+      // Wait for the transaction runner to be scheduled.
+      UtilWaitThread.sleep(3000);
+
       // wait for call() to be called
       callStarted.await();
       // cancel the transaction
@@ -369,6 +431,69 @@ public void testCancelWhileInCall() throws Exception {
 
   }
 
+  @Test
+  public void testRepoFails() throws Exception {
+    /*
+     * This test ensures that when an exception occurs in a Repo's call() or isReady() methods, that
+     * undo() will be called back up the chain of Repo's and in the correct order. The test works as
+     * follows: 1) Repo1 is called and returns Repo2, 2) Repo2 is called and returns Repo3, 3) Repo3
+     * is called and throws an exception (in call() or isReady()). It is then expected that: 1)
+     * undo() is called on Repo3, 2) undo() is called on Repo2, 3) undo() is called on Repo1
+     */
+    final ZooStore<Manager> zooStore = new ZooStore<Manager>(ZK_ROOT + Constants.ZFATE, zk);
+    final AgeOffStore<Manager> store =
+        new AgeOffStore<Manager>(zooStore, 3000, System::currentTimeMillis);
+
+    Manager manager = createMock(Manager.class);
+    ServerContext sctx = createMock(ServerContext.class);
+    expect(manager.getContext()).andReturn(sctx).anyTimes();
+    expect(sctx.getZooKeeperRoot()).andReturn(ZK_ROOT).anyTimes();
+    expect(sctx.getZooReaderWriter()).andReturn(zk).anyTimes();
+    replay(manager, sctx);
+
+    Fate<Manager> fate = new Fate<Manager>(manager, store, TraceRepo::toLogString);
+    try {
+      ConfigurationCopy config = new ConfigurationCopy();
+      config.set(Property.GENERAL_THREADPOOL_SIZE, "2");
+      config.set(Property.MANAGER_FATE_THREADPOOL_SIZE, "1");
+      fate.startTransactionRunners(config);
+
+      // Wait for the transaction runner to be scheduled.
+      UtilWaitThread.sleep(3000);
+
+      List<String> expectedUndoOrder = List.of("OP3", "OP2", "OP1");
+      /*
+       * Test exception in call()
+       */
+      undoLatch = new CountDownLatch(TestOperationFails.TOTAL_NUM_OPS);
+      long txid = fate.startTransaction();
+      assertEquals(NEW, getTxStatus(zk, txid));
+      fate.seedTransaction("TestOperationFails", txid,
+          new TestOperationFails(1, ExceptionLocation.CALL), false, "Test Op Fails");
+      // Wait for all the undo() calls to complete
+      undoLatch.await();
+      assertEquals(expectedUndoOrder, TestOperationFails.undoOrder);
+      assertEquals(FAILED, fate.waitForCompletion(txid));
+      assertTrue(fate.getException(txid).getMessage().contains("call() failed"));
+      /*
+       * Test exception in isReady()
+       */
+      TestOperationFails.undoOrder = new ArrayList<>();
+      undoLatch = new CountDownLatch(TestOperationFails.TOTAL_NUM_OPS);
+      txid = fate.startTransaction();
+      assertEquals(NEW, getTxStatus(zk, txid));
+      fate.seedTransaction("TestOperationFails", txid,
+          new TestOperationFails(1, ExceptionLocation.IS_READY), false, "Test Op Fails");
+      // Wait for all the undo() calls to complete
+      undoLatch.await();
+      assertEquals(expectedUndoOrder, TestOperationFails.undoOrder);
+      assertEquals(FAILED, fate.waitForCompletion(txid));
+      assertTrue(fate.getException(txid).getMessage().contains("isReady() failed"));
+    } finally {
+      fate.shutdown();
+    }
+  }
+
   private static void inCall() throws InterruptedException {
     // signal that call started
     callStarted.countDown();