Fix the coredump that occurs when calling KeyValueStorageRocksDB.count after rocksdb has been closed

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/KeyValueStorageRocksDB.java

@@ -74,6 +74,7 @@ public class KeyValueStorageRocksDB implements KeyValueStorage {
     static KeyValueStorageFactory factory = (defaultBasePath, subPath, dbConfigType, conf) ->
             new KeyValueStorageRocksDB(defaultBasePath, subPath, dbConfigType, conf);
 
+    private volatile boolean closed = true;
     private final RocksDB db;
     private RocksObject options;
     private List<ColumnFamilyDescriptor> columnFamilyDescriptors;
@@ -147,6 +148,7 @@ public KeyValueStorageRocksDB(String basePath, String subPath, DbConfigType dbCo
         optionDontCache.setFillCache(false);
 
         this.writeBatchMaxSize = conf.getMaxOperationNumbersInSingleRocksDBBatch();
+        this.closed = false;
     }
 
     private RocksDB initializeRocksDBWithConfFile(String basePath, String subPath, DbConfigType dbConfigType,
@@ -287,6 +289,7 @@ private RocksDB initializeRocksDBWithBookieConf(String basePath, String subPath,
 
     @Override
     public void close() throws IOException {
+        closed = true;
         db.close();
         if (cache != null) {
             cache.close();
@@ -513,6 +516,8 @@ public void close() {
     @Override
     public long count() throws IOException {
         try {
+            if (closed)
+                throw new IOException("RocksDB is closed");
             return db.getLongProperty("rocksdb.estimate-num-keys");
         } catch (RocksDBException e) {
             throw new IOException("Error in getting records count", e);

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/KeyValueStorageRocksDBTest.java

@@ -21,9 +21,11 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
 import java.io.File;
+import java.io.IOException;
 import java.net.URL;
 import java.nio.file.Files;
 import java.nio.file.Paths;
@@ -131,4 +133,19 @@ public void testLevelCompactionDynamicLevelBytesFromConfigurationFile() throws E
         ColumnFamilyOptions familyOptions = columnFamilyDescriptorList.get(0).getOptions();
         assertEquals(true, familyOptions.levelCompactionDynamicLevelBytes());
     }
+
+    @Test
+    public void testCallCountAfterClose() throws IOException {
+        ServerConfiguration configuration = new ServerConfiguration();
+        URL url = getClass().getClassLoader().getResource("test_entry_location_rocksdb.conf");
+        configuration.setEntryLocationRocksdbConf(url.getPath());
+        File tmpDir = Files.createTempDirectory("bk-kv-rocksdbtest-file").toFile();
+        Files.createDirectory(Paths.get(tmpDir.toString(), "subDir"));
+        KeyValueStorageRocksDB rocksDB = new KeyValueStorageRocksDB(tmpDir.toString(), "subDir",
+                KeyValueStorageFactory.DbConfigType.EntryLocation, configuration);
+        assertNotNull(rocksDB.getColumnFamilyDescriptors());
+        rocksDB.close();
+        IOException exception = assertThrows(IOException.class, rocksDB::count);
+        assertEquals("RocksDB is closed", exception.getMessage());
+    }
 }