HDDS-1401. Static ContainerCache in Datanodes can result in overwrite…

… of container db. Contributed by Mukul Kumar Singh. (#708)

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

@@ -69,15 +69,15 @@ public synchronized static ContainerCache getInstance(Configuration conf) {
   /**
    * Closes a db instance.
    *
-   * @param containerID - ID of the container to be closed.
+   * @param containerPath - path of the container db to be closed.
    * @param db - db instance to close.
    */
-  private void closeDB(long containerID, MetadataStore db) {
+  private void closeDB(String containerPath, MetadataStore db) {
     if (db != null) {
       try {
         db.close();
-      } catch (IOException e) {
-        LOG.error("Error closing DB. Container: " + containerID, e);
+      } catch (Exception e) {
+        LOG.error("Error closing DB. Container: " + containerPath, e);
       }
     }
   }
@@ -93,7 +93,7 @@ public void shutdownCache() {
       while (iterator.hasNext()) {
         iterator.next();
         MetadataStore db = (MetadataStore) iterator.getValue();
-        closeDB(((Number)iterator.getKey()).longValue(), db);
+        closeDB((String)iterator.getKey(), db);
       }
       // reset the cache
       cache.clear();
@@ -107,14 +107,18 @@ public void shutdownCache() {
    */
   @Override
   protected boolean removeLRU(LinkEntry entry) {
+    MetadataStore db = (MetadataStore) entry.getValue();
+    String dbFile = (String)entry.getKey();
     lock.lock();
     try {
-      MetadataStore db = (MetadataStore) entry.getValue();
-      closeDB(((Number)entry.getKey()).longValue(), db);
+      closeDB(dbFile, db);
+      return true;
+    } catch (Exception e) {
+      LOG.error("Eviction for db:{} failed", dbFile, e);
+      return false;
     } finally {
       lock.unlock();
     }
-    return true;
   }
 
   /**
@@ -133,7 +137,7 @@ public MetadataStore getDB(long containerID, String containerDBType,
         "Container ID cannot be negative.");
     lock.lock();
     try {
-      MetadataStore db = (MetadataStore) this.get(containerID);
+      MetadataStore db = (MetadataStore) this.get(containerDBPath);
 
       if (db == null) {
         db = MetadataStoreBuilder.newBuilder()
@@ -142,7 +146,7 @@ public MetadataStore getDB(long containerID, String containerDBType,
             .setConf(conf)
             .setDBType(containerDBType)
             .build();
-        this.put(containerID, db);
+        this.put(containerDBPath, db);
       }
       return db;
     } catch (Exception e) {
@@ -157,16 +161,14 @@ public MetadataStore getDB(long containerID, String containerDBType,
   /**
    * Remove a DB handler from cache.
    *
-   * @param containerID - ID of the container.
+   * @param containerPath - path of the container db file.
    */
-  public void removeDB(long containerID) {
-    Preconditions.checkState(containerID >= 0,
-        "Container ID cannot be negative.");
+  public void removeDB(String containerPath) {
     lock.lock();
     try {
-      MetadataStore db = (MetadataStore)this.get(containerID);
-      closeDB(containerID, db);
-      this.remove(containerID);
+      MetadataStore db = (MetadataStore)this.get(containerPath);
+      closeDB(containerPath, db);
+      this.remove(containerPath);
     } finally {
       lock.unlock();
     }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/BlockUtils.java

@@ -95,7 +95,7 @@ public static void removeDB(KeyValueContainerData container, Configuration
     Preconditions.checkNotNull(container);
     ContainerCache cache = ContainerCache.getInstance(conf);
     Preconditions.checkNotNull(cache);
-    cache.removeDB(container.getContainerID());
+    cache.removeDB(container.getDbFile().getAbsolutePath());
   }
 
   /**

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/BlockManagerImpl.java

@@ -160,7 +160,7 @@ public BlockData getBlock(Container container, BlockID blockID)
     }
     byte[] kData = db.get(Longs.toByteArray(blockID.getLocalID()));
     if (kData == null) {
-      throw new StorageContainerException("Unable to find the block.",
+      throw new StorageContainerException("Unable to find the block." + blockID,
           NO_SUCH_BLOCK);
     }
     ContainerProtos.BlockData blockData =

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

@@ -195,7 +195,7 @@ public void testContainerImportExport() throws Exception {
     for (int i = 0; i < numberOfKeysToWrite; i++) {
       metadataStore.put(("test" + i).getBytes(UTF_8), "test".getBytes(UTF_8));
     }
-    metadataStore.close();
+    BlockUtils.removeDB(keyValueContainerData, conf);
 
     Map<String, String> metadata = new HashMap<>();
     metadata.put("key1", "value1");

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/BlockManagerImpl.java

@@ -189,8 +189,8 @@ public AllocatedBlock allocateBlock(final long size, ReplicationType type,
           // factors are handled by pipeline creator
           pipeline = pipelineManager.createPipeline(type, factor);
         } catch (IOException e) {
-          LOG.error("pipeline creation failed type:{} factor:{}", type,
-              factor, e);
+          LOG.error("Pipeline creation failed for type:{} factor:{}",
+              type, factor, e);
           break;
         }
       } else {

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerPersistence.java

@@ -255,7 +255,7 @@ public void testDeleteContainer() throws Exception {
         "Container cannot be deleted because it is not empty.");
     container2.delete();
     Assert.assertTrue(containerSet.getContainerMapCopy()
-        .containsKey(testContainerID1));
+        .containsKey(testContainerID2));
   }
 
   @Test

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestCloseContainerByPipeline.java

@@ -34,16 +34,21 @@
 import org.apache.hadoop.ozone.client.OzoneClientFactory;
 import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
 import org.apache.hadoop.ozone.container.common.impl.ContainerData;
+import org.apache.hadoop.ozone.container.common.interfaces.Container;
+import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
+import org.apache.hadoop.ozone.container.keyvalue.helpers.BlockUtils;
 import org.apache.hadoop.ozone.om.helpers.OmKeyArgs;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.apache.hadoop.ozone.protocol.commands.CloseContainerCommand;
 import org.apache.hadoop.test.GenericTestUtils;
+import org.apache.hadoop.utils.MetadataStore;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.concurrent.TimeoutException;
@@ -221,13 +226,25 @@ public void testCloseContainerViaRatis() throws IOException,
     List<DatanodeDetails> datanodes = pipeline.getNodes();
     Assert.assertEquals(3, datanodes.size());
 
+    List<MetadataStore> metadataStores = new ArrayList<>(datanodes.size());
     for (DatanodeDetails details : datanodes) {
       Assert.assertFalse(isContainerClosed(cluster, containerID, details));
       //send the order to close the container
       cluster.getStorageContainerManager().getScmNodeManager()
           .addDatanodeCommand(details.getUuid(),
               new CloseContainerCommand(containerID, pipeline.getId()));
+      int index = cluster.getHddsDatanodeIndex(details);
+      Container dnContainer = cluster.getHddsDatanodes().get(index)
+          .getDatanodeStateMachine().getContainer().getContainerSet()
+          .getContainer(containerID);
+      metadataStores.add(BlockUtils.getDB((KeyValueContainerData) dnContainer
+          .getContainerData(), conf));
     }
+
+    // There should be as many rocks db as the number of datanodes in pipeline.
+    Assert.assertEquals(datanodes.size(),
+        metadataStores.stream().distinct().count());
+
     // Make sure that it is CLOSED
     for (DatanodeDetails datanodeDetails : datanodes) {
       GenericTestUtils.waitFor(