Retry on TXnAbortedException during unlockTableAfterCheckpoint (#3473)

This exception could be caused on txn.delete(activeCheckpointTable) as well.
For this case, we need to retry. In case of TAE due to updating checkpoint status
of the current table, retrying still won't cause an issue since we have the method
protected with compactorCycleCount, managerStatus and the tableStatus.

runtime/src/main/java/org/corfudb/runtime/DistributedCheckpointer.java

@@ -107,6 +107,7 @@ private boolean unlockTableAfterCheckpoint(@NonNull CompactorMetadataTables comp
                 CheckpointingStatus tableStatus = (CheckpointingStatus) txn.getRecord(
                         CompactorMetadataTables.CHECKPOINT_STATUS_TABLE_NAME, tableName).getPayload();
                 if (tableStatus.getStatus() == StatusType.FAILED) {
+                    //Leader marked me as failed
                     log.error("Table status for {}${} has already been marked as FAILED",
                             tableName.getNamespace(), tableName.getTableName());
                     txn.commit();
@@ -120,7 +121,6 @@ private boolean unlockTableAfterCheckpoint(@NonNull CompactorMetadataTables comp
             } catch (TransactionAbortedException e) {
                 log.error("TransactionAbortedException exception while trying to unlock table {}${}: {}",
                         tableName.getNamespace(), tableName.getTableName(), e.getMessage());
-                break; //Leader marked me as failed
             } catch (RuntimeException re) {
                 if (isCriticalRuntimeException(re, retry, MAX_RETRIES)) {
                     throw new IllegalStateException(re);

test/src/test/java/org/corfudb/runtime/DistributedCheckpointerUnitTest.java

@@ -169,6 +169,17 @@ public void unlockTableAfterCheckpointTest() {
         //Fail on different cycleCount values
         assert !distributedCheckpointer.tryCheckpointTable(tableName, t -> cpw);
 
+        when((CheckpointingStatus) corfuStoreEntry.getPayload())
+                .thenReturn(CheckpointingStatus.newBuilder().setStatus(StatusType.STARTED).build())
+                .thenReturn(CheckpointingStatus.newBuilder().setStatus(StatusType.IDLE).build())
+                .thenReturn(CheckpointingStatus.newBuilder().setStatus(StatusType.STARTED).build());
+        when(txn.commit()).thenReturn(Timestamp.getDefaultInstance()) //commit in tryLockTableToCheckpoint()
+                .thenThrow(new TransactionAbortedException(
+                        new TxResolutionInfo(UUID.randomUUID(), new Token(0, 0)),
+                        AbortCause.CONFLICT, new Throwable(), null))
+                .thenReturn(Timestamp.getDefaultInstance());
+        assert distributedCheckpointer.tryCheckpointTable(tableName, t -> cpw);
+
         when((CheckpointingStatus) corfuStoreEntry.getPayload())
                 .thenReturn(CheckpointingStatus.newBuilder().setStatus(StatusType.STARTED).build())
                 .thenReturn(CheckpointingStatus.newBuilder().setStatus(StatusType.IDLE).build())