HDDS-1625 : ConcurrentModificationException when SCM has containers of different owners. (#883)

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

@@ -43,6 +43,7 @@ import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.NavigableSet;
@@ -469,15 +470,17 @@ public class SCMContainerManager implements ContainerManager {
    */
   private NavigableSet<ContainerID> getContainersForOwner(
       NavigableSet<ContainerID> containerIDs, String owner) {
-    for (ContainerID cid : containerIDs) {
+    Iterator<ContainerID> containerIDIterator = containerIDs.iterator();
+    while (containerIDIterator.hasNext()) {
+      ContainerID cid = containerIDIterator.next();
       try {
         if (!getContainer(cid).getOwner().equals(owner)) {
-          containerIDs.remove(cid);
+          containerIDIterator.remove();
         }
       } catch (ContainerNotFoundException e) {
         LOG.error("Could not find container info for container id={} {}", cid,
             e);
-        containerIDs.remove(cid);
+        containerIDIterator.remove();
       }
     }
     return containerIDs;

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

@@ -122,6 +122,30 @@ public class TestContainerStateManagerIntegration {
     Assert.assertEquals(3, numContainers);
   }
 
+  @Test
+  public void testAllocateContainerWithDifferentOwner() throws IOException {
+
+    // Allocate a container and verify the container info
+    ContainerWithPipeline container1 = scm.getClientProtocolServer()
+        .allocateContainer(xceiverClientManager.getType(),
+            xceiverClientManager.getFactor(), containerOwner);
+    ContainerInfo info = containerManager
+        .getMatchingContainer(OzoneConsts.GB * 3, containerOwner,
+            container1.getPipeline());
+    Assert.assertNotNull(info);
+
+    String newContainerOwner = "OZONE_NEW";
+    ContainerWithPipeline container2 = scm.getClientProtocolServer()
+        .allocateContainer(xceiverClientManager.getType(),
+            xceiverClientManager.getFactor(), newContainerOwner);
+    ContainerInfo info2 = containerManager
+        .getMatchingContainer(OzoneConsts.GB * 3, newContainerOwner,
+            container1.getPipeline());
+    Assert.assertNotNull(info2);
+
+    Assert.assertNotEquals(info.containerID(), info2.containerID());
+  }
+
   @Test
   public void testContainerStateManagerRestart() throws IOException,
       TimeoutException, InterruptedException, AuthenticationException {