hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aengin...@apache.org
Subject [2/2] hadoop git commit: HDDS-13. Refactor StorageContainerManager into seperate RPC endpoints. Contributed by Anu Engineer.
Date Tue, 01 May 2018 05:43:18 GMT
HDDS-13. Refactor StorageContainerManager into seperate RPC endpoints. Contributed by Anu Engineer.


Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/f0c3dc4c
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/f0c3dc4c
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/f0c3dc4c

Branch: refs/heads/trunk
Commit: f0c3dc4cf40575497ca6f29c037e43fa50e0ffdd
Parents: 2d319e3
Author: Anu Engineer <aengineer@apache.org>
Authored: Mon Apr 30 21:41:10 2018 -0700
Committer: Anu Engineer <aengineer@apache.org>
Committed: Mon Apr 30 21:41:10 2018 -0700

----------------------------------------------------------------------
 .../org/apache/hadoop/hdds/scm/SCMMXBean.java   |   50 -
 .../org/apache/hadoop/hdds/scm/SCMStorage.java  |   73 -
 .../hdds/scm/StorageContainerManager.java       | 1290 ------------------
 .../scm/StorageContainerManagerHttpServer.java  |   76 --
 .../hadoop/hdds/scm/node/SCMNodeManager.java    |    5 +-
 .../hdds/scm/server/SCMBlockProtocolServer.java |  222 +++
 .../scm/server/SCMClientProtocolServer.java     |  314 +++++
 .../scm/server/SCMDatanodeProtocolServer.java   |  350 +++++
 .../hadoop/hdds/scm/server/SCMMXBean.java       |   50 +
 .../hadoop/hdds/scm/server/SCMStorage.java      |   73 +
 .../scm/server/StorageContainerManager.java     |  722 ++++++++++
 .../StorageContainerManagerHttpServer.java      |   77 ++
 .../hadoop/hdds/scm/server/package-info.java    |   22 +
 .../TestStorageContainerManagerHttpServer.java  |    7 +-
 hadoop-ozone/common/src/main/bin/ozone          |    2 +-
 .../container/TestContainerStateManager.java    |   29 +-
 .../apache/hadoop/ozone/MiniOzoneCluster.java   |   34 +-
 .../hadoop/ozone/MiniOzoneClusterImpl.java      |   10 +-
 .../ozone/TestStorageContainerManager.java      |   20 +-
 .../TestStorageContainerManagerHelper.java      |    2 +-
 .../ozone/ksm/TestContainerReportWithKeys.java  |    2 +-
 .../hadoop/ozone/ksm/TestKeySpaceManager.java   |    8 +-
 .../org/apache/hadoop/ozone/scm/TestSCMCli.java |   13 +-
 .../apache/hadoop/ozone/scm/TestSCMMXBean.java  |    2 +-
 .../apache/hadoop/ozone/scm/TestSCMMetrics.java |   16 +-
 25 files changed, 1912 insertions(+), 1557 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0c3dc4c/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMMXBean.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMMXBean.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMMXBean.java
