diff --git a/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/rocksdb/checkpoint/RocksCheckpointer.java b/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/rocksdb/checkpoint/RocksCheckpointer.java
index 3764847..b26a2f0 100644
--- a/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/rocksdb/checkpoint/RocksCheckpointer.java
+++ b/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/rocksdb/checkpoint/RocksCheckpointer.java
@@ -20,6 +20,7 @@
 import com.google.common.io.MoreFiles;
 import com.google.common.io.RecursiveDeleteOption;
 import java.io.File;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.file.Files;
@@ -123,6 +124,8 @@
                     latestCheckpointId = checkpointId;
                     latestCheckpoint = ckpt;
                 }
+            } catch (FileNotFoundException fnfe) {
+                log.error("Metadata is corrupt for the checkpoint {}. Skipping it.", checkpointId);
             }
         }
         return Pair.of(latestCheckpointId, latestCheckpoint);
diff --git a/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/rocksdb/checkpoint/RocksdbCheckpointTask.java b/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/rocksdb/checkpoint/RocksdbCheckpointTask.java
index efdc4a4..3500ec5 100644
--- a/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/rocksdb/checkpoint/RocksdbCheckpointTask.java
+++ b/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/rocksdb/checkpoint/RocksdbCheckpointTask.java
@@ -30,6 +30,7 @@
 import java.util.List;
 import java.util.Set;
 import java.util.UUID;
+
 import lombok.extern.slf4j.Slf4j;
 import org.apache.bookkeeper.statelib.api.checkpoint.CheckpointStore;
 import org.apache.bookkeeper.statelib.api.exceptions.StateStoreException;
@@ -44,6 +45,15 @@
 @Slf4j
 public class RocksdbCheckpointTask {
 
+    /**
+     * Error injection support for testing of the checkpoint.
+     * @param <T>
+     */
+    @FunctionalInterface
+    public interface InjectedError<T> {
+        void accept(T t) throws IOException;
+    }
+
     private final String dbName;
     private final Checkpoint checkpoint;
     private final File checkpointDir;
@@ -52,6 +62,9 @@
     private final boolean removeLocalCheckpointAfterSuccessfulCheckpoint;
     private final boolean removeRemoteCheckpointsAfterSuccessfulCheckpoint;
 
+    // for testing only
+    private InjectedError<String> injectedError = (String checkpointId) -> {};
+
     public RocksdbCheckpointTask(String dbName,
                                  Checkpoint checkpoint,
                                  File checkpointDir,
@@ -67,6 +80,10 @@
         this.removeRemoteCheckpointsAfterSuccessfulCheckpoint = removeRemoteCheckpoints;
     }
 
+    public void setInjectedError(InjectedError<String> injectedError) {
+        this.injectedError = injectedError;
+    }
+
     public String checkpoint(byte[] txid) throws StateStoreException {
         String checkpointId = UUID.randomUUID().toString();
 
@@ -89,6 +106,8 @@
                 checkpointStore.createDirectories(sstsPath);
             }
 
+            injectedError.accept(checkpointId);
+
             // get the files to copy
             List<File> filesToCopy = getFilesToCopy(tempDir);
 
diff --git a/stream/statelib/src/test/java/org/apache/bookkeeper/statelib/impl/rocksdb/checkpoint/RocksCheckpointerTest.java b/stream/statelib/src/test/java/org/apache/bookkeeper/statelib/impl/rocksdb/checkpoint/RocksCheckpointerTest.java
index e493579..40e3517 100644
--- a/stream/statelib/src/test/java/org/apache/bookkeeper/statelib/impl/rocksdb/checkpoint/RocksCheckpointerTest.java
+++ b/stream/statelib/src/test/java/org/apache/bookkeeper/statelib/impl/rocksdb/checkpoint/RocksCheckpointerTest.java
@@ -24,11 +24,13 @@
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 import com.google.common.collect.Sets;
 import com.google.common.io.ByteStreams;
 import com.google.common.io.MoreFiles;
 import com.google.common.io.RecursiveDeleteOption;
+
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.InputStream;
@@ -36,6 +38,8 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+
 import lombok.Cleanup;
 import org.apache.bookkeeper.common.coder.StringUtf8Coder;
 import org.apache.bookkeeper.common.kv.KV;
@@ -413,4 +417,74 @@
         verifyNumKvs(100);
     }
 
+    /*
+    Bookie can crash or get killed by an operator/automation at any point for any reason.
+    This test covers the situation when this happens mid-checkpoint.
+     */
+    @Test
+    public void testCheckpointRestoreAfterCrash() throws Exception {
+
+        final int numGoodCheckpoints = 3;
+        createMultipleCheckpoints(numGoodCheckpoints, false, false);
+
+        final int numKvs = 100;
+        final String dbName = runtime.getMethodName();
+        final byte[] txid = runtime.getMethodName().getBytes(UTF_8);
+
+        // first prepare rocksdb with 100 kvs;
+        writeNumKvs(numKvs, 100 * numGoodCheckpoints);
+
+        // create a checkpoint with corrupt metadata
+        Checkpoint checkpoint = Checkpoint.create(store.getDb());
+
+        // checkpoint
+        RocksdbCheckpointTask checkpointTask = new RocksdbCheckpointTask(
+                dbName,
+                checkpoint,
+                localCheckpointsDir,
+                checkpointStore,
+                false,
+                false);
+
+        // let's simulate the crash.
+        // crash happens after the createDirectories() succeeded but before
+        // the finalizeCheckpoint() completes.
+        final AtomicReference<String> idRef = new AtomicReference<>();
+        checkpointTask.setInjectedError((id) -> {
+            idRef.set(id);
+            throw new RuntimeException("test");
+        });
+
+        try {
+            checkpointTask.checkpoint(txid);
+            fail("expected RuntimeException");
+        } catch (RuntimeException se) {
+            // noop
+            // in real life case ths is simply crash,
+            // so "finally" at the checkpoint() won't run either
+        }
+
+        // remove local checkpointed dir
+        File checkpointedDir = new File(localCheckpointsDir, idRef.get());
+        MoreFiles.deleteRecursively(
+                Paths.get(checkpointedDir.getAbsolutePath()),
+                RecursiveDeleteOption.ALLOW_INSECURE);
+        assertFalse(checkpointedDir.exists());
+        store.close();
+
+        // restore the checkpoint
+        RocksCheckpointer.restore(dbName, localCheckpointsDir, checkpointStore);
+
+        // al of the following succeeds if the exception from RocksCheckpointer.restore
+        // is ignored
+
+        // make sure all the kvs are readable
+        store = new RocksdbKVStore<>();
+        store.init(spec);
+
+        verifyNumKvs((numGoodCheckpoints + 1) * numKvs);
+        writeNumKvs(numKvs, (numGoodCheckpoints + 1) * numKvs);
+        verifyNumKvs((numGoodCheckpoints + 2) * numKvs);
+    }
+
 }