HDDS-5821 Container cache management for closing RockDB

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/ContainerCache.java

@@ -143,13 +143,15 @@ public ReferenceCountedDB getDB(long containerID, String containerDBType,
       lock.lock();
       try {
         db = (ReferenceCountedDB) this.get(containerDBPath);
-        if (db != null) {
+        if (db != null && !db.isClosed()) {
           metrics.incNumCacheHits();
           db.incrementReference();
           return db;
-        } else {
-          metrics.incNumCacheMisses();
         }
+        if (db != null && db.isClosed()) {
+          removeDB(containerDBPath);
+        }
+        metrics.incNumCacheMisses();
       } finally {
         lock.unlock();
       }
@@ -170,18 +172,19 @@ public ReferenceCountedDB getDB(long containerID, String containerDBType,
       try {
         ReferenceCountedDB currentDB =
             (ReferenceCountedDB) this.get(containerDBPath);
-        if (currentDB != null) {
+        if (currentDB != null && !currentDB.isClosed()) {
           // increment the reference before returning the object
           currentDB.incrementReference();
           // clean the db created in previous step
           cleanupDb(db);
           return currentDB;
-        } else {
-          this.put(containerDBPath, db);
-          // increment the reference before returning the object
-          db.incrementReference();
-          return db;
+        } else if (currentDB != null && currentDB.isClosed()) {
+          removeDB(containerDBPath);
         }
+        this.put(containerDBPath, db);
+        // increment the reference before returning the object
+        db.incrementReference();
+        return db;
       } finally {
         lock.unlock();
       }
@@ -200,8 +203,11 @@ public void removeDB(String containerDBPath) {
     try {
       ReferenceCountedDB db = (ReferenceCountedDB)this.get(containerDBPath);
       if (db != null) {
-        Preconditions.checkArgument(cleanupDb(db), "refCount:",
-            db.getReferenceCount());
+        boolean cleaned = cleanupDb(db);
+        if (!db.isClosed()) {
+          Preconditions.checkArgument(cleaned, "refCount:",
+              db.getReferenceCount());
+        }
       }
       this.remove(containerDBPath);
     } finally {

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/ReferenceCountedDB.java

@@ -26,6 +26,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.Closeable;
+import java.io.IOException;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
@@ -70,7 +71,7 @@ public void decrementReference() {
   }
 
   public boolean cleanup() {
-    if (referenceCount.get() == 0 && store != null) {
+    if (store != null && store.isClosed() || referenceCount.get() == 0) {
       if (LOG.isDebugEnabled()) {
         LOG.debug("Close {} refCnt {}", containerDBPath,
             referenceCount.get());
@@ -80,7 +81,7 @@ public boolean cleanup() {
         return true;
       } catch (Exception e) {
         LOG.error("Error closing DB. Container: " + containerDBPath, e);
-        return false;
+        return true;
       }
     } else {
       return false;
@@ -92,7 +93,15 @@ public DatanodeStore getStore() {
   }
 
   @Override
-  public void close() {
+  public void close() throws IOException {
     decrementReference();
   }
+
+  /**
+   * Returns if the underlying DB is closed. This call is threadsafe.
+   * @return true if the DB is closed.
+   */
+  public boolean isClosed() {
+    return store.isClosed();
+  }
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/AbstractDatanodeStore.java

@@ -189,6 +189,19 @@ public BlockIterator<BlockData> getBlockIterator(KeyPrefixFilter filter) {
             blockDataTableWithIterator.iterator(), filter);
   }
 
+  @Override
+  public boolean isClosed() {
+    if (this.store == null) {
+      return true;
+    }
+    return this.store.isClosed();
+  }
+
+  @Override
+  public void close() throws IOException {
+    this.store.close();
+  }
+
   @Override
   public void flushDB() throws IOException {
     store.flushDB();

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeStore.java

@@ -27,12 +27,13 @@
 import org.apache.hadoop.ozone.container.common.helpers.ChunkInfoList;
 import org.apache.hadoop.ozone.container.common.interfaces.BlockIterator;
 
+import java.io.Closeable;
 import java.io.IOException;
 
 /**
  * Interface for interacting with datanode databases.
  */
-public interface DatanodeStore {
+public interface DatanodeStore extends Closeable {
 
   /**
    * Start datanode manager.
@@ -91,4 +92,11 @@ public interface DatanodeStore {
 
   BlockIterator<BlockData>
       getBlockIterator(MetadataKeyFilters.KeyPrefixFilter filter);
+
+  /**
+   * Returns if the underlying DB is closed. This call is thread safe.
+   * @return true if the DB is closed.
+   */
+  boolean isClosed();
+
 }

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestContainerCache.java

@@ -191,4 +191,41 @@ public void testConcurrentDBGet() throws Exception {
     Assert.assertEquals(1, cache.size());
     db.cleanup();
   }
+
+  @Test
+  public void testUnderlyingDBzIsClosed() throws Exception {
+    File root = new File(testRoot);
+    root.mkdirs();
+
+    OzoneConfiguration conf = new OzoneConfiguration();
+    conf.setInt(OzoneConfigKeys.OZONE_CONTAINER_CACHE_SIZE, 2);
+
+    ContainerCache cache = ContainerCache.getInstance(conf);
+    cache.clear();
+    Assert.assertEquals(0, cache.size());
+    File containerDir1 = new File(root, "cont100");
+
+    createContainerDB(conf, containerDir1);
+    ReferenceCountedDB db1 = cache.getDB(100, "RocksDB",
+        containerDir1.getPath(),
+        VersionedDatanodeFeatures.SchemaV2.chooseSchemaVersion(), conf);
+    ReferenceCountedDB db2 = cache.getDB(100, "RocksDB",
+        containerDir1.getPath(),
+        VersionedDatanodeFeatures.SchemaV2.chooseSchemaVersion(), conf);
+    Assert.assertEquals(db1, db2);
+    db1.getStore().getStore().close();
+    ReferenceCountedDB db3 = cache.getDB(100, "RocksDB",
+        containerDir1.getPath(),
+        VersionedDatanodeFeatures.SchemaV2.chooseSchemaVersion(), conf);
+    ReferenceCountedDB db4 = cache.getDB(100, "RocksDB",
+        containerDir1.getPath(),
+        VersionedDatanodeFeatures.SchemaV2.chooseSchemaVersion(), conf);
+    Assert.assertNotEquals(db3, db2);
+    Assert.assertEquals(db4, db3);
+    db1.close();
+    db2.close();
+    db3.close();
+    db4.close();
+    cache.clear();
+  }
 }

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainer.java

@@ -28,6 +28,7 @@
     .StorageContainerException;
 import org.apache.hadoop.hdds.utils.db.DBProfile;
 import org.apache.hadoop.hdds.utils.db.RDBStore;
+import org.apache.hadoop.hdds.utils.db.RocksDatabase.ColumnFamily;
 import org.apache.hadoop.hdds.utils.db.Table;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.container.common.helpers.BlockData;
@@ -57,9 +58,7 @@
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.mockito.Mockito;
-import org.rocksdb.ColumnFamilyHandle;
 import org.rocksdb.ColumnFamilyOptions;
-import org.rocksdb.RocksDBException;
 
 import java.io.File;
 
@@ -504,8 +503,7 @@ public void testUpdateContainerUnsupportedRequest() throws Exception {
   }
 
   @Test
-  public void testContainerRocksDB()
-      throws StorageContainerException, RocksDBException {
+  public void testContainerRocksDB() throws Exception {
     closeContainer();
     keyValueContainer = new KeyValueContainer(
         keyValueContainerData, CONF);
@@ -518,7 +516,7 @@ public void testContainerRocksDB()
       long cacheSize = Long.parseLong(store
           .getProperty("rocksdb.block-cache-capacity"));
       Assert.assertEquals(defaultCacheSize, cacheSize);
-      for (ColumnFamilyHandle handle : store.getColumnFamilyHandles()) {
+      for (ColumnFamily handle : store.getColumnFamilies()) {
         cacheSize = Long.parseLong(
             store.getProperty(handle, "rocksdb.block-cache-capacity"));
         Assert.assertEquals(defaultCacheSize, cacheSize);

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/DBStore.java

@@ -19,6 +19,7 @@
 
 package org.apache.hadoop.hdds.utils.db;
 
+import java.io.Closeable;
 import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
@@ -35,7 +36,7 @@
  *
  */
 @InterfaceStability.Evolving
-public interface DBStore extends AutoCloseable, BatchOperationHandler {
+public interface DBStore extends Closeable, BatchOperationHandler {
 
   /**
    * Gets an existing TableStore.
@@ -197,4 +198,10 @@ DBUpdatesWrapper getUpdatesSince(long sequenceNumber)
    */
   DBUpdatesWrapper getUpdatesSince(long sequenceNumber, long limitCount)
       throws SequenceNumberNotFoundException;
+
+  /**
+   * Return if the underlying DB is closed. This call is thread safe.
+   * @return true if the DB is closed.
+   */
+  boolean isClosed();
 }

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/RDBBatchOperation.java

@@ -18,14 +18,12 @@
  */
 package org.apache.hadoop.hdds.utils.db;
 
-import java.io.IOException;
-
-import org.rocksdb.ColumnFamilyHandle;
-import org.rocksdb.RocksDB;
-import org.rocksdb.RocksDBException;
+import org.apache.hadoop.hdds.utils.db.RocksDatabase.ColumnFamily;
 import org.rocksdb.WriteBatch;
 import org.rocksdb.WriteOptions;
 
+import java.io.IOException;
+
 /**
  * Batch operation implementation for rocks db.
  */
@@ -41,33 +39,26 @@ public RDBBatchOperation(WriteBatch writeBatch) {
     this.writeBatch = writeBatch;
   }
 
-  public void commit(RocksDB db, WriteOptions writeOptions) throws IOException {
-    try {
-      db.write(writeOptions, writeBatch);
-    } catch (RocksDBException e) {
-      throw new IOException("Unable to write the batch.", e);
-    }
+  public void commit(RocksDatabase db) throws IOException {
+    db.batchWrite(writeBatch);
+  }
+
+  public void commit(RocksDatabase db, WriteOptions writeOptions)
+      throws IOException {
+    db.batchWrite(writeBatch, writeOptions);
   }
 
   @Override
   public void close() {
     writeBatch.close();
   }
 
-  public void delete(ColumnFamilyHandle handle, byte[] key) throws IOException {
-    try {
-      writeBatch.delete(handle, key);
-    } catch (RocksDBException e) {
-      throw new IOException("Can't record batch delete operation.", e);
-    }
+  public void delete(ColumnFamily family, byte[] key) throws IOException {
+    family.batchDelete(writeBatch, key);
   }
 
-  public void put(ColumnFamilyHandle handle, byte[] key, byte[] value)
+  public void put(ColumnFamily family, byte[] key, byte[] value)
       throws IOException {
-    try {
-      writeBatch.put(handle, key, value);
-    } catch (RocksDBException e) {
-      throw new IOException("Can't record batch put operation.", e);
-    }
+    family.batchPut(writeBatch, key, value);
   }
 }

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/RDBCheckpointManager.java

@@ -19,15 +19,15 @@
 
 package org.apache.hadoop.hdds.utils.db;
 
+import java.io.IOException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.time.Duration;
 import java.time.Instant;
 
 import org.apache.commons.lang3.StringUtils;
-import org.rocksdb.Checkpoint;
-import org.rocksdb.RocksDB;
-import org.rocksdb.RocksDBException;
+import org.apache.hadoop.hdds.utils.db.RocksDatabase.RocksCheckpoint;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -36,29 +36,21 @@
  */
 public class RDBCheckpointManager {
 
-  private final Checkpoint checkpoint;
-  private final RocksDB db;
+  private final RocksCheckpoint checkpoint;
   public static final String RDB_CHECKPOINT_DIR_PREFIX = "checkpoint_";
   private static final Logger LOG =
       LoggerFactory.getLogger(RDBCheckpointManager.class);
-  private String checkpointNamePrefix = "";
-
-  public RDBCheckpointManager(RocksDB rocksDB) {
-    this.db = rocksDB;
-    this.checkpoint = Checkpoint.create(rocksDB);
-  }
+  private final String checkpointNamePrefix;
 
   /**
    * Create a checkpoint manager with a prefix to be added to the
    * snapshots created.
    *
-   * @param rocksDB          DB instance
    * @param checkpointPrefix prefix string.
    */
-  public RDBCheckpointManager(RocksDB rocksDB, String checkpointPrefix) {
-    this.db = rocksDB;
+  public RDBCheckpointManager(RocksDatabase db, String checkpointPrefix) {
     this.checkpointNamePrefix = checkpointPrefix;
-    this.checkpoint = Checkpoint.create(rocksDB);
+    this.checkpoint = db.createCheckpoint();
   }
 
   /**
@@ -79,20 +71,21 @@ public RocksDBCheckpoint createCheckpoint(String parentDir) {
 
       Path checkpointPath = Paths.get(parentDir, checkpointDir);
       Instant start = Instant.now();
-      checkpoint.createCheckpoint(checkpointPath.toString());
+      checkpoint.createCheckpoint(checkpointPath);
+      //Best guesstimate here. Not accurate.
+      final long latest = checkpoint.getLatestSequenceNumber();
       Instant end = Instant.now();
 
       long duration = Duration.between(start, end).toMillis();
       LOG.info("Created checkpoint at {} in {} milliseconds",
-              checkpointPath.toString(), duration);
+              checkpointPath, duration);
 
       return new RocksDBCheckpoint(
           checkpointPath,
           currentTime,
-          db.getLatestSequenceNumber(), //Best guesstimate here. Not accurate.
+          latest,
           duration);
-
-    } catch (RocksDBException e) {
+    } catch (IOException e) {
       LOG.error("Unable to create RocksDB Snapshot.", e);
     }
     return null;