HDDS-1209. Fix the block allocation logic in SCM when client wants to exclude all available open containers in a chosen pipeline.

Signed-off-by: Nanda kumar <nanda@apache.org>

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

@@ -35,7 +35,6 @@ import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ScmOps;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.scm.ScmUtils;
 import org.apache.hadoop.hdds.scm.chillmode.ChillModePrecheck;
-import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.ContainerManager;
 import org.apache.hadoop.hdds.scm.container.common.helpers.AllocatedBlock;
@@ -60,7 +59,6 @@ import static org.apache.hadoop.ozone.OzoneConfigKeys
     .OZONE_BLOCK_DELETING_SERVICE_TIMEOUT;
 import static org.apache.hadoop.ozone.OzoneConfigKeys
     .OZONE_BLOCK_DELETING_SERVICE_TIMEOUT_DEFAULT;
-import java.util.function.Predicate;
 
 
 /** Block Manager manages the block access for SCM. */
@@ -200,15 +198,10 @@ public class BlockManagerImpl implements BlockManager, BlockmanagerMXBean {
       }
 
       // look for OPEN containers that match the criteria.
-      containerInfo = containerManager
-          .getMatchingContainer(size, owner, pipeline);
-
-      // TODO: if getMachingContainer results in containers which are in exclude
-      // list, we may end up in this loop forever. This case needs to be
-      // addressed.
-      if (containerInfo != null && (excludeList.getContainerIds() == null
-          || !discardContainer(containerInfo.containerID(),
-          excludeList.getContainerIds()))) {
+      containerInfo = containerManager.getMatchingContainer(size, owner,
+          pipeline, excludeList.getContainerIds());
+
+      if (containerInfo != null) {
         return newBlock(containerInfo);
       }
     }
@@ -221,11 +214,6 @@ public class BlockManagerImpl implements BlockManager, BlockmanagerMXBean {
     return null;
   }
 
-  private boolean discardContainer(ContainerID containerId,
-      List<ContainerID> containers) {
-    Predicate<ContainerID> predicate = p -> p.equals(containerId);
-    return containers.parallelStream().anyMatch(predicate);
-  }
   /**
    * newBlock - returns a new block assigned to a container.
    *

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerManager.java

@@ -152,4 +152,15 @@ public interface ContainerManager extends Closeable {
    */
   ContainerInfo getMatchingContainer(long size, String owner,
       Pipeline pipeline);
+
+  /**
+   * Returns ContainerInfo which matches the requirements.
+   * @param size - the amount of space required in the container
+   * @param owner - the user which requires space in its owned container
+   * @param pipeline - pipeline to which the container should belong.
+   * @param excludedContainerIDS - containerIds to be excluded.
+   * @return ContainerInfo for the matching container.
+   */
+  ContainerInfo getMatchingContainer(long size, String owner,
+      Pipeline pipeline, List<ContainerID> excludedContainerIDS);
 }

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/SCMContainerManager.java

@@ -353,6 +353,12 @@ public class SCMContainerManager implements ContainerManager {
    */
   public ContainerInfo getMatchingContainer(final long sizeRequired,
       String owner, Pipeline pipeline) {
+    return getMatchingContainer(sizeRequired, owner, pipeline, Collections
+        .emptyList());
+  }
+
+  public ContainerInfo getMatchingContainer(final long sizeRequired,
+      String owner, Pipeline pipeline, List<ContainerID> excludedContainers) {
     try {
       //TODO: #CLUTIL See if lock is required here
       NavigableSet<ContainerID> containerIDs =
@@ -378,6 +384,7 @@ public class SCMContainerManager implements ContainerManager {
         }
       }
 
+      containerIDs.removeAll(excludedContainers);
       ContainerInfo containerInfo =
           containerStateManager.getMatchingContainer(sizeRequired, owner,
               pipeline.getId(), containerIDs);

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/container/TestContainerStateManagerIntegration.java

@@ -16,6 +16,8 @@
  */
 package org.apache.hadoop.hdds.scm.container;
 
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.NavigableSet;
@@ -195,6 +197,59 @@ public class TestContainerStateManagerIntegration {
         info.getContainerID());
   }
 
+  @Test
+  public void testGetMatchingContainerWithExcludedList() throws IOException {
+    long cid;
+    ContainerWithPipeline container1 = scm.getClientProtocolServer().
+        allocateContainer(xceiverClientManager.getType(),
+            xceiverClientManager.getFactor(), containerOwner);
+    cid = container1.getContainerInfo().getContainerID();
+
+    // each getMatchingContainer call allocates a container in the
+    // pipeline till the pipeline has numContainerPerOwnerInPipeline number of
+    // containers.
+    for (int i = 1; i < numContainerPerOwnerInPipeline; i++) {
+      ContainerInfo info = containerManager
+          .getMatchingContainer(OzoneConsts.GB * 3, containerOwner,
+              container1.getPipeline());
+      Assert.assertTrue(info.getContainerID() > cid);
+      cid = info.getContainerID();
+    }
+
+    // At this point there are already three containers in the pipeline.
+    // next container should be the same as first container
+    ContainerInfo info = containerManager
+        .getMatchingContainer(OzoneConsts.GB * 3, containerOwner,
+            container1.getPipeline(), Collections.singletonList(new
+                ContainerID(1)));
+    Assert.assertNotEquals(container1.getContainerInfo().getContainerID(),
+        info.getContainerID());
+  }
+
+
+  @Test
+  public void testCreateContainerLogicWithExcludedList() throws IOException {
+    long cid;
+    ContainerWithPipeline container1 = scm.getClientProtocolServer().
+        allocateContainer(xceiverClientManager.getType(),
+            xceiverClientManager.getFactor(), containerOwner);
+    cid = container1.getContainerInfo().getContainerID();
+
+    for (int i = 1; i < numContainerPerOwnerInPipeline; i++) {
+      ContainerInfo info = containerManager
+          .getMatchingContainer(OzoneConsts.GB * 3, containerOwner,
+              container1.getPipeline());
+      Assert.assertTrue(info.getContainerID() > cid);
+      cid = info.getContainerID();
+    }
+
+    ContainerInfo info = containerManager
+        .getMatchingContainer(OzoneConsts.GB * 3, containerOwner,
+            container1.getPipeline(), Arrays.asList(new ContainerID(1), new
+                ContainerID(2), new ContainerID(3)));
+    Assert.assertEquals(info.getContainerID(), 4);
+  }
+
   @Test
   @Ignore("TODO:HDDS-1159")
   public void testGetMatchingContainerMultipleThreads()