HDFS-12966. Ozone: owner name should be set properly when the container allocation happens. Contributed by Shashikant Banerjee.

13 files changed, 76 insertions, 55 deletions

diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/CBlockManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/CBlockManager.java
index 57c47d7e35..bb4252c901 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/CBlockManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/CBlockManager.java
@@ -62,6 +62,7 @@ import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 
 import static org.apache.hadoop.cblock.CBlockConfigKeys
     .DFS_CBLOCK_CONTAINER_SIZE_GB_DEFAULT;
@@ -122,7 +123,11 @@ public class CBlockManager implements CBlockServiceProtocol,
 
   public CBlockManager(OzoneConfiguration conf,
       ScmClient storageClient) throws IOException {
-    storageManager = new StorageManager(storageClient, conf);
+    // Fix the cBlockManagerId generattion code here. Should support
+    // cBlockManager --init command which will generate a cBlockManagerId and
+    // persist it locally.
+    storageManager =
+        new StorageManager(storageClient, conf, "CBLOCK");
 
     dbPath = conf.getTrimmed(DFS_CBLOCK_SERVICE_LEVELDB_PATH_KEY,
         DFS_CBLOCK_SERVICE_LEVELDB_PATH_DEFAULT);
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/storage/StorageManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/storage/StorageManager.java
index edab8225ff..65b9b49574 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/storage/StorageManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/cblock/storage/StorageManager.java
@@ -58,6 +58,7 @@ public class StorageManager {
   private static final int MAX_THREADS =
       Runtime.getRuntime().availableProcessors() * 2;
   private static final int MAX_QUEUE_CAPACITY = 1024;
+  private final String cblockId;
 
   /**
    * We will NOT have the situation where same kv pair getting
@@ -77,13 +78,14 @@ public class StorageManager {
   private long containerSizeB;
 
   public StorageManager(ScmClient storageClient,
-      OzoneConfiguration ozoneConfig) throws IOException {
+      OzoneConfiguration ozoneConfig, String cblockId) throws IOException {
     this.storageClient = storageClient;
     this.user2VolumeMap = new ConcurrentHashMap<>();
     this.containerSizeB = storageClient.getContainerSize(null);
     this.numThreads =
         ozoneConfig.getInt(CBlockConfigKeys.DFS_CBLOCK_MANAGER_POOL_SIZE,
             CBlockConfigKeys.DFS_CBLOCK_MANAGER_POOL_SIZE_DEFAULT);
+    this.cblockId = cblockId;
   }
 
   /**
@@ -188,7 +190,7 @@ public class StorageManager {
             OzoneProtos.ReplicationType.STAND_ALONE,
             OzoneProtos.ReplicationFactor.ONE,
             KeyUtil.getContainerName(volume.getUserName(),
-                volume.getVolumeName(), containerIdx));
+                volume.getVolumeName(), containerIdx), cblockId);
 
         container = new ContainerDescriptor(pipeline.getContainerName());
 
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeyManagerImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeyManagerImpl.java
index b4a13dec91..10c61fc96c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeyManagerImpl.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeyManagerImpl.java
@@ -95,9 +95,11 @@ public class KeyManagerImpl implements KeyManager {
 
   private final long preallocateMax;
   private final Random random;
+  private final String ksmId;
 
   public KeyManagerImpl(ScmBlockLocationProtocol scmBlockClient,
-      KSMMetadataManager metadataManager, OzoneConfiguration conf) {
+      KSMMetadataManager metadataManager, OzoneConfiguration conf,
+      String ksmId) {
     this.scmBlockClient = scmBlockClient;
     this.metadataManager = metadataManager;
     this.scmBlockSize = conf.getLong(OZONE_SCM_BLOCK_SIZE_IN_MB,
@@ -123,6 +125,7 @@ public class KeyManagerImpl implements KeyManager {
     openKeyCleanupService = new OpenKeyCleanupService(
         scmBlockClient, this, openkeyCheckInterval, serviceTimeout);
     random = new Random();
+    this.ksmId = ksmId;
   }
 
   @Override
@@ -190,7 +193,7 @@ public class KeyManagerImpl implements KeyManager {
             KSMException.ResultCodes.FAILED_KEY_NOT_FOUND);
       }
       AllocatedBlock allocatedBlock =
-          scmBlockClient.allocateBlock(scmBlockSize, type, factor);
+          scmBlockClient.allocateBlock(scmBlockSize, type, factor, ksmId);
       KsmKeyInfo keyInfo =
           KsmKeyInfo.getFromProtobuf(KeyInfo.parseFrom(keyData));
       KsmKeyLocationInfo info = new KsmKeyLocationInfo.Builder()
@@ -245,7 +248,7 @@ public class KeyManagerImpl implements KeyManager {
       while (requestedSize > 0) {
         long allocateSize = Math.min(scmBlockSize, requestedSize);
         AllocatedBlock allocatedBlock =
-            scmBlockClient.allocateBlock(allocateSize, type, factor);
+            scmBlockClient.allocateBlock(allocateSize, type, factor, ksmId);
         KsmKeyLocationInfo subKeyInfo = new KsmKeyLocationInfo.Builder()
             .setContainerName(allocatedBlock.getPipeline().getContainerName())
             .setBlockID(allocatedBlock.getKey())
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeySpaceManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeySpaceManager.java
index 9147390bd7..ef311651bb 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeySpaceManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeySpaceManager.java
@@ -182,8 +182,9 @@ public class KeySpaceManager extends ServiceRuntimeInfoImpl
     volumeManager = new VolumeManagerImpl(metadataManager, configuration);
     bucketManager = new BucketManagerImpl(metadataManager);
     metrics = KSMMetrics.create();
-    keyManager = new KeyManagerImpl(scmBlockClient, metadataManager,
-        configuration);
+    keyManager =
+        new KeyManagerImpl(scmBlockClient, metadataManager, configuration,
+            ksmStorage.getKsmId());
     httpServer = new KeySpaceManagerHttpServer(configuration, this);
   }
 
@@ -704,7 +705,7 @@ public class KeySpaceManager extends ServiceRuntimeInfoImpl
   }
 
   @Override
-  public KsmKeyLocationInfo allocateBlock(KsmKeyArgs args, int clientID)
+  public KsmKeyLocationInfo  allocateBlock(KsmKeyArgs args, int clientID)
       throws IOException {
     try {
       metrics.incNumBlockAllocateCalls();
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/protocolPB/ScmBlockLocationProtocolServerSideTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/protocolPB/ScmBlockLocationProtocolServerSideTranslatorPB.java
index 9bdfe983f5..2ab11badeb 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/protocolPB/ScmBlockLocationProtocolServerSideTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/protocolPB/ScmBlockLocationProtocolServerSideTranslatorPB.java
@@ -107,7 +107,7 @@ public final class ScmBlockLocationProtocolServerSideTranslatorPB
     try {
       AllocatedBlock allocatedBlock =
           impl.allocateBlock(request.getSize(), request.getType(),
-              request.getFactor());
+              request.getFactor(), request.getOwner());
       if (allocatedBlock != null) {
         return
             AllocateScmBlockResponseProto.newBuilder()
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/protocolPB/StorageContainerLocationProtocolServerSideTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/protocolPB/StorageContainerLocationProtocolServerSideTranslatorPB.java
index df1fa40418..ae99a373c1 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/protocolPB/StorageContainerLocationProtocolServerSideTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/protocolPB/StorageContainerLocationProtocolServerSideTranslatorPB.java
@@ -76,7 +76,8 @@ public final class StorageContainerLocationProtocolServerSideTranslatorPB
       ContainerRequestProto request) throws ServiceException {
     try {
       Pipeline pipeline = impl.allocateContainer(request.getReplicationType(),
-          request.getReplicationFactor(), request.getContainerName());
+          request.getReplicationFactor(), request.getContainerName(),
+          request.getOwner());
       return ContainerResponseProto.newBuilder()
           .setPipeline(pipeline.getProtobufMessage())
           .setErrorCode(ContainerResponseProto.Error.success)
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/StorageContainerManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/StorageContainerManager.java
index 42bb082a56..ca1fa4865f 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/StorageContainerManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/StorageContainerManager.java
@@ -788,16 +788,13 @@ public class StorageContainerManager extends ServiceRuntimeInfoImpl
    */
   @Override
   public Pipeline allocateContainer(OzoneProtos.ReplicationType replicationType,
-      OzoneProtos.ReplicationFactor replicationFactor, String containerName)
-      throws IOException {
+      OzoneProtos.ReplicationFactor replicationFactor, String containerName,
+      String owner) throws IOException {
 
-    //TODO : FIX ME : Pass the owner argument to this function.
-    // This causes a lot of test change and cblock change to filing
-    // another JIRA to fix it.
-    final OzoneProtos.Owner owner = OzoneProtos.Owner.OZONE;
     checkAdminAccess();
-    return scmContainerManager.allocateContainer(replicationType,
-        replicationFactor, containerName, owner).getPipeline();
+    return scmContainerManager
+        .allocateContainer(replicationType, replicationFactor, containerName,
+            owner).getPipeline();
   }
 
   /**
@@ -1115,9 +1112,10 @@ public class StorageContainerManager extends ServiceRuntimeInfoImpl
    * @throws IOException
    */
   @Override
-  public AllocatedBlock allocateBlock(long size, OzoneProtos.ReplicationType
-      type, OzoneProtos.ReplicationFactor factor) throws IOException {
-    return scmBlockManager.allocateBlock(size, type, factor);
+  public AllocatedBlock allocateBlock(long size,
+      OzoneProtos.ReplicationType type, OzoneProtos.ReplicationFactor factor,
+      String owner) throws IOException {
+    return scmBlockManager.allocateBlock(size, type, factor, owner);
   }
 
   /**
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/block/BlockManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/block/BlockManager.java
index b0f4da6c36..cfed7a8bf6 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/block/BlockManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/block/BlockManager.java
@@ -40,7 +40,7 @@ public interface BlockManager extends Closeable {
    * @throws IOException
    */
   AllocatedBlock allocateBlock(long size, OzoneProtos.ReplicationType type,
-      OzoneProtos.ReplicationFactor factor) throws IOException;
+      OzoneProtos.ReplicationFactor factor, String owner) throws IOException;
 
   /**
    *  Give the key to the block, get the pipeline info.
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/block/BlockManagerImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/block/BlockManagerImpl.java
index 345db1ad15..57710802ba 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/block/BlockManagerImpl.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/block/BlockManagerImpl.java
@@ -22,7 +22,6 @@ import org.apache.hadoop.hdfs.DFSUtil;
 import org.apache.hadoop.metrics2.util.MBeans;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.protocol.proto.OzoneProtos;
-import org.apache.hadoop.ozone.protocol.proto.OzoneProtos.Owner;
 import org.apache.hadoop.ozone.protocol.proto.OzoneProtos.ReplicationFactor;
 import org.apache.hadoop.ozone.protocol.proto.OzoneProtos.ReplicationType;
 import org.apache.hadoop.ozone.scm.container.Mapping;
@@ -76,7 +75,6 @@ public class BlockManagerImpl implements BlockManager, BlockmanagerMXBean {
   // TODO : FIX ME : Hard coding the owner.
   // Currently only user of the block service is Ozone, CBlock manages blocks
   // by itself and does not rely on the Block service offered by SCM.
-  private final Owner owner = Owner.OZONE;
 
   private final NodeManager nodeManager;
   private final Mapping containerManager;
@@ -178,7 +176,7 @@ public class BlockManagerImpl implements BlockManager, BlockmanagerMXBean {
    * @throws IOException
    */
   private void preAllocateContainers(int count, ReplicationType type,
-      ReplicationFactor factor)
+      ReplicationFactor factor, String owner)
       throws IOException {
     lock.lock();
     try {
@@ -214,8 +212,8 @@ public class BlockManagerImpl implements BlockManager, BlockmanagerMXBean {
    * @throws IOException on failure.
    */
   @Override
-  public AllocatedBlock allocateBlock(
-      final long size, ReplicationType type, ReplicationFactor factor)
+  public AllocatedBlock allocateBlock(final long size,
+      ReplicationType type, ReplicationFactor factor, String owner)
       throws IOException {
     LOG.trace("Size;{} , type : {}, factor : {} ", size, type, factor);
 
@@ -289,7 +287,7 @@ public class BlockManagerImpl implements BlockManager, BlockmanagerMXBean {
       // that most of our containers are full or we have not allocated
       // containers of the type and replication factor. So let us go and
       // allocate some.
-      preAllocateContainers(containerProvisionBatchSize, type, factor);
+      preAllocateContainers(containerProvisionBatchSize, type, factor, owner);
 
       // Since we just allocated a set of containers this should work
       containerInfo =
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/cli/container/CreateContainerHandler.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/cli/container/CreateContainerHandler.java
index b68d872054..569c4a3558 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/cli/container/CreateContainerHandler.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/cli/container/CreateContainerHandler.java
@@ -36,6 +36,7 @@ public class CreateContainerHandler extends OzoneCommandHandler {
 
   public static final String CONTAINER_CREATE = "create";
   public static final String OPT_CONTAINER_NAME = "c";
+  public static final String containerOwner = "OZONE";
   // TODO Support an optional -p <pipelineID> option to create
   // container on given datanodes.
 
@@ -59,7 +60,7 @@ public class CreateContainerHandler extends OzoneCommandHandler {
     String containerName = cmd.getOptionValue(OPT_CONTAINER_NAME);
 
     logOut("Creating container : %s.", containerName);
-    getScmClient().createContainer(containerName);
+    getScmClient().createContainer(containerName, containerOwner);
     logOut("Container created.");
   }
 
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/container/ContainerMapping.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/container/ContainerMapping.java
index 4ae18833fa..79974d9ca7 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/container/ContainerMapping.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/container/ContainerMapping.java
@@ -26,7 +26,6 @@ import org.apache.hadoop.ozone.lease.Lease;
 import org.apache.hadoop.ozone.lease.LeaseException;
 import org.apache.hadoop.ozone.lease.LeaseManager;
 import org.apache.hadoop.ozone.protocol.proto.OzoneProtos;
-import org.apache.hadoop.ozone.protocol.proto.OzoneProtos.Owner;
 import org.apache.hadoop.ozone.protocol.proto.OzoneProtos.ReplicationFactor;
 import org.apache.hadoop.ozone.protocol.proto.OzoneProtos.ReplicationType;
 import org.apache.hadoop.ozone.protocol.proto.StorageContainerDatanodeProtocolProtos;
@@ -198,7 +197,7 @@ public class ContainerMapping implements Mapping {
       ReplicationType type,
       ReplicationFactor replicationFactor,
       final String containerName,
-      Owner owner)
+      String owner)
       throws IOException {
     Preconditions.checkNotNull(containerName);
     Preconditions.checkState(!containerName.isEmpty());
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/container/ContainerStateManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/container/ContainerStateManager.java
index 9b67473c73..0240f692ae 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/container/ContainerStateManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/container/ContainerStateManager.java
@@ -54,7 +54,6 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 import org.apache.hadoop.ozone.protocol.proto.OzoneProtos.LifeCycleState;
 import org.apache.hadoop.ozone.protocol.proto.OzoneProtos.ReplicationType;
-import org.apache.hadoop.ozone.protocol.proto.OzoneProtos.Owner;
 import org.apache.hadoop.ozone.protocol.proto.OzoneProtos.LifeCycleEvent;
 import org.apache.hadoop.ozone.protocol.proto.OzoneProtos.ReplicationFactor;
 
@@ -167,7 +166,6 @@ public class ContainerStateManager implements Closeable {
 
     lock = new ReentrantReadWriteLock();
     containers = new HashMap<>();
-    initializeContainerMaps();
     loadExistingContainers(containerMapping);
     containerCloseQueue = new ConcurrentLinkedQueue<>();
   }
@@ -185,17 +183,15 @@ public class ContainerStateManager implements Closeable {
    * of these {ALLOCATED, CREATING, OPEN, CLOSED, DELETING, DELETED}  container
    * states
    */
-  private void initializeContainerMaps() {
+  private void initializeContainerMaps(String owner) {
     // Called only from Ctor path, hence no lock is held.
     Preconditions.checkNotNull(containers);
-    for (OzoneProtos.Owner owner : OzoneProtos.Owner.values()) {
-      for (ReplicationType type : ReplicationType.values()) {
-        for (ReplicationFactor factor : ReplicationFactor.values()) {
-          for (LifeCycleState state : LifeCycleState.values()) {
-            ContainerKey key = new ContainerKey(owner, type, factor, state);
-            PriorityQueue<ContainerInfo> queue = new PriorityQueue<>();
-            containers.put(key, queue);
-          }
+    for (ReplicationType type : ReplicationType.values()) {
+      for (ReplicationFactor factor : ReplicationFactor.values()) {
+        for (LifeCycleState state : LifeCycleState.values()) {
+          ContainerKey key = new ContainerKey(owner, type, factor, state);
+          PriorityQueue<ContainerInfo> queue = new PriorityQueue<>();
+          containers.put(key, queue);
         }
       }
     }
@@ -208,12 +204,18 @@ public class ContainerStateManager implements Closeable {
    */
   private void loadExistingContainers(Mapping containerMapping) {
     try {
+      List<String> ownerList = new ArrayList<>();
       List<ContainerInfo> containerList =
           containerMapping.listContainer(null, null, Integer.MAX_VALUE);
       for (ContainerInfo container : containerList) {
-        ContainerKey key = new ContainerKey(container.getOwner(),
-            container.getPipeline().getType(),
-            container.getPipeline().getFactor(), container.getState());
+        String owner = container.getOwner();
+        if (ownerList.isEmpty() || !ownerList.contains(owner)) {
+          ownerList.add(owner);
+          initializeContainerMaps(owner);
+        }
+        ContainerKey key =
+            new ContainerKey(owner, container.getPipeline().getType(),
+                container.getPipeline().getFactor(), container.getState());
         containers.get(key).add(container);
       }
     } catch (IOException e) {
@@ -317,7 +319,7 @@ public class ContainerStateManager implements Closeable {
    */
   public ContainerInfo allocateContainer(PipelineSelector selector, OzoneProtos
       .ReplicationType type, OzoneProtos.ReplicationFactor replicationFactor,
-      final String containerName, OzoneProtos.Owner owner) throws
+      final String containerName, String owner) throws
       IOException {
 
     Pipeline pipeline = selector.getReplicationPipeline(type,
@@ -340,7 +342,10 @@ public class ContainerStateManager implements Closeable {
       ContainerKey key = new ContainerKey(owner, type, replicationFactor,
           containerInfo.getState());
       PriorityQueue<ContainerInfo> queue = containers.get(key);
-      Preconditions.checkNotNull(queue);
+      if (queue == null) {
+        initializeContainerMaps(owner);
+        queue = containers.get(key);
+      }
       queue.add(containerInfo);
       LOG.trace("New container allocated: {}", containerInfo);
     } finally {
@@ -431,12 +436,16 @@ public class ContainerStateManager implements Closeable {
    * @return ContainerInfo
    */
   public ContainerInfo getMatchingContainer(final long size,
-      Owner owner, ReplicationType type, ReplicationFactor factor,
+      String owner, ReplicationType type, ReplicationFactor factor,
       LifeCycleState state) {
     ContainerKey key = new ContainerKey(owner, type, factor, state);
     lock.writeLock().lock();
     try {
       PriorityQueue<ContainerInfo> queue = containers.get(key);
+      if (queue == null) {
+        initializeContainerMaps(owner);
+        queue = containers.get(key);
+      }
       if (queue.size() == 0) {
         // We don't have any Containers of this type.
         return null;
@@ -466,13 +475,17 @@ public class ContainerStateManager implements Closeable {
   }
 
   @VisibleForTesting
-  public List<ContainerInfo> getMatchingContainers(Owner owner,
+  public List<ContainerInfo> getMatchingContainers(String owner,
       ReplicationType type, ReplicationFactor factor, LifeCycleState state) {
     ContainerKey key = new ContainerKey(owner, type, factor, state);
     lock.readLock().lock();
     try {
-      return Arrays.asList((ContainerInfo[]) containers.get(key)
-          .toArray(new ContainerInfo[0]));
+      if (containers.get(key) == null) {
+        return null;
+      } else {
+        return Arrays.asList((ContainerInfo[]) containers.get(key)
+            .toArray(new ContainerInfo[0]));
+      }
     } catch (Exception e) {
       LOG.error("Could not get matching containers", e);
     } finally {
@@ -492,7 +505,7 @@ public class ContainerStateManager implements Closeable {
   private static class ContainerKey {
     private final LifeCycleState state;
     private final ReplicationType type;
-    private final OzoneProtos.Owner owner;
+    private final String owner;
     private final ReplicationFactor replicationFactor;
 
     /**
@@ -503,7 +516,7 @@ public class ContainerStateManager implements Closeable {
      * @param factor - Replication Factors
      * @param state - LifeCycle State
      */
-    ContainerKey(Owner owner, ReplicationType type,
+    ContainerKey(String owner, ReplicationType type,
         ReplicationFactor factor, LifeCycleState state) {
       this.state = state;
       this.type = type;
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/container/Mapping.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/container/Mapping.java
index ab4f3df89f..5d2518ee13 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/container/Mapping.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/container/Mapping.java
@@ -73,7 +73,7 @@ public interface Mapping extends Closeable {
    */
   ContainerInfo allocateContainer(OzoneProtos.ReplicationType type,
       OzoneProtos.ReplicationFactor replicationFactor,
-      String containerName, OzoneProtos.Owner owner) throws IOException;
+      String containerName, String owner) throws IOException;
 
   /**
    * Deletes a container from SCM.