Close all opened tables on Exception in DistributedCheckpointer (#3543)

The DistributedCheckpointer doesn't close the tables on exception thrown during appendCheckpoint. With a retry in place for runtime exception, rocksdb throws errors when it tries to open the table again. This fix addresses the above issue and closes the opened tables in a finally block.
CorfuDB · Mar 11, 2023 · 864cc68 · 864cc68
1 parent 5e56945
commit 864cc68
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 4 deletions.
diff --git a/runtime/src/main/java/org/corfudb/runtime/DistributedCheckpointer.java b/runtime/src/main/java/org/corfudb/runtime/DistributedCheckpointer.java
@@ -169,17 +169,24 @@ private CheckpointingStatus appendCheckpoint(TableName tableName,
         this.livenessUpdater.updateLiveness(tableName);
         StatusType returnStatus = StatusType.FAILED;
         for (int retry = 0; retry < MAX_RETRIES; retry++) {
+            CheckpointWriter<ICorfuTable<?,?>> cpw = null;
             try {
-                CheckpointWriter<ICorfuTable<?,?>> cpw = checkpointWriterFn.apply(tableName);
+                cpw = checkpointWriterFn.apply(tableName);
                 cpw.appendCheckpoint(Optional.of(livenessUpdater));
-                cpw.getCorfuTable().close();
                 returnStatus = StatusType.COMPLETED;
                 break;
             } catch (RuntimeException re) {
+                log.warn("Encountered RuntimeException, Message: {}, ", re.getMessage(), re);
                 if (isCriticalRuntimeException(re, retry, MAX_RETRIES)) {
                     break; // stop on non-retryable exceptions
                 }
-            }
+            } catch (Exception ex) {
+                log.warn("Encountered unexpected Exception, Message: {}, ", ex.getMessage(), ex);
+            } finally {
+                if (cpw != null) {
+                    cpw.getCorfuTable().close();
+                }
+             }
         }
         this.livenessUpdater.notifyOnSyncComplete();
         return CheckpointingStatus.newBuilder()

diff --git a/test/src/test/java/org/corfudb/runtime/DistributedCheckpointerUnitTest.java b/test/src/test/java/org/corfudb/runtime/DistributedCheckpointerUnitTest.java
@@ -12,6 +12,7 @@
 import org.corfudb.runtime.exceptions.AbortCause;
 import org.corfudb.runtime.exceptions.NetworkException;
 import org.corfudb.runtime.exceptions.TransactionAbortedException;
+import org.corfudb.runtime.exceptions.WriteSizeException;
 import org.corfudb.runtime.exceptions.WrongClusterException;
 import org.junit.Assert;
 import org.junit.Before;
@@ -107,7 +108,8 @@ public void appendCheckpointTest() {
                 .thenReturn(CheckpointingStatus.newBuilder().setStatus(StatusType.IDLE).build());
         //When appendCheckpoint throws an exception
         CheckpointWriter<ICorfuTable<?,?>> cpwThrowException = mock(CheckpointWriter.class);
-        when(cpwThrowException.appendCheckpoint(any(Optional.class))).thenThrow(new IllegalStateException());
+        when(cpwThrowException.appendCheckpoint(any(Optional.class))).thenThrow(new WriteSizeException(2, 1));
+        when(cpwThrowException.getCorfuTable()).thenReturn(mock(CorfuTable.class));
         assert distributedCheckpointer.tryCheckpointTable(tableName, t -> cpwThrowException);
 
         when((CheckpointingStatus) corfuStoreEntry.getPayload())