HDDS-2048: State check during container state transition in datanode should be lock protected (#1375)

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

@@ -82,7 +82,9 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
- * Class to perform KeyValue Container operations.
+ * Class to perform KeyValue Container operations. Any modifications to
+ * KeyValueContainer object should ideally be done via api exposed in
+ * KeyValueHandler class.
  */
 public class KeyValueContainer implements Container<KeyValueContainerData> {
 
@@ -554,6 +556,8 @@ public class KeyValueContainer implements Container<KeyValueContainerData> {
    * Acquire write lock.
    */
   public void writeLock() {
+    // TODO: The lock for KeyValueContainer object should not be exposed
+    // publicly.
     this.lock.writeLock().lock();
   }
 

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

@@ -881,16 +881,23 @@ public class KeyValueHandler extends Handler {
   @Override
   public void markContainerForClose(Container container)
       throws IOException {
+    container.writeLock();
+    try {
       // Move the container to CLOSING state only if it's OPEN
       if (container.getContainerState() == State.OPEN) {
         container.markContainerForClose();
         sendICR(container);
       }
+    } finally {
+      container.writeUnlock();
+    }
   }
 
   @Override
   public void markContainerUnhealthy(Container container)
       throws IOException {
+    container.writeLock();
+    try {
       if (container.getContainerState() != State.UNHEALTHY) {
         try {
           container.markContainerUnhealthy();
@@ -898,17 +905,22 @@ public class KeyValueHandler extends Handler {
           // explicitly catch IOException here since the this operation
           // will fail if the Rocksdb metadata is corrupted.
           long id = container.getContainerData().getContainerID();
-        LOG.warn("Unexpected error while marking container "
+          LOG.warn("Unexpected error while marking container " + id
-                +id+ " as unhealthy", ex);
+              + " as unhealthy", ex);
         } finally {
           sendICR(container);
         }
       }
+    } finally {
+      container.writeUnlock();
+    }
   }
 
   @Override
   public void quasiCloseContainer(Container container)
       throws IOException {
+    container.writeLock();
+    try {
       final State state = container.getContainerState();
       // Quasi close call is idempotent.
       if (state == State.QUASI_CLOSED) {
@@ -916,19 +928,25 @@ public class KeyValueHandler extends Handler {
       }
       // The container has to be in CLOSING state.
       if (state != State.CLOSING) {
-      ContainerProtos.Result error = state == State.INVALID ?
+        ContainerProtos.Result error =
-          INVALID_CONTAINER_STATE : CONTAINER_INTERNAL_ERROR;
+            state == State.INVALID ? INVALID_CONTAINER_STATE :
-      throw new StorageContainerException("Cannot quasi close container #" +
+                CONTAINER_INTERNAL_ERROR;
-          container.getContainerData().getContainerID() + " while in " +
+        throw new StorageContainerException(
-          state + " state.", error);
+            "Cannot quasi close container #" + container.getContainerData()
+                .getContainerID() + " while in " + state + " state.", error);
       }
       container.quasiClose();
       sendICR(container);
+    } finally {
+      container.writeUnlock();
+    }
   }
 
   @Override
   public void closeContainer(Container container)
       throws IOException {
+    container.writeLock();
+    try {
       final State state = container.getContainerState();
       // Close call is idempotent.
       if (state == State.CLOSED) {
@@ -942,14 +960,18 @@ public class KeyValueHandler extends Handler {
       }
       // The container has to be either in CLOSING or in QUASI_CLOSED state.
       if (state != State.CLOSING && state != State.QUASI_CLOSED) {
-      ContainerProtos.Result error = state == State.INVALID ?
+        ContainerProtos.Result error =
-          INVALID_CONTAINER_STATE : CONTAINER_INTERNAL_ERROR;
+            state == State.INVALID ? INVALID_CONTAINER_STATE :
-      throw new StorageContainerException("Cannot close container #" +
+                CONTAINER_INTERNAL_ERROR;
-          container.getContainerData().getContainerID() + " while in " +
+        throw new StorageContainerException(
-          state + " state.", error);
+            "Cannot close container #" + container.getContainerData()
+                .getContainerID() + " while in " + state + " state.", error);
       }
       container.close();
       sendICR(container);
+    } finally {
+      container.writeUnlock();
+    }
   }
 
   @Override

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

@@ -258,14 +258,15 @@ public class BlockManagerImpl implements BlockManager {
     Preconditions.checkArgument(count > 0,
         "Count must be a positive number.");
     container.readLock();
+    try {
       List<BlockData> result = null;
-    KeyValueContainerData cData = (KeyValueContainerData) container
+      KeyValueContainerData cData =
-        .getContainerData();
+          (KeyValueContainerData) container.getContainerData();
       try (ReferenceCountedDB db = BlockUtils.getDB(cData, config)) {
         result = new ArrayList<>();
         byte[] startKeyInBytes = Longs.toByteArray(startLocalID);
-      List<Map.Entry<byte[], byte[]>> range =
+        List<Map.Entry<byte[], byte[]>> range = db.getStore()
-          db.getStore().getSequentialRangeKVs(startKeyInBytes, count,
+            .getSequentialRangeKVs(startKeyInBytes, count,
                 MetadataKeyFilters.getNormalKeyFilter());
         for (Map.Entry<byte[], byte[]> entry : range) {
           BlockData value = BlockUtils.getBlockData(entry.getValue());
@@ -274,6 +275,9 @@ public class BlockManagerImpl implements BlockManager {
         }
         return result;
       }
+    } finally {
+      container.readUnlock();
+    }
   }
 
   /**