deleted file mode 100644
index 17b6814..0000000
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMMXBean.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.hadoop.hdds.scm;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.hdds.server.ServiceRuntimeInfo;
-
-import java.util.Map;
-
-/**
- *
- * This is the JMX management interface for scm information.
- */
-@InterfaceAudience.Private
-public interface SCMMXBean extends ServiceRuntimeInfo {
-
-  /**
-   * Get the SCM RPC server port that used to listen to datanode requests.
-   * @return SCM datanode RPC server port
-   */
-  String getDatanodeRpcPort();
-
-  /**
-   * Get the SCM RPC server port that used to listen to client requests.
-   * @return SCM client RPC server port
-   */
-  String getClientRpcPort();
-
-  /**
-   * Get container report info that includes container IO stats of nodes.
-   * @return The datanodeUUid to report json string mapping
-   */
-  Map<String, String> getContainerReport();
-}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0c3dc4c/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMStorage.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMStorage.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMStorage.java
deleted file mode 100644
index 27e9363..0000000
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMStorage.java
+++ /dev/null
@@ -1,73 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.hdds.scm;
-
-import org.apache.hadoop.hdds.conf.OzoneConfiguration;
-import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeType;
-import org.apache.hadoop.ozone.common.Storage;
-
-import java.io.IOException;
-import java.util.Properties;
-import java.util.UUID;
-
-import static org.apache.hadoop.hdds.server.ServerUtils.getOzoneMetaDirPath;
-import static org.apache.hadoop.ozone.OzoneConsts.SCM_ID;
-import static org.apache.hadoop.ozone.OzoneConsts.STORAGE_DIR;
-
-/**
- * SCMStorage is responsible for management of the StorageDirectories used by
- * the SCM.
- */
-public class SCMStorage extends Storage {
-
-  /**
-   * Construct SCMStorage.
-   * @throws IOException if any directories are inaccessible.
-   */
-  public SCMStorage(OzoneConfiguration conf) throws IOException {
-    super(NodeType.SCM, getOzoneMetaDirPath(conf), STORAGE_DIR);
-  }
-
-  public void setScmId(String scmId) throws IOException {
-    if (getState() == StorageState.INITIALIZED) {
-      throw new IOException("SCM is already initialized.");
-    } else {
-      getStorageInfo().setProperty(SCM_ID, scmId);
-    }
-  }
-
-  /**
-   * Retrieves the SCM ID from the version file.
-   * @return SCM_ID
-   */
-  public String getScmId() {
-    return getStorageInfo().getProperty(SCM_ID);
-  }
-
-  @Override
-  protected Properties getNodeProperties() {
-    String scmId = getScmId();
-    if (scmId == null) {
-      scmId = UUID.randomUUID().toString();
-    }
-    Properties scmProperties = new Properties();
-    scmProperties.setProperty(SCM_ID, scmId);
-    return scmProperties;
-  }
-
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0c3dc4c/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/StorageContainerManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/StorageContainerManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/StorageContainerManager.java
deleted file mode 100644
index ce0d4f8..0000000
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/StorageContainerManager.java
+++ /dev/null
@@ -1,1290 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with this
- * work for additional information regarding copyright ownership.  The ASF
- * licenses this file to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations under
- * the License.
- */
-
-package org.apache.hadoop.hdds.scm;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
-import com.google.common.cache.RemovalListener;
-import com.google.common.cache.RemovalNotification;
-import com.google.protobuf.BlockingService;
-import com.google.protobuf.InvalidProtocolBufferException;
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.hdds.HddsUtils;
-import org.apache.hadoop.hdds.scm.block.BlockManager;
-import org.apache.hadoop.hdds.scm.block.BlockManagerImpl;
-import org.apache.hadoop.hdds.scm.container.ContainerMapping;
-import org.apache.hadoop.hdds.scm.container.Mapping;
-import org.apache.hadoop.hdds.scm.container.common.helpers.AllocatedBlock;
-import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerInfo;
-import org.apache.hadoop.hdds.scm.container.common.helpers.DeleteBlockResult;
-import org.apache.hadoop.hdds.scm.container.common.helpers.Pipeline;
-import org.apache.hadoop.hdds.scm.container.placement.metrics.ContainerStat;
-import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMMetrics;
-import org.apache.hadoop.hdds.scm.exceptions.SCMException;
-import org.apache.hadoop.hdds.scm.exceptions.SCMException.ResultCodes;
-import org.apache.hadoop.hdds.scm.node.NodeManager;
-import org.apache.hadoop.hdds.scm.node.SCMNodeManager;
-import org.apache.hadoop.hdds.scm.protocol.ScmBlockLocationProtocol;
-import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocol;
-import org.apache.hadoop.hdds.scm.protocolPB.ScmBlockLocationProtocolPB;
-import org.apache.hadoop.hdds.scm.protocolPB.StorageContainerLocationProtocolPB;
-import org.apache.hadoop.hdfs.DFSUtil;
-import org.apache.hadoop.hdds.conf.OzoneConfiguration;
-import org.apache.hadoop.hdds.protocol.DatanodeDetails;
-import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
-import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeDetailsProto;
-import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState;
-import org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.ContainerBlocksDeletionACKProto;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.ContainerBlocksDeletionACKProto
-    .DeleteBlockTransactionResult;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos
-    .ContainerBlocksDeletionACKResponseProto;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.ContainerReportsRequestProto;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.ContainerReportsResponseProto;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.ReportState;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.SCMCmdType;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.SCMCommandResponseProto;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.SCMHeartbeatResponseProto;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.SCMNodeAddressList;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.SCMNodeReport;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.SCMRegisteredCmdResponseProto;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.SCMReregisterCmdResponseProto;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.SCMVersionRequestProto;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.SCMVersionResponseProto;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.SendContainerReportProto;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerLocationProtocolProtos;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerLocationProtocolProtos.ObjectStageChangeRequestProto;
-import org.apache.hadoop.hdds.server.ServiceRuntimeInfoImpl;
-import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.ipc.ProtobufRpcEngine;
-import org.apache.hadoop.ipc.RPC;
-import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
-import org.apache.hadoop.metrics2.util.MBeans;
-import org.apache.hadoop.ozone.OzoneConfigKeys;
-import org.apache.hadoop.ozone.common.BlockGroup;
-import org.apache.hadoop.ozone.common.DeleteBlockGroupResult;
-import org.apache.hadoop.ozone.common.Storage.StorageState;
-import org.apache.hadoop.ozone.common.StorageInfo;
-import org.apache.hadoop.ozone.protocol.StorageContainerDatanodeProtocol;
-import org.apache.hadoop.ozone.protocol.commands.CloseContainerCommand;
-import org.apache.hadoop.ozone.protocol.commands.DeleteBlocksCommand;
-import org.apache.hadoop.ozone.protocol.commands.RegisteredCommand;
-import org.apache.hadoop.ozone.protocol.commands.SCMCommand;
-import org.apache.hadoop.ozone.protocolPB
-    .ScmBlockLocationProtocolServerSideTranslatorPB;
-import org.apache.hadoop.ozone.protocolPB.StorageContainerDatanodeProtocolPB;
-import org.apache.hadoop.ozone.protocolPB
-    .StorageContainerDatanodeProtocolServerSideTranslatorPB;
-import org.apache.hadoop.ozone.protocolPB
-    .StorageContainerLocationProtocolServerSideTranslatorPB;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.util.GenericOptionsParser;
-import org.apache.hadoop.util.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.management.ObjectName;
-import java.io.IOException;
-import java.io.PrintStream;
-import java.net.InetSocketAddress;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.EnumSet;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeSet;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-
-import static org.apache.hadoop.hdds.scm.ScmConfigKeys
-    .OZONE_SCM_BLOCK_CLIENT_ADDRESS_KEY;
-import static org.apache.hadoop.hdds.scm.ScmConfigKeys
-    .OZONE_SCM_CLIENT_ADDRESS_KEY;
-import static org.apache.hadoop.hdds.scm.ScmConfigKeys
-    .OZONE_SCM_DATANODE_ADDRESS_KEY;
-import static org.apache.hadoop.hdds.scm.ScmConfigKeys
-    .OZONE_SCM_DB_CACHE_SIZE_DEFAULT;
-import static org.apache.hadoop.hdds.scm.ScmConfigKeys
-    .OZONE_SCM_DB_CACHE_SIZE_MB;
-import static org.apache.hadoop.hdds.scm.ScmConfigKeys
-    .OZONE_SCM_HANDLER_COUNT_DEFAULT;
-import static org.apache.hadoop.hdds.scm.ScmConfigKeys
-    .OZONE_SCM_HANDLER_COUNT_KEY;
-import static org.apache.hadoop.hdds.protocol.proto
-    .ScmBlockLocationProtocolProtos.DeleteScmBlockResult.Result;
-import static org.apache.hadoop.hdds.server.ServerUtils.updateRPCListenAddress;
-import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ENABLED;
-import static org.apache.hadoop.util.ExitUtil.terminate;
-
-/**
- * StorageContainerManager is the main entry point for the service that provides
- * information about which SCM nodes host containers.
- *
- * DataNodes report to StorageContainerManager using heartbeat
- * messages. SCM allocates containers and returns a pipeline.
- *
- * A client once it gets a pipeline (a list of datanodes) will connect to the
- * datanodes and create a container, which then can be used to store data.
- */
-@InterfaceAudience.LimitedPrivate({"HDFS", "CBLOCK", "OZONE", "HBASE"})
-public class StorageContainerManager extends ServiceRuntimeInfoImpl
-    implements StorageContainerDatanodeProtocol,
-    StorageContainerLocationProtocol, ScmBlockLocationProtocol, SCMMXBean {
-
-  private static final Logger LOG =
-      LoggerFactory.getLogger(StorageContainerManager.class);
-
-  /**
-   *  Startup options.
-   */
-  public enum StartupOption {
-    INIT("-init"),
-    CLUSTERID("-clusterid"),
-    GENCLUSTERID("-genclusterid"),
-    REGULAR("-regular"),
-    HELP("-help");
-
-    private final String name;
-    private String clusterId = null;
-
-    public void setClusterId(String cid) {
-      if(cid != null && !cid.isEmpty()) {
-        clusterId = cid;
-      }
-    }
-
-    public String getClusterId() {
-      return clusterId;
-    }
-
-    StartupOption(String arg) {
-      this.name = arg;
-    }
-
-    public String getName() {
-      return name;
-    }
-  }
-
-  /**
-   * NodeManager and container Managers for SCM.
-   */
-  private final NodeManager scmNodeManager;
-  private final Mapping scmContainerManager;
-  private final BlockManager scmBlockManager;
-  private final SCMStorage scmStorage;
-
-  /** The RPC server that listens to requests from DataNodes. */
-  private final RPC.Server datanodeRpcServer;
-  private final InetSocketAddress datanodeRpcAddress;
-
-  /** The RPC server that listens to requests from clients. */
-  private final RPC.Server clientRpcServer;
-  private final InetSocketAddress clientRpcAddress;
-
-  /** The RPC server that listens to requests from block service clients. */
-  private final RPC.Server blockRpcServer;
-  private final InetSocketAddress blockRpcAddress;
-
-  private final StorageContainerManagerHttpServer httpServer;
-
-  /** SCM mxbean. */
-  private ObjectName scmInfoBeanName;
-
-  /** SCM super user. */
-  private final String scmUsername;
-  private final Collection<String> scmAdminUsernames;
-
-  /** SCM metrics. */
-  private static SCMMetrics metrics;
-  /** Key = DatanodeUuid, value = ContainerStat. */
-  private Cache<String, ContainerStat> containerReportCache;
-
-
-  private static final String USAGE =
-      "Usage: \n ozone scm [genericOptions] "
-          + "[ " + StartupOption.INIT.getName() + " [ "
-          + StartupOption.CLUSTERID.getName() + " <cid> ] ]\n "
-          + "ozone scm [genericOptions] [ "
-          + StartupOption.GENCLUSTERID.getName() + " ]\n " +
-          "ozone scm [ "
-          + StartupOption.HELP.getName() + " ]\n";
-  /**
-   * Creates a new StorageContainerManager.  Configuration will be updated with
-   * information on the actual listening addresses used for RPC servers.
-   *
-   * @param conf configuration
-   */
-  private StorageContainerManager(OzoneConfiguration conf)
-      throws IOException {
-
-    final int handlerCount = conf.getInt(
-        OZONE_SCM_HANDLER_COUNT_KEY, OZONE_SCM_HANDLER_COUNT_DEFAULT);
-    final int cacheSize = conf.getInt(OZONE_SCM_DB_CACHE_SIZE_MB,
-        OZONE_SCM_DB_CACHE_SIZE_DEFAULT);
-
-    StorageContainerManager.initMetrics();
-    initContainerReportCache(conf);
-
-    scmStorage = new SCMStorage(conf);
-    if (scmStorage.getState() != StorageState.INITIALIZED) {
-      throw new SCMException("SCM not initialized.",
-          ResultCodes.SCM_NOT_INITIALIZED);
-    }
-    scmNodeManager = new SCMNodeManager(conf, scmStorage.getClusterID(), this);
-    scmContainerManager = new ContainerMapping(conf, scmNodeManager, cacheSize);
-    scmBlockManager = new BlockManagerImpl(conf, scmNodeManager,
-        scmContainerManager, cacheSize);
-
-    scmAdminUsernames = conf.getTrimmedStringCollection(
-        OzoneConfigKeys.OZONE_ADMINISTRATORS);
-    scmUsername = UserGroupInformation.getCurrentUser().getUserName();
-    if (!scmAdminUsernames.contains(scmUsername)) {
-      scmAdminUsernames.add(scmUsername);
-    }
-
-    RPC.setProtocolEngine(conf, StorageContainerDatanodeProtocolPB.class,
-        ProtobufRpcEngine.class);
-    RPC.setProtocolEngine(conf, StorageContainerLocationProtocolPB.class,
-        ProtobufRpcEngine.class);
-    RPC.setProtocolEngine(conf, ScmBlockLocationProtocolPB.class,
-        ProtobufRpcEngine.class);
-
-    BlockingService dnProtoPbService = StorageContainerDatanodeProtocolProtos.
-        StorageContainerDatanodeProtocolService.newReflectiveBlockingService(
-        new StorageContainerDatanodeProtocolServerSideTranslatorPB(this));
-
-    final InetSocketAddress datanodeRpcAddr =
-        HddsServerUtil.getScmDataNodeBindAddress(conf);
-    datanodeRpcServer = startRpcServer(conf, datanodeRpcAddr,
-        StorageContainerDatanodeProtocolPB.class, dnProtoPbService,
-        handlerCount);
-    datanodeRpcAddress = updateRPCListenAddress(conf,
-        OZONE_SCM_DATANODE_ADDRESS_KEY, datanodeRpcAddr, datanodeRpcServer);
-
-    // SCM Container Service RPC
-    BlockingService storageProtoPbService =
-        StorageContainerLocationProtocolProtos
-            .StorageContainerLocationProtocolService
-            .newReflectiveBlockingService(
-            new StorageContainerLocationProtocolServerSideTranslatorPB(this));
-
-    final InetSocketAddress scmAddress =
-        HddsServerUtil.getScmClientBindAddress(conf);
-    clientRpcServer = startRpcServer(conf, scmAddress,
-        StorageContainerLocationProtocolPB.class, storageProtoPbService,
-        handlerCount);
-    clientRpcAddress = updateRPCListenAddress(conf,
-        OZONE_SCM_CLIENT_ADDRESS_KEY, scmAddress, clientRpcServer);
-
-    // SCM Block Service RPC
-    BlockingService blockProtoPbService =
-        ScmBlockLocationProtocolProtos
-            .ScmBlockLocationProtocolService
-            .newReflectiveBlockingService(
-            new ScmBlockLocationProtocolServerSideTranslatorPB(this));
-
-    final InetSocketAddress scmBlockAddress =
-        HddsServerUtil.getScmBlockClientBindAddress(conf);
-    blockRpcServer = startRpcServer(conf, scmBlockAddress,
-        ScmBlockLocationProtocolPB.class, blockProtoPbService,
-        handlerCount);
-    blockRpcAddress = updateRPCListenAddress(conf,
-        OZONE_SCM_BLOCK_CLIENT_ADDRESS_KEY, scmBlockAddress, blockRpcServer);
-
-    httpServer = new StorageContainerManagerHttpServer(conf);
-
-    registerMXBean();
-  }
-
-  /**
-   * Initialize container reports cache that sent from datanodes.
-   *
-   * @param conf
-   */
-  private void initContainerReportCache(OzoneConfiguration conf) {
-    containerReportCache = CacheBuilder.newBuilder()
-        .expireAfterAccess(Long.MAX_VALUE, TimeUnit.MILLISECONDS)
-        .maximumSize(Integer.MAX_VALUE)
-        .removalListener(new RemovalListener<String, ContainerStat>() {
-          @Override
-          public void onRemoval(
-              RemovalNotification<String, ContainerStat> removalNotification) {
-            synchronized (containerReportCache) {
-              ContainerStat stat = removalNotification.getValue();
-              // remove invalid container report
-              metrics.decrContainerStat(stat);
-              LOG.debug(
-                  "Remove expired container stat entry for datanode: {}.",
-                  removalNotification.getKey());
-            }
-          }
-        }).build();
-  }
-
-  /**
-   * Builds a message for logging startup information about an RPC server.
-   *
-   * @param description RPC server description
-   * @param addr RPC server listening address
-   * @return server startup message
-   */
-  private static String buildRpcServerStartMessage(String description,
-      InetSocketAddress addr) {
-    return addr != null ? String.format("%s is listening at %s",
-        description, addr.toString()) :
-        String.format("%s not started", description);
-  }
-
-  /**
-   * Starts an RPC server, if configured.
-   *
-   * @param conf configuration
-   * @param addr configured address of RPC server
-   * @param protocol RPC protocol provided by RPC server
-   * @param instance RPC protocol implementation instance
-   * @param handlerCount RPC server handler count
-   *
-   * @return RPC server
-   * @throws IOException if there is an I/O error while creating RPC server
-   */
-  private static RPC.Server startRpcServer(OzoneConfiguration conf,
-      InetSocketAddress addr, Class<?> protocol, BlockingService instance,
-      int handlerCount)
-      throws IOException {
-    RPC.Server rpcServer = new RPC.Builder(conf)
-        .setProtocol(protocol)
-        .setInstance(instance)
-        .setBindAddress(addr.getHostString())
-        .setPort(addr.getPort())
-        .setNumHandlers(handlerCount)
-        .setVerbose(false)
-        .setSecretManager(null)
-        .build();
-
-    DFSUtil.addPBProtocol(conf, protocol, instance, rpcServer);
-    return rpcServer;
-  }
-
-  private void registerMXBean() {
-    Map<String, String> jmxProperties = new HashMap<>();
-    jmxProperties.put("component", "ServerRuntime");
-    this.scmInfoBeanName =
-        MBeans.register("StorageContainerManager",
-            "StorageContainerManagerInfo",
-            jmxProperties,
-            this);
-  }
-
-  private void unregisterMXBean() {
-    if(this.scmInfoBeanName != null) {
-      MBeans.unregister(this.scmInfoBeanName);
-      this.scmInfoBeanName = null;
-    }
-  }
-
-  /**
-   * Main entry point for starting StorageContainerManager.
-   *
-   * @param argv arguments
-   * @throws IOException if startup fails due to I/O error
-   */
-  public static void main(String[] argv) throws IOException {
-    if (DFSUtil.parseHelpArgument(argv, USAGE,
-        System.out, true)) {
-      System.exit(0);
-    }
-    try {
-      OzoneConfiguration conf = new OzoneConfiguration();
-      GenericOptionsParser hParser = new GenericOptionsParser(conf, argv);
-      if (!hParser.isParseSuccessful()) {
-        System.err.println("USAGE: " + USAGE + "\n");
-        hParser.printGenericCommandUsage(System.err);
-        System.exit(1);
-      }
-      StringUtils.startupShutdownMessage(StorageContainerManager.class,
-          argv, LOG);
-      StorageContainerManager scm = createSCM(hParser.getRemainingArgs(), conf);
-      if (scm != null) {
-        scm.start();
-        scm.join();
-      }
-    } catch (Throwable t) {
-      LOG.error("Failed to start the StorageContainerManager.", t);
-      terminate(1, t);
-    }
-  }
-
-  private static void printUsage(PrintStream out) {
-    out.println(USAGE + "\n");
-  }
-
-  public static StorageContainerManager createSCM(String[] argv,
-      OzoneConfiguration conf) throws IOException {
-    if (!HddsUtils.isHddsEnabled(conf)) {
-      System.err.println("SCM cannot be started in secure mode or when " +
-          OZONE_ENABLED + " is set to false");
-      System.exit(1);
-    }
-    StartupOption startOpt = parseArguments(argv);
-    if (startOpt == null) {
-      printUsage(System.err);
-      terminate(1);
-      return null;
-    }
-    switch (startOpt) {
-    case INIT:
-      terminate(scmInit(conf) ? 0 : 1);
-      return null;
-    case GENCLUSTERID:
-      System.out.println("Generating new cluster id:");
-      System.out.println(StorageInfo.newClusterID());
-      terminate(0);
-      return null;
-    case HELP:
-      printUsage(System.err);
-      terminate(0);
-      return null;
-    default:
-      return new StorageContainerManager(conf);
-    }
-  }
-
-  /**
-   * Routine to set up the Version info for StorageContainerManager.
-   *
-   * @param conf OzoneConfiguration
-   * @return true if SCM initialization is successful, false otherwise.
-   * @throws IOException if init fails due to I/O error
-   */
-  public static boolean scmInit(OzoneConfiguration conf) throws IOException {
-    SCMStorage scmStorage = new SCMStorage(conf);
-    StorageState state = scmStorage.getState();
-    if (state != StorageState.INITIALIZED) {
-      try {
-        String clusterId = StartupOption.INIT.getClusterId();
-        if (clusterId != null && !clusterId.isEmpty()) {
-          scmStorage.setClusterId(clusterId);
-        }
-        scmStorage.initialize();
-        System.out.println("SCM initialization succeeded." +
-            "Current cluster id for sd=" + scmStorage.getStorageDir() + ";cid="
-                + scmStorage.getClusterID());
-        return true;
-      } catch (IOException ioe) {
-        LOG.error("Could not initialize SCM version file", ioe);
-        return false;
-      }
-    } else {
-      System.out.println("SCM already initialized. Reusing existing" +
-          " cluster id for sd=" + scmStorage.getStorageDir() + ";cid="
-              + scmStorage.getClusterID());
-      return true;
-    }
-  }
-
-  private static StartupOption parseArguments(String[] args) {
-    int argsLen = (args == null) ? 0 : args.length;
-    StartupOption startOpt = StartupOption.HELP;
-    if (argsLen == 0) {
-      startOpt = StartupOption.REGULAR;
-    }
-    for (int i = 0; i < argsLen; i++) {
-      String cmd = args[i];
-      if (StartupOption.INIT.getName().equalsIgnoreCase(cmd)) {
-        startOpt = StartupOption.INIT;
-        if (argsLen > 3) {
-          return null;
-        }
-        for (i = i + 1; i < argsLen; i++) {
-          if (args[i].equalsIgnoreCase(StartupOption.CLUSTERID.getName())) {
-            i++;
-            if (i < argsLen && !args[i].isEmpty()) {
-              startOpt.setClusterId(args[i]);
-            } else {
-              // if no cluster id specified or is empty string, return null
-              LOG.error("Must specify a valid cluster ID after the "
-                  + StartupOption.CLUSTERID.getName() + " flag");
-              return null;
-            }
-          } else {
-            return null;
-          }
-        }
-      } else if (StartupOption.GENCLUSTERID.getName().equalsIgnoreCase(cmd)) {
-        if (argsLen > 1) {
-          return null;
-        }
-        startOpt = StartupOption.GENCLUSTERID;
-      }
-    }
-    return startOpt;
-  }
-
-  /**
-   * Returns a SCMCommandRepose from the SCM Command.
-   * @param cmd - Cmd
-   * @return SCMCommandResponseProto
-   * @throws InvalidProtocolBufferException
-   */
-  @VisibleForTesting
-  public SCMCommandResponseProto getCommandResponse(SCMCommand cmd,
-      final String datanodID)
-      throws IOException {
-    SCMCmdType type = cmd.getType();
-    SCMCommandResponseProto.Builder builder =
-        SCMCommandResponseProto.newBuilder()
-        .setDatanodeUUID(datanodID);
-    switch (type) {
-    case registeredCommand:
-      return builder.setCmdType(SCMCmdType.registeredCommand)
-          .setRegisteredProto(
-              SCMRegisteredCmdResponseProto.getDefaultInstance())
-          .build();
-    case versionCommand:
-      return builder.setCmdType(SCMCmdType.versionCommand)
-          .setVersionProto(SCMVersionResponseProto.getDefaultInstance())
-          .build();
-    case sendContainerReport:
-      return builder.setCmdType(SCMCmdType.sendContainerReport)
-          .setSendReport(SendContainerReportProto.getDefaultInstance())
-          .build();
-    case reregisterCommand:
-      return builder.setCmdType(SCMCmdType.reregisterCommand)
-          .setReregisterProto(SCMReregisterCmdResponseProto
-              .getDefaultInstance())
-          .build();
-    case deleteBlocksCommand:
-      // Once SCM sends out the deletion message, increment the count.
-      // this is done here instead of when SCM receives the ACK, because
-      // DN might not be able to response the ACK for sometime. In case
-      // it times out, SCM needs to re-send the message some more times.
-      List<Long> txs = ((DeleteBlocksCommand) cmd).blocksTobeDeleted()
-          .stream().map(tx -> tx.getTxID()).collect(Collectors.toList());
-      this.getScmBlockManager().getDeletedBlockLog().incrementCount(txs);
-      return builder.setCmdType(SCMCmdType.deleteBlocksCommand)
-          .setDeleteBlocksProto(((DeleteBlocksCommand) cmd).getProto())
-          .build();
-    case closeContainerCommand:
-      return builder.setCmdType(SCMCmdType.closeContainerCommand)
-          .setCloseContainerProto(((CloseContainerCommand)cmd).getProto())
-          .build();
-    default:
-      throw new IllegalArgumentException("Not implemented");
-    }
-  }
-
-  @VisibleForTesting
-  public static SCMRegisteredCmdResponseProto getRegisteredResponse(
-      SCMCommand cmd, SCMNodeAddressList addressList) {
-    Preconditions.checkState(cmd.getClass() == RegisteredCommand.class);
-    RegisteredCommand rCmd = (RegisteredCommand) cmd;
-    SCMCmdType type = cmd.getType();
-    if (type != SCMCmdType.registeredCommand) {
-      throw new IllegalArgumentException("Registered command is not well " +
-          "formed. Internal Error.");
-    }
-    return SCMRegisteredCmdResponseProto.newBuilder()
-        //TODO : Fix this later when we have multiple SCM support.
-        //.setAddressList(addressList)
-        .setErrorCode(rCmd.getError())
-        .setClusterID(rCmd.getClusterID())
-        .setDatanodeUUID(rCmd.getDatanodeUUID()).build();
-  }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public Pipeline getContainer(String containerName) throws IOException {
-    checkAdminAccess();
-    return scmContainerManager.getContainer(containerName).getPipeline();
-  }
-
-  @VisibleForTesting
-  public ContainerInfo getContainerInfo(String containerName)
-      throws IOException {
-    return scmContainerManager.getContainer(containerName);
-  }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public List<ContainerInfo> listContainer(String startName,
-      String prefixName, int count) throws IOException {
-    return scmContainerManager.listContainer(startName, prefixName, count);
-  }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public void deleteContainer(String containerName) throws IOException {
-    checkAdminAccess();
-    scmContainerManager.deleteContainer(containerName);
-  }
-
-  /**
-   * Queries a list of Node Statuses.
-   *
-   * @param nodeStatuses
-   * @param queryScope
-   * @param poolName @return List of Datanodes.
-   */
-  @Override
-  public HddsProtos.NodePool queryNode(EnumSet<NodeState> nodeStatuses,
-      HddsProtos.QueryScope queryScope, String poolName) throws IOException {
-
-    if (queryScope == HddsProtos.QueryScope.POOL) {
-      throw new IllegalArgumentException("Not Supported yet");
-    }
-
-    List<DatanodeDetails> datanodes = queryNode(nodeStatuses);
-    HddsProtos.NodePool.Builder poolBuilder =
-        HddsProtos.NodePool.newBuilder();
-
-    for (DatanodeDetails datanode : datanodes) {
-      HddsProtos.Node node = HddsProtos.Node.newBuilder()
-          .setNodeID(datanode.getProtoBufMessage())
-          .addAllNodeStates(nodeStatuses)
-          .build();
-      poolBuilder.addNodes(node);
-    }
-
-    return poolBuilder.build();
-  }
-
-  /**
-   * Notify from client when begin/finish operation for container/pipeline
-   * objects on datanodes.
-   * @param type
-   * @param name
-   * @param op
-   * @param stage
-   */
-  @Override
-  public void notifyObjectStageChange(
-      ObjectStageChangeRequestProto.Type type, String name,
-      ObjectStageChangeRequestProto.Op op,
-      ObjectStageChangeRequestProto.Stage stage) throws IOException {
-
-    LOG.info("Object type {} name {} op {} new stage {}",
-        type, name, op, stage);
-    if (type == ObjectStageChangeRequestProto.Type.container) {
-      if (op == ObjectStageChangeRequestProto.Op.create) {
-        if (stage == ObjectStageChangeRequestProto.Stage.begin) {
-          scmContainerManager.updateContainerState(name,
-              HddsProtos.LifeCycleEvent.CREATE);
-        } else {
-          scmContainerManager.updateContainerState(name,
-              HddsProtos.LifeCycleEvent.CREATED);
-        }
-      } else if (op == ObjectStageChangeRequestProto.Op.close) {
-        if (stage == ObjectStageChangeRequestProto.Stage.begin) {
-          scmContainerManager.updateContainerState(name,
-              HddsProtos.LifeCycleEvent.FINALIZE);
-        } else {
-          scmContainerManager.updateContainerState(name,
-              HddsProtos.LifeCycleEvent.CLOSE);
-        }
-      }
-    } //else if (type == ObjectStageChangeRequestProto.Type.pipeline) {
-    // TODO: pipeline state update will be addressed in future patch.
-    //}
-  }
-
-  /**
-   * Creates a replication pipeline of a specified type.
-   */
-  @Override
-  public Pipeline createReplicationPipeline(
-      HddsProtos.ReplicationType replicationType,
-      HddsProtos.ReplicationFactor factor,
-      HddsProtos.NodePool nodePool)
-      throws IOException {
-     // TODO: will be addressed in future patch.
-    return null;
-  }
-
-  /**
-   * Queries a list of Node that match a set of statuses.
-   * <p>
-   * For example, if the nodeStatuses is HEALTHY and RAFT_MEMBER,
-   * then this call will return all healthy nodes which members in
-   * Raft pipeline.
-   * <p>
-   * Right now we don't support operations, so we assume it is an AND operation
-   * between the operators.
-   *
-   * @param nodeStatuses - A set of NodeStates.
-   * @return List of Datanodes.
-   */
-
-  public List<DatanodeDetails> queryNode(EnumSet<NodeState> nodeStatuses) {
-    Preconditions.checkNotNull(nodeStatuses, "Node Query set cannot be null");
-    Preconditions.checkState(nodeStatuses.size() > 0, "No valid arguments " +
-        "in the query set");
-    List<DatanodeDetails> resultList = new LinkedList<>();
-    Set<DatanodeDetails> currentSet = new TreeSet<>();
-
-    for (NodeState nodeState : nodeStatuses) {
-      Set<DatanodeDetails> nextSet = queryNodeState(nodeState);
-      if ((nextSet == null) || (nextSet.size() == 0)) {
-        // Right now we only support AND operation. So intersect with
-        // any empty set is null.
-        return resultList;
-      }
-      // First time we have to add all the elements, next time we have to
-      // do an intersection operation on the set.
-      if (currentSet.size() == 0) {
-        currentSet.addAll(nextSet);
-      } else {
-        currentSet.retainAll(nextSet);
-      }
-    }
-
-    resultList.addAll(currentSet);
-    return resultList;
-  }
-
-  /**
-   * Query the System for Nodes.
-   *
-   * @param nodeState - NodeState that we are interested in matching.
-   * @return Set of Datanodes that match the NodeState.
-   */
-  private Set<DatanodeDetails> queryNodeState(NodeState nodeState) {
-    if (nodeState == NodeState.RAFT_MEMBER ||
-        nodeState == NodeState.FREE_NODE) {
-      throw new IllegalStateException("Not implemented yet");
-    }
-    Set<DatanodeDetails> returnSet = new TreeSet<>();
-    List<DatanodeDetails> tmp = getScmNodeManager().getNodes(nodeState);
-    if ((tmp != null) && (tmp.size() > 0)) {
-      returnSet.addAll(tmp);
-    }
-    return returnSet;
-  }
-
-  /**
-   * Asks SCM where a container should be allocated. SCM responds with the set
-   * of datanodes that should be used creating this container.
-   *
-   * @param containerName - Name of the container.
-   * @param replicationFactor - replication factor.
-   * @return pipeline
-   * @throws IOException
-   */
-  @Override
-  public Pipeline allocateContainer(HddsProtos.ReplicationType replicationType,
-      HddsProtos.ReplicationFactor replicationFactor, String containerName,
-      String owner) throws IOException {
-
-    checkAdminAccess();
-    return scmContainerManager
-        .allocateContainer(replicationType, replicationFactor, containerName,
-            owner).getPipeline();
-  }
-
-  /**
-   * Returns listening address of StorageLocation Protocol RPC server.
-   *
-   * @return listen address of StorageLocation RPC server
-   */
-  @VisibleForTesting
-  public InetSocketAddress getClientRpcAddress() {
-    return clientRpcAddress;
-  }
-
-  @Override
-  public String getClientRpcPort() {
-    InetSocketAddress addr = getClientRpcAddress();
-    return addr == null ? "0" : Integer.toString(addr.getPort());
-  }
-
-  /**
-   * Returns listening address of StorageDatanode Protocol RPC server.
-   *
-   * @return Address where datanode are communicating.
-   */
-  public InetSocketAddress getDatanodeRpcAddress() {
-    return datanodeRpcAddress;
-  }
-
-  @Override
-  public String getDatanodeRpcPort() {
-    InetSocketAddress addr = getDatanodeRpcAddress();
-    return addr == null ? "0" : Integer.toString(addr.getPort());
-  }
-
-  /**
-   * Start service.
-   */
-  public void start() throws IOException {
-    LOG.info(buildRpcServerStartMessage(
-        "StorageContainerLocationProtocol RPC server", clientRpcAddress));
-    DefaultMetricsSystem.initialize("StorageContainerManager");
-    clientRpcServer.start();
-    LOG.info(buildRpcServerStartMessage(
-        "ScmBlockLocationProtocol RPC server", blockRpcAddress));
-    blockRpcServer.start();
-    LOG.info(buildRpcServerStartMessage("RPC server for DataNodes",
-        datanodeRpcAddress));
-    datanodeRpcServer.start();
-    httpServer.start();
-    scmBlockManager.start();
-
-    setStartTime();
-
-  }
-
-  /**
-   * Stop service.
-   */
-  public void stop() {
-    try {
-      LOG.info("Stopping block service RPC server");
-      blockRpcServer.stop();
-    } catch (Exception ex) {
-      LOG.error("Storage Container Manager blockRpcServer stop failed.", ex);
-    }
-
-    try {
-      LOG.info("Stopping the StorageContainerLocationProtocol RPC server");
-      clientRpcServer.stop();
-    } catch (Exception ex) {
-      LOG.error("Storage Container Manager clientRpcServer stop failed.", ex);
-    }
-
-    try {
-      LOG.info("Stopping the RPC server for DataNodes");
-      datanodeRpcServer.stop();
-    } catch (Exception ex) {
-      LOG.error("Storage Container Manager datanodeRpcServer stop failed.", ex);
-    }
-
-    try {
-      LOG.info("Stopping Storage Container Manager HTTP server.");
-      httpServer.stop();
-    } catch (Exception ex) {
-      LOG.error("Storage Container Manager HTTP server stop failed.", ex);
-    }
-
-    try {
-      LOG.info("Stopping Block Manager Service.");
-      scmBlockManager.stop();
-    } catch (Exception ex) {
-      LOG.error("SCM block manager service stop failed.", ex);
-    }
-
-    if (containerReportCache != null) {
-      containerReportCache.invalidateAll();
-      containerReportCache.cleanUp();
-    }
-
-    if (metrics != null) {
-      metrics.unRegister();
-    }
-
-    unregisterMXBean();
-    IOUtils.cleanupWithLogger(LOG, scmContainerManager);
-    IOUtils.cleanupWithLogger(LOG, scmNodeManager);
-  }
-
-  /**
-   * Wait until service has completed shutdown.
-   */
-  public void join() {
-    try {
-      blockRpcServer.join();
-      clientRpcServer.join();
-      datanodeRpcServer.join();
-    } catch (InterruptedException e) {
-      Thread.currentThread().interrupt();
-      LOG.info("Interrupted during StorageContainerManager join.");
-    }
-  }
-
-  /**
-   * Returns SCM version.
-   *
-   * @return Version info.
-   */
-  @Override
-  public SCMVersionResponseProto getVersion(
-      SCMVersionRequestProto versionRequest) throws IOException {
-    return getScmNodeManager().getVersion(versionRequest).getProtobufMessage();
-  }
-
-  /**
-   * Used by data node to send a Heartbeat.
-   *
-   * @param datanodeDetails - Datanode Details.
-   * @param nodeReport - Node Report
-   * @param reportState - Container report ready info.
-   * @return - SCMHeartbeatResponseProto
-   * @throws IOException
-   */
-  @Override
-  public SCMHeartbeatResponseProto sendHeartbeat(
-      DatanodeDetailsProto datanodeDetails, SCMNodeReport nodeReport,
-      ReportState reportState) throws IOException {
-    List<SCMCommand> commands =
-        getScmNodeManager().sendHeartbeat(datanodeDetails, nodeReport,
-            reportState);
-    List<SCMCommandResponseProto> cmdResponses = new LinkedList<>();
-    for (SCMCommand cmd : commands) {
-      cmdResponses.add(getCommandResponse(cmd, datanodeDetails.getUuid()
-          .toString()));
-    }
-    return SCMHeartbeatResponseProto.newBuilder().addAllCommands(cmdResponses)
-        .build();
-  }
-
-  /**
-   * Register Datanode.
-   *
-   * @param datanodeDetails - DatanodID.
-   * @param scmAddresses - List of SCMs this datanode is configured to
-   * communicate.
-   * @return SCM Command.
-   */
-  @Override
-  public StorageContainerDatanodeProtocolProtos.SCMRegisteredCmdResponseProto
-      register(DatanodeDetailsProto datanodeDetails, String[] scmAddresses) {
-    // TODO : Return the list of Nodes that forms the SCM HA.
-    return getRegisteredResponse(
-        scmNodeManager.register(datanodeDetails), null);
-  }
-
-  /**
-   * Send a container report.
-   *
-   * @param reports -- Container report
-   * @return HeartbeatRespose.nullcommand.
-   * @throws IOException
-   */
-  @Override
-  public ContainerReportsResponseProto sendContainerReport(
-      ContainerReportsRequestProto reports) throws IOException {
-    updateContainerReportMetrics(reports);
-
-    // should we process container reports async?
-    scmContainerManager.processContainerReports(reports);
-    return ContainerReportsResponseProto.newBuilder().build();
-  }
-
-  private void updateContainerReportMetrics(
-      ContainerReportsRequestProto reports) {
-    ContainerStat newStat = null;
-    // TODO: We should update the logic once incremental container report
-    // type is supported.
-    if (reports
-        .getType() == ContainerReportsRequestProto.reportType.fullReport) {
-      newStat = new ContainerStat();
-      for (StorageContainerDatanodeProtocolProtos.ContainerInfo info : reports
-          .getReportsList()) {
-        newStat.add(new ContainerStat(info.getSize(), info.getUsed(),
-            info.getKeyCount(), info.getReadBytes(), info.getWriteBytes(),
-            info.getReadCount(), info.getWriteCount()));
-      }
-
-      // update container metrics
-      metrics.setLastContainerStat(newStat);
-    }
-
-    // Update container stat entry, this will trigger a removal operation if it
-    // exists in cache.
-    synchronized (containerReportCache) {
-      String datanodeUuid = reports.getDatanodeDetails().getUuid();
-      if (datanodeUuid != null && newStat != null) {
-        containerReportCache.put(datanodeUuid, newStat);
-        // update global view container metrics
-        metrics.incrContainerStat(newStat);
-      }
-    }
-  }
-
-  /**
-   * Handles the block deletion ACKs sent by datanodes. Once ACKs recieved,
-   * SCM considers the blocks are deleted and update the metadata in SCM DB.
-   *
-   * @param acks
-   * @return
-   * @throws IOException
-   */
-  @Override
-  public ContainerBlocksDeletionACKResponseProto sendContainerBlocksDeletionACK(
-      ContainerBlocksDeletionACKProto acks) throws IOException {
-    if (acks.getResultsCount() > 0) {
-      List<DeleteBlockTransactionResult> resultList = acks.getResultsList();
-      for (DeleteBlockTransactionResult result : resultList) {
-        if (LOG.isDebugEnabled()) {
-          LOG.debug("Got block deletion ACK from datanode, TXIDs={}, "
-                  + "success={}", result.getTxID(), result.getSuccess());
-        }
-        if (result.getSuccess()) {
-          LOG.debug("Purging TXID={} from block deletion log",
-              result.getTxID());
-          this.getScmBlockManager().getDeletedBlockLog()
-              .commitTransactions(Collections.singletonList(result.getTxID()));
-        } else {
-          LOG.warn("Got failed ACK for TXID={}, prepare to resend the "
-              + "TX in next interval", result.getTxID());
-        }
-      }
-    }
-    return ContainerBlocksDeletionACKResponseProto.newBuilder()
-        .getDefaultInstanceForType();
-  }
-
-  /**
-   * Returns the Number of Datanodes that are communicating with SCM.
-   *
-   * @param nodestate Healthy, Dead etc.
-   * @return int -- count
-   */
-  public int getNodeCount(NodeState nodestate) {
-    return scmNodeManager.getNodeCount(nodestate);
-  }
-
-  /**
-   * Returns SCM container manager.
-   */
-  @VisibleForTesting
-  public Mapping getScmContainerManager() {
-    return scmContainerManager;
-  }
-
-  /**
-   * Returns node manager.
-   * @return - Node Manager
-   */
-  @VisibleForTesting
-  public NodeManager getScmNodeManager() {
-    return scmNodeManager;
-  }
-
-  @VisibleForTesting
-  public BlockManager getScmBlockManager() {
-    return scmBlockManager;
-  }
-
-  /**
-   * Get block locations.
-   * @param keys batch of block keys to retrieve.
-   * @return set of allocated blocks.
-   * @throws IOException
-   */
-  @Override
-  public Set<AllocatedBlock> getBlockLocations(final Set<String> keys)
-      throws IOException {
-    Set<AllocatedBlock> locatedBlocks = new HashSet<>();
-    for (String key: keys) {
-      Pipeline pipeline = scmBlockManager.getBlock(key);
-      AllocatedBlock block = new AllocatedBlock.Builder()
-          .setKey(key)
-          .setPipeline(pipeline).build();
-      locatedBlocks.add(block);
-    }
-    return locatedBlocks;
-  }
-
-  /**
-   * Asks SCM where a block should be allocated. SCM responds with the set of
-   * datanodes that should be used creating this block.
-   *
-   * @param size - size of the block.
-   * @param type - Replication type.
-   * @param factor
-   * @return allocated block accessing info (key, pipeline).
-   * @throws IOException
-   */
-  @Override
-  public AllocatedBlock allocateBlock(long size,
-      HddsProtos.ReplicationType type, HddsProtos.ReplicationFactor factor,
-      String owner) throws IOException {
-    return scmBlockManager.allocateBlock(size, type, factor, owner);
-  }
-
-  /**
-   * Get the clusterId and SCM Id from the version file in SCM.
-   */
-  @Override
-  public ScmInfo getScmInfo() throws IOException {
-    ScmInfo.Builder builder = new ScmInfo.Builder()
-        .setClusterId(scmStorage.getClusterID())
-        .setScmId(scmStorage.getScmId());
-    return builder.build();
-  }
-  /**
-   * Delete blocks for a set of object keys.
-   *
-   * @param keyBlocksInfoList list of block keys with object keys to delete.
-   * @return deletion results.
-   */
-  public List<DeleteBlockGroupResult> deleteKeyBlocks(
-      List<BlockGroup> keyBlocksInfoList) throws IOException {
-    LOG.info("SCM is informed by KSM to delete {} blocks",
-        keyBlocksInfoList.size());
-    List<DeleteBlockGroupResult> results = new ArrayList<>();
-    for (BlockGroup keyBlocks : keyBlocksInfoList) {
-      Result resultCode;
-      try {
-        // We delete blocks in an atomic operation to prevent getting
-        // into state like only a partial of blocks are deleted,
-        // which will leave key in an inconsistent state.
-        scmBlockManager.deleteBlocks(keyBlocks.getBlockIDList());
-        resultCode = Result.success;
-      } catch (SCMException scmEx) {
-        LOG.warn("Fail to delete block: {}", keyBlocks.getGroupID(), scmEx);
-        switch (scmEx.getResult()) {
-        case CHILL_MODE_EXCEPTION:
-          resultCode = Result.chillMode;
-          break;
-        case FAILED_TO_FIND_BLOCK:
-          resultCode = Result.errorNotFound;
-          break;
-        default:
-          resultCode = Result.unknownFailure;
-        }
-      } catch (IOException ex) {
-        LOG.warn("Fail to delete blocks for object key: {}",
-            keyBlocks.getGroupID(), ex);
-        resultCode = Result.unknownFailure;
-      }
-      List<DeleteBlockResult> blockResultList = new ArrayList<>();
-      for (String blockKey : keyBlocks.getBlockIDList()) {
-        blockResultList.add(new DeleteBlockResult(blockKey, resultCode));
-      }
-      results.add(new DeleteBlockGroupResult(keyBlocks.getGroupID(),
-          blockResultList));
-    }
-    return results;
-  }
-
-  @VisibleForTesting
-  public String getPpcRemoteUsername() {
-    UserGroupInformation user = ProtobufRpcEngine.Server.getRemoteUser();
-    return user == null ? null : user.getUserName();
-  }
-
-  private void checkAdminAccess() throws IOException {
-    String remoteUser = getPpcRemoteUsername();
-    if(remoteUser != null) {
-      if (!scmAdminUsernames.contains(remoteUser)) {
-        throw new IOException(
-            "Access denied for user " + remoteUser
-                + ". Superuser privilege is required.");
-      }
-    }
-  }
-
-  /**
-   * Initialize SCM metrics.
-   */
-  public static void initMetrics() {
-    metrics = SCMMetrics.create();
-  }
-
-  /**
-   * Return SCM metrics instance.
-   */
-  public static SCMMetrics getMetrics() {
-    return metrics == null ? SCMMetrics.create() : metrics;
-  }
-
-  /**
-   * Invalidate container stat entry for given datanode.
-   *
-   * @param datanodeUuid
-   */
-  public void removeContainerReport(String datanodeUuid) {
-    synchronized (containerReportCache) {
-      containerReportCache.invalidate(datanodeUuid);
-    }
-  }
-
-  /**
-   * Get container stat of specified datanode.
-   *
-   * @param datanodeUuid
-   * @return
-   */
-  public ContainerStat getContainerReport(String datanodeUuid) {
-    ContainerStat stat = null;
-    synchronized (containerReportCache) {
-      stat = containerReportCache.getIfPresent(datanodeUuid);
-    }
-
-    return stat;
-  }
-
-  /**
-   * Returns a view of the container stat entries. Modifications made to the
-   * map will directly affect the cache.
-   *
-   * @return
-   */
-  public ConcurrentMap<String, ContainerStat> getContainerReportCache() {
-    return containerReportCache.asMap();
-  }
-
-  @Override
-  public Map<String, String> getContainerReport() {
-    Map<String, String> id2StatMap = new HashMap<>();
-    synchronized (containerReportCache) {
-      ConcurrentMap<String, ContainerStat> map = containerReportCache.asMap();
-      for (Map.Entry<String, ContainerStat> entry : map.entrySet()) {
-        id2StatMap.put(entry.getKey(), entry.getValue().toJsonString());
-      }
-    }
-
-    return id2StatMap;
-  }
-}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0c3dc4c/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/StorageContainerManagerHttpServer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/StorageContainerManagerHttpServer.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/StorageContainerManagerHttpServer.java
deleted file mode 100644
index 1ca059c..0000000
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/StorageContainerManagerHttpServer.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with this
- * work for additional information regarding copyright ownership.  The ASF
- * licenses this file to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations under
- * the License.
- */
-
-package org.apache.hadoop.hdds.scm;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hdds.server.BaseHttpServer;
-import org.apache.hadoop.ozone.OzoneConfigKeys;
-
-import java.io.IOException;
-
-/**
- * HttpServer2 wrapper for the Ozone Storage Container Manager.
- */
-public class StorageContainerManagerHttpServer extends BaseHttpServer {
-
-  public StorageContainerManagerHttpServer(Configuration conf)
-      throws IOException {
-    super(conf, "scm");
-  }
-
-  @Override protected String getHttpAddressKey() {
-    return ScmConfigKeys.OZONE_SCM_HTTP_ADDRESS_KEY;
-  }
-
-  @Override protected String getHttpBindHostKey() {
-    return ScmConfigKeys.OZONE_SCM_HTTP_BIND_HOST_KEY;
-  }
-
-  @Override protected String getHttpsAddressKey() {
-    return ScmConfigKeys.OZONE_SCM_HTTPS_ADDRESS_KEY;
-  }
-
-  @Override protected String getHttpsBindHostKey() {
-    return ScmConfigKeys.OZONE_SCM_HTTPS_BIND_HOST_KEY;
-  }
-
-  @Override protected String getBindHostDefault() {
-    return ScmConfigKeys.OZONE_SCM_HTTP_BIND_HOST_DEFAULT;
-  }
-
-  @Override protected int getHttpBindPortDefault() {
-    return ScmConfigKeys.OZONE_SCM_HTTP_BIND_PORT_DEFAULT;
-  }
-
-  @Override protected int getHttpsBindPortDefault() {
-    return ScmConfigKeys.OZONE_SCM_HTTPS_BIND_PORT_DEFAULT;
-  }
-
-  @Override protected String getKeytabFile() {
-    return ScmConfigKeys.OZONE_SCM_KEYTAB_FILE;
-  }
-
-  @Override protected String getSpnegoPrincipal() {
-    return OzoneConfigKeys.OZONE_SCM_WEB_AUTHENTICATION_KERBEROS_PRINCIPAL;
-  }
-
-  @Override protected String getEnabledKey() {
-    return ScmConfigKeys.OZONE_SCM_HTTP_ENABLED_KEY;
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0c3dc4c/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java
index 6857b11..c72e2a1 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java
@@ -21,7 +21,7 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.hadoop.hdds.scm.HddsServerUtil;
-import org.apache.hadoop.hdds.scm.StorageContainerManager;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
 import org.apache.hadoop.hdds.scm.VersionInfo;
 import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeMetric;
 import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeStat;
@@ -849,10 +849,11 @@ public class SCMNodeManager
               .setNodeReport(nodeReport)
               .setContainerReportState(containerReportState)
               .build());
+      return commandQueue.getCommand(datanodeDetails.getUuid());
     } else {
       LOG.error("Datanode ID in heartbeat is null");
     }
-    return commandQueue.getCommand(datanodeDetails.getUuid());
+    return null;
   }
 
   /**

http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0c3dc4c/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMBlockProtocolServer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMBlockProtocolServer.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMBlockProtocolServer.java
new file mode 100644
index 0000000..e0560a1
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMBlockProtocolServer.java
@@ -0,0 +1,222 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license
+ * agreements. See the NOTICE file distributed with this work for additional
+ * information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache
+ * License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the
+ * License. You may obtain a
+ * copy of the License at
+ *
+ * <p>http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * <p>Unless required by applicable law or agreed to in writing, software
+ * distributed under the
+ * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+ * CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdds.scm.server;
+
+import com.google.protobuf.BlockingService;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos;
+import org.apache.hadoop.hdds.scm.HddsServerUtil;
+import org.apache.hadoop.hdds.scm.ScmInfo;
+import org.apache.hadoop.hdds.scm.container.common.helpers.AllocatedBlock;
+import org.apache.hadoop.hdds.scm.container.common.helpers.DeleteBlockResult;
+import org.apache.hadoop.hdds.scm.container.common.helpers.Pipeline;
+import org.apache.hadoop.hdds.scm.exceptions.SCMException;
+import org.apache.hadoop.hdds.scm.protocol.ScmBlockLocationProtocol;
+import org.apache.hadoop.hdds.scm.protocolPB.ScmBlockLocationProtocolPB;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.ozone.common.BlockGroup;
+import org.apache.hadoop.ozone.common.DeleteBlockGroupResult;
+import org.apache.hadoop.ozone.protocolPB
+    .ScmBlockLocationProtocolServerSideTranslatorPB;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+    .OZONE_SCM_BLOCK_CLIENT_ADDRESS_KEY;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+    .OZONE_SCM_HANDLER_COUNT_DEFAULT;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+    .OZONE_SCM_HANDLER_COUNT_KEY;
+import static org.apache.hadoop.hdds.server.ServerUtils.updateRPCListenAddress;
+import static org.apache.hadoop.hdds.scm.server.StorageContainerManager
+    .startRpcServer;
+
+/**
+ * SCM block protocol is the protocol used by Namenode and OzoneManager to get
+ * blocks from the SCM.
+ */
+public class SCMBlockProtocolServer implements ScmBlockLocationProtocol {
+  private static final Logger LOG =
+      LoggerFactory.getLogger(SCMBlockProtocolServer.class);
+
+  private final StorageContainerManager scm;
+  private final OzoneConfiguration conf;
+  private final RPC.Server blockRpcServer;
+  private final InetSocketAddress blockRpcAddress;
+
+  /**
+   * The RPC server that listens to requests from block service clients.
+   */
+  public SCMBlockProtocolServer(OzoneConfiguration conf,
+      StorageContainerManager scm) throws IOException {
+    this.scm = scm;
+    this.conf = conf;
+    final int handlerCount =
+        conf.getInt(OZONE_SCM_HANDLER_COUNT_KEY,
+            OZONE_SCM_HANDLER_COUNT_DEFAULT);
+
+    RPC.setProtocolEngine(conf, ScmBlockLocationProtocolPB.class,
+        ProtobufRpcEngine.class);
+    // SCM Block Service RPC
+    BlockingService blockProtoPbService =
+        ScmBlockLocationProtocolProtos.ScmBlockLocationProtocolService
+            .newReflectiveBlockingService(
+                new ScmBlockLocationProtocolServerSideTranslatorPB(this));
+
+    final InetSocketAddress scmBlockAddress = HddsServerUtil
+        .getScmBlockClientBindAddress(conf);
+    blockRpcServer =
+        startRpcServer(
+            conf,
+            scmBlockAddress,
+            ScmBlockLocationProtocolPB.class,
+            blockProtoPbService,
+            handlerCount);
+    blockRpcAddress =
+        updateRPCListenAddress(
+            conf, OZONE_SCM_BLOCK_CLIENT_ADDRESS_KEY, scmBlockAddress,
+            blockRpcServer);
+
+  }
+
+  public RPC.Server getBlockRpcServer() {
+    return blockRpcServer;
+  }
+
+  public InetSocketAddress getBlockRpcAddress() {
+    return blockRpcAddress;
+  }
+
+  public void start() {
+    LOG.info(
+        StorageContainerManager.buildRpcServerStartMessage(
+            "RPC server for Block Protocol", getBlockRpcAddress()));
+    getBlockRpcServer().start();
+  }
+
+  public void stop() {
+    try {
+      LOG.info("Stopping the RPC server for Block Protocol");
+      getBlockRpcServer().stop();
+    } catch (Exception ex) {
+      LOG.error("Block Protocol RPC stop failed.", ex);
+    }
+    IOUtils.cleanupWithLogger(LOG, scm.getScmNodeManager());
+  }
+
+  public void join() throws InterruptedException {
+    LOG.trace("Join RPC server for Block Protocol");
+    getBlockRpcServer().join();
+  }
+
+  @Override
+  public Set<AllocatedBlock> getBlockLocations(Set<String> keys) throws
+      IOException {
+    Set<AllocatedBlock> locatedBlocks = new HashSet<>();
+    for (String key : keys) {
+      Pipeline pipeline = scm.getScmBlockManager().getBlock(key);
+      AllocatedBlock block = new AllocatedBlock.Builder().setKey(key)
+          .setPipeline(pipeline).build();
+      locatedBlocks.add(block);
+    }
+    return locatedBlocks;
+
+  }
+
+  @Override
+  public AllocatedBlock allocateBlock(long size, HddsProtos.ReplicationType
+      type, HddsProtos.ReplicationFactor factor, String owner) throws
+      IOException {
+    return scm.getScmBlockManager().allocateBlock(size, type, factor, owner);
+  }
+
+  /**
+   * Delete blocks for a set of object keys.
+   *
+   * @param keyBlocksInfoList list of block keys with object keys to delete.
+   * @return deletion results.
+   */
+  @Override
+  public List<DeleteBlockGroupResult> deleteKeyBlocks(
+      List<BlockGroup> keyBlocksInfoList) throws IOException {
+    LOG.info("SCM is informed by KSM to delete {} blocks", keyBlocksInfoList
+        .size());
+    List<DeleteBlockGroupResult> results = new ArrayList<>();
+    for (BlockGroup keyBlocks : keyBlocksInfoList) {
+      ScmBlockLocationProtocolProtos.DeleteScmBlockResult.Result resultCode;
+      try {
+        // We delete blocks in an atomic operation to prevent getting
+        // into state like only a partial of blocks are deleted,
+        // which will leave key in an inconsistent state.
+        scm.getScmBlockManager().deleteBlocks(keyBlocks.getBlockIDList());
+        resultCode = ScmBlockLocationProtocolProtos.DeleteScmBlockResult
+            .Result.success;
+      } catch (SCMException scmEx) {
+        LOG.warn("Fail to delete block: {}", keyBlocks.getGroupID(), scmEx);
+        switch (scmEx.getResult()) {
+        case CHILL_MODE_EXCEPTION:
+          resultCode = ScmBlockLocationProtocolProtos.DeleteScmBlockResult
+              .Result.chillMode;
+          break;
+        case FAILED_TO_FIND_BLOCK:
+          resultCode = ScmBlockLocationProtocolProtos.DeleteScmBlockResult
+              .Result.errorNotFound;
+          break;
+        default:
+          resultCode = ScmBlockLocationProtocolProtos.DeleteScmBlockResult
+              .Result.unknownFailure;
+        }
+      } catch (IOException ex) {
+        LOG.warn("Fail to delete blocks for object key: {}", keyBlocks
+            .getGroupID(), ex);
+        resultCode = ScmBlockLocationProtocolProtos.DeleteScmBlockResult
+            .Result.unknownFailure;
+      }
+      List<DeleteBlockResult> blockResultList = new ArrayList<>();
+      for (String blockKey : keyBlocks.getBlockIDList()) {
+        blockResultList.add(new DeleteBlockResult(blockKey, resultCode));
+      }
+      results.add(new DeleteBlockGroupResult(keyBlocks.getGroupID(),
+          blockResultList));
+    }
+    return results;
+  }
+
+  @Override
+  public ScmInfo getScmInfo() throws IOException {
+    ScmInfo.Builder builder =
+        new ScmInfo.Builder()
+            .setClusterId(scm.getScmStorage().getClusterID())
+            .setScmId(scm.getScmStorage().getScmId());
+    return builder.build();
+  }
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/f0c3dc4c/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMClientProtocolServer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMClientProtocolServer.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMClientProtocolServer.java
new file mode 100644
index 0000000..42cce2f
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMClientProtocolServer.java
@@ -0,0 +1,314 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license
+ * agreements. See the NOTICE file distributed with this work for additional
+ * information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache
+ * License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the
+ * License. You may obtain a
+ * copy of the License at
+ *
+ * <p>http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * <p>Unless required by applicable law or agreed to in writing, software
+ * distributed under the
+ * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+ * CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdds.scm.server;
+
+import com.google.common.base.Preconditions;
+import com.google.protobuf.BlockingService;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.protocol.proto
+    .StorageContainerLocationProtocolProtos;
+import org.apache.hadoop.hdds.scm.HddsServerUtil;
+import org.apache.hadoop.hdds.scm.ScmInfo;
+import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerInfo;
+import org.apache.hadoop.hdds.scm.container.common.helpers.Pipeline;
+import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocol;
+import org.apache.hadoop.hdds.scm.protocolPB.StorageContainerLocationProtocolPB;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.ozone.protocolPB
+    .StorageContainerLocationProtocolServerSideTranslatorPB;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.EnumSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+
+import static org.apache.hadoop.hdds.protocol.proto
+    .StorageContainerLocationProtocolProtos
+    .StorageContainerLocationProtocolService.newReflectiveBlockingService;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+    .OZONE_SCM_CLIENT_ADDRESS_KEY;
+
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+    .OZONE_SCM_HANDLER_COUNT_DEFAULT;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+    .OZONE_SCM_HANDLER_COUNT_KEY;
+import static org.apache.hadoop.hdds.server.ServerUtils.updateRPCListenAddress;
+import static org.apache.hadoop.hdds.scm.server.StorageContainerManager
+    .startRpcServer;
+
+/**
+ * The RPC server that listens to requests from clients.
+ */
+public class SCMClientProtocolServer implements
+    StorageContainerLocationProtocol {
+  private static final Logger LOG =
+      LoggerFactory.getLogger(SCMClientProtocolServer.class);
+  private final RPC.Server clientRpcServer;
+  private final InetSocketAddress clientRpcAddress;
+  private final StorageContainerManager scm;
+  private final OzoneConfiguration conf;
+
+  public SCMClientProtocolServer(OzoneConfiguration conf,
+      StorageContainerManager scm) throws IOException {
+    this.scm = scm;
+    this.conf = conf;
+    final int handlerCount =
+        conf.getInt(OZONE_SCM_HANDLER_COUNT_KEY,
+            OZONE_SCM_HANDLER_COUNT_DEFAULT);
+    RPC.setProtocolEngine(conf, StorageContainerLocationProtocolPB.class,
+        ProtobufRpcEngine.class);
+
+    // SCM Container Service RPC
+    BlockingService storageProtoPbService =
+        newReflectiveBlockingService(
+            new StorageContainerLocationProtocolServerSideTranslatorPB(this));
+
+    final InetSocketAddress scmAddress = HddsServerUtil
+        .getScmClientBindAddress(conf);
+    clientRpcServer =
+        startRpcServer(
+            conf,
+            scmAddress,
+            StorageContainerLocationProtocolPB.class,
+            storageProtoPbService,
+            handlerCount);
+    clientRpcAddress =
+        updateRPCListenAddress(conf, OZONE_SCM_CLIENT_ADDRESS_KEY,
+            scmAddress, clientRpcServer);
+
+  }
+
+  public RPC.Server getClientRpcServer() {
+    return clientRpcServer;
+  }
+
+  public InetSocketAddress getClientRpcAddress() {
+    return clientRpcAddress;
+  }
+
+  public void start() {
+    LOG.info(
+        StorageContainerManager.buildRpcServerStartMessage(
+            "RPC server for Client ", getClientRpcAddress()));
+    getClientRpcServer().start();
+  }
+
+  public void stop() {
+    try {
+      LOG.info("Stopping the RPC server for Client Protocol");
+      getClientRpcServer().stop();
+    } catch (Exception ex) {
+      LOG.error("Client Protocol RPC stop failed.", ex);
+    }
+    IOUtils.cleanupWithLogger(LOG, scm.getScmNodeManager());
+  }
+
+  public void join() throws InterruptedException {
+    LOG.trace("Join RPC server for Client Protocol");
+    getClientRpcServer().join();
+  }
+
+  @Override
+  public Pipeline allocateContainer(HddsProtos.ReplicationType
+      replicationType, HddsProtos.ReplicationFactor factor, String
+      containerName, String owner) throws IOException {
+    scm.checkAdminAccess();
+    return scm.getScmContainerManager()
+        .allocateContainer(replicationType, factor, containerName, owner)
+        .getPipeline();
+  }
+
+  @Override
+  public Pipeline getContainer(String containerName) throws IOException {
+    return scm.getScmContainerManager()
+        .getContainer(containerName).getPipeline();
+  }
+
+  @Override
+  public List<ContainerInfo> listContainer(String startName,
+      String prefixName, int count) throws IOException {
+    return scm.getScmContainerManager()
+        .listContainer(startName, prefixName, count);
+  }
+
+  @Override
+  public void deleteContainer(String containerName) throws IOException {
+    scm.checkAdminAccess();
+    scm.getScmContainerManager().deleteContainer(containerName);
+
+  }
+
+  @Override
+  public HddsProtos.NodePool queryNode(EnumSet<HddsProtos.NodeState>
+      nodeStatuses, HddsProtos.QueryScope queryScope, String poolName) throws
+      IOException {
+
+    if (queryScope == HddsProtos.QueryScope.POOL) {
+      throw new IllegalArgumentException("Not Supported yet");
+    }
+
+    List<DatanodeDetails> datanodes = queryNode(nodeStatuses);
+    HddsProtos.NodePool.Builder poolBuilder = HddsProtos.NodePool.newBuilder();
+
+    for (DatanodeDetails datanode : datanodes) {
+      HddsProtos.Node node =
+          HddsProtos.Node.newBuilder()
+              .setNodeID(datanode.getProtoBufMessage())
+              .addAllNodeStates(nodeStatuses)
+              .build();
+      poolBuilder.addNodes(node);
+    }
+
+    return poolBuilder.build();
+
+  }
+
+  @Override
+  public void notifyObjectStageChange(StorageContainerLocationProtocolProtos
+      .ObjectStageChangeRequestProto.Type type, String name,
+      StorageContainerLocationProtocolProtos.ObjectStageChangeRequestProto.Op
+          op, StorageContainerLocationProtocolProtos
+      .ObjectStageChangeRequestProto.Stage stage) throws IOException {
+
+    LOG.info("Object type {} name {} op {} new stage {}", type, name, op,
+        stage);
+    if (type == StorageContainerLocationProtocolProtos
+        .ObjectStageChangeRequestProto.Type.container) {
+      if (op == StorageContainerLocationProtocolProtos
+          .ObjectStageChangeRequestProto.Op.create) {
+        if (stage == StorageContainerLocationProtocolProtos
+            .ObjectStageChangeRequestProto.Stage.begin) {
+          scm.getScmContainerManager().updateContainerState(name, HddsProtos
+              .LifeCycleEvent.CREATE);
+        } else {
+          scm.getScmContainerManager().updateContainerState(name, HddsProtos
+              .LifeCycleEvent.CREATED);
+        }
+      } else {
+        if (op == StorageContainerLocationProtocolProtos
+            .ObjectStageChangeRequestProto.Op.close) {
+          if (stage == StorageContainerLocationProtocolProtos
+              .ObjectStageChangeRequestProto.Stage.begin) {
+            scm.getScmContainerManager().updateContainerState(name, HddsProtos
+                .LifeCycleEvent.FINALIZE);
+          } else {
+            scm.getScmContainerManager().updateContainerState(name, HddsProtos
+                .LifeCycleEvent.CLOSE);
+          }
+        }
+      }
+    } // else if (type == ObjectStageChangeRequestProto.Type.pipeline) {
+    // TODO: pipeline state update will be addressed in future patch.
+    // }
+
+  }
+
+  @Override
+  public Pipeline createReplicationPipeline(HddsProtos.ReplicationType type,
+      HddsProtos.ReplicationFactor factor, HddsProtos.NodePool nodePool)
+      throws IOException {
+    // TODO: will be addressed in future patch.
+    // This is needed only for debugging purposes to make sure cluster is
+    // working correctly. 
+    return null;
+  }
+
+  @Override
+  public ScmInfo getScmInfo() throws IOException {
+    ScmInfo.Builder builder =
+        new ScmInfo.Builder()
+            .setClusterId(scm.getScmStorage().getClusterID())
+            .setScmId(scm.getScmStorage().getScmId());
+    return builder.build();
+  }
+
+  /**
+   * Queries a list of Node that match a set of statuses.
+   *
+   * <p>For example, if the nodeStatuses is HEALTHY and RAFT_MEMBER, then
+   * this call will return all
+   * healthy nodes which members in Raft pipeline.
+   *
+   * <p>Right now we don't support operations, so we assume it is an AND
+   * operation between the
+   * operators.
+   *
+   * @param nodeStatuses - A set of NodeStates.
+   * @return List of Datanodes.
+   */
+  public List<DatanodeDetails> queryNode(EnumSet<HddsProtos.NodeState>
+      nodeStatuses) {
+    Preconditions.checkNotNull(nodeStatuses, "Node Query set cannot be null");
+    Preconditions.checkState(nodeStatuses.size() > 0, "No valid arguments " +
+        "in the query set");
+    List<DatanodeDetails> resultList = new LinkedList<>();
+    Set<DatanodeDetails> currentSet = new TreeSet<>();
+
+    for (HddsProtos.NodeState nodeState : nodeStatuses) {
+      Set<DatanodeDetails> nextSet = queryNodeState(nodeState);
+      if ((nextSet == null) || (nextSet.size() == 0)) {
+        // Right now we only support AND operation. So intersect with
+        // any empty set is null.
+        return resultList;
+      }
+      // First time we have to add all the elements, next time we have to
+      // do an intersection operation on the set.
+      if (currentSet.size() == 0) {
+        currentSet.addAll(nextSet);
+      } else {
+        currentSet.retainAll(nextSet);
+      }
+    }
+
+    resultList.addAll(currentSet);
+    return resultList;
+  }
+
+  /**
+   * Query the System for Nodes.
+   *
+   * @param nodeState - NodeState that we are interested in matching.
+   * @return Set of Datanodes that match the NodeState.
+   */
+  private Set<DatanodeDetails> queryNodeState(HddsProtos.NodeState nodeState) {
+    if (nodeState == HddsProtos.NodeState.RAFT_MEMBER || nodeState ==
+        HddsProtos.NodeState
+        .FREE_NODE) {
+      throw new IllegalStateException("Not implemented yet");
+    }
+    Set<DatanodeDetails> returnSet = new TreeSet<>();
+    List<DatanodeDetails> tmp = scm.getScmNodeManager().getNodes(nodeState);
+    if ((tmp != null) && (tmp.size() > 0)) {
+      returnSet.addAll(tmp);
+    }
+    return returnSet;
+  }
+}


---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org


Mime
View raw message