hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aengin...@apache.org
Subject [26/29] hadoop git commit: HDDS-737. Introduce Incremental Container Report. Contributed by Nanda kumar.
Date Thu, 08 Nov 2018 18:11:38 GMT
HDDS-737. Introduce Incremental Container Report.
Contributed by Nanda kumar.


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

Branch: refs/heads/HDDS-4
Commit: c80f753b0e95eb722a972f836c1e4d16fd823434
Parents: e1bbf7d
Author: Nanda kumar <nanda@apache.org>
Authored: Thu Nov 8 18:33:38 2018 +0530
Committer: Nanda kumar <nanda@apache.org>
Committed: Thu Nov 8 18:33:38 2018 +0530

----------------------------------------------------------------------
 .../hdds/scm/container/ContainerInfo.java       |   8 +
 .../statemachine/DatanodeStateMachine.java      |  17 +-
 .../common/statemachine/StateContext.java       |  13 +-
 .../CloseContainerCommandHandler.java           |  73 +++--
 .../states/endpoint/HeartbeatEndpointTask.java  |  12 +-
 .../StorageContainerDatanodeProtocol.proto      |   2 +-
 .../scm/command/CommandStatusReportHandler.java |   4 -
 .../container/CloseContainerEventHandler.java   |  27 --
 .../scm/container/CloseContainerWatcher.java    | 101 -------
 .../hdds/scm/container/ContainerManager.java    |  11 -
 .../hdds/scm/container/ContainerReplica.java    |  30 +-
 .../scm/container/ContainerReportHandler.java   | 202 +++++++++----
 .../IncrementalContainerReportHandler.java      |  98 +++++++
 .../hdds/scm/container/SCMContainerManager.java | 107 -------
 .../scm/container/states/ContainerStateMap.java |   7 -
 .../hadoop/hdds/scm/events/SCMEvents.java       |  34 +--
 .../hadoop/hdds/scm/node/DeadNodeHandler.java   |  37 ++-
 .../hadoop/hdds/scm/node/NewNodeHandler.java    |  16 +-
 .../hadoop/hdds/scm/node/NodeManager.java       |  50 +---
 .../hadoop/hdds/scm/node/NodeReportHandler.java |   2 +-
 .../hadoop/hdds/scm/node/NodeStateManager.java  |  69 +----
 .../hadoop/hdds/scm/node/SCMNodeManager.java    |  66 +----
 .../hadoop/hdds/scm/node/StaleNodeHandler.java  |   2 +-
 .../hdds/scm/node/states/NodeStateMap.java      |  67 ++---
 .../server/SCMDatanodeHeartbeatDispatcher.java  |  17 +-
 .../scm/server/SCMDatanodeProtocolServer.java   |   6 +-
 .../scm/server/StorageContainerManager.java     |  21 +-
 .../command/TestCommandStatusReportHandler.java |  14 -
 .../hdds/scm/container/MockNodeManager.java     |  66 ++---
 .../container/TestContainerReportHandler.java   |  15 +-
 .../container/TestContainerStateManager.java    |   3 +
 .../scm/container/TestSCMContainerManager.java  | 107 +------
 .../replication/TestReplicationManager.java     |  12 +-
 .../hdds/scm/node/TestDeadNodeHandler.java      |  24 +-
 .../container/TestCloseContainerWatcher.java    | 289 -------------------
 .../ozone/container/common/TestEndPoint.java    |   6 +-
 .../testutils/ReplicationNodeManagerMock.java   |  55 +---
 .../TestContainerStateManagerIntegration.java   |  11 +-
 .../hdds/scm/pipeline/TestNode2PipelineMap.java |   4 +-
 .../hdds/scm/pipeline/TestPipelineClose.java    |   4 +-
 .../commandhandler/TestBlockDeletion.java       |   6 +-
 41 files changed, 558 insertions(+), 1157 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerInfo.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerInfo.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerInfo.java
index 5a9484a..edfa0f9 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerInfo.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerInfo.java
@@ -158,10 +158,18 @@ public class ContainerInfo implements Comparator<ContainerInfo>,
     return usedBytes;
   }
 
+  public void setUsedBytes(long value) {
+    usedBytes = value;
+  }
+
   public long getNumberOfKeys() {
     return numberOfKeys;
   }
 
+  public void setNumberOfKeys(long value) {
+    numberOfKeys = value;
+  }
+
   public long getDeleteTransactionId() {
     return deleteTransactionId;
   }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java
index 4768cf8..12c33ff 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java
@@ -168,10 +168,14 @@ public class DatanodeStateMachine implements Closeable {
             TimeUnit.MILLISECONDS);
         now = Time.monotonicNow();
         if (now < nextHB.get()) {
-          Thread.sleep(nextHB.get() - now);
+          if(!Thread.interrupted()) {
+            Thread.sleep(nextHB.get() - now);
+          }
         }
       } catch (InterruptedException e) {
-        // Ignore this exception.
+        // Some one has sent interrupt signal, this could be because
+        // 1. Trigger heartbeat immediately
+        // 2. Shutdown has be initiated.
       } catch (Exception e) {
         LOG.error("Unable to finish the execution.", e);
       }
@@ -325,6 +329,15 @@ public class DatanodeStateMachine implements Closeable {
   }
 
   /**
+   * Calling this will immediately trigger a heartbeat to the SCMs.
+   * This heartbeat will also include all the reports which are ready to
+   * be sent by datanode.
+   */
+  public void triggerHeartbeat() {
+    stateMachineThread.interrupt();
+  }
+
+  /**
    * Waits for DatanodeStateMachine to exit.
    *
    * @throws InterruptedException

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java
index 12c196b..e928824 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java
@@ -447,6 +447,11 @@ public class StateContext {
    * @param cmd - {@link SCMCommand}.
    */
   public void addCmdStatus(SCMCommand cmd) {
+    if (cmd.getType().equals(Type.closeContainerCommand)) {
+      // We will be removing CommandStatus completely.
+      // As a first step, removed it for CloseContainerCommand.
+      return;
+    }
     CommandStatusBuilder statusBuilder;
     if (cmd.getType() == Type.deleteBlocksCommand) {
       statusBuilder = new DeleteBlockCommandStatusBuilder();
@@ -469,14 +474,6 @@ public class StateContext {
   }
 
   /**
-   * Remove object from cache in StateContext#cmdStatusMap.
-   *
-   */
-  public void removeCommandStatus(Long cmdId) {
-    cmdStatusMap.remove(cmdId);
-  }
-
-  /**
    * Updates status of a pending status command.
    * @param cmdId       command id
    * @param cmdStatusUpdater Consumer to update command status.

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/CloseContainerCommandHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/CloseContainerCommandHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/CloseContainerCommandHandler.java
index 2c3db61..591fe56 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/CloseContainerCommandHandler.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/CloseContainerCommandHandler.java
@@ -19,6 +19,8 @@ package org.apache.hadoop.ozone.container.common.statemachine.commandhandler;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.protocol.proto
+    .StorageContainerDatanodeProtocolProtos.IncrementalContainerReportProto;
+import org.apache.hadoop.hdds.protocol.proto
     .StorageContainerDatanodeProtocolProtos.SCMCommandProto;
 import org.apache.hadoop.hdds.protocol.proto
     .StorageContainerDatanodeProtocolProtos.CloseContainerCommandProto;
@@ -63,62 +65,55 @@ public class CloseContainerCommandHandler implements CommandHandler {
       StateContext context, SCMConnectionManager connectionManager) {
     LOG.debug("Processing Close Container command.");
     invocationCount++;
-    cmdExecuted = false;
     long startTime = Time.monotonicNow();
     // TODO: define this as INVALID_CONTAINER_ID in HddsConsts.java (TBA)
     long containerID = -1;
     try {
-
-      CloseContainerCommandProto
-          closeContainerProto =
-          CloseContainerCommandProto
-              .parseFrom(command.getProtoBufMessage());
+      CloseContainerCommandProto closeContainerProto =
+          CloseContainerCommandProto.parseFrom(command.getProtoBufMessage());
       containerID = closeContainerProto.getContainerID();
-      if (container.getContainerSet().getContainer(containerID)
+      // CloseContainer operation is idempotent, if the container is already
+      // closed, then do nothing.
+      if (!container.getContainerSet().getContainer(containerID)
           .getContainerData().isClosed()) {
-        LOG.debug("Container {} is already closed", containerID);
-        // It might happen that the where the first attempt of closing the
-        // container failed with NOT_LEADER_EXCEPTION. In such cases, SCM will
-        // retry to check the container got really closed via Ratis.
-        // In such cases of the retry attempt, if the container is already
-        // closed via Ratis, we should just return.
-        cmdExecuted = true;
-        return;
-      }
-      HddsProtos.PipelineID pipelineID = closeContainerProto.getPipelineID();
-      HddsProtos.ReplicationType replicationType =
-          closeContainerProto.getReplicationType();
+        LOG.debug("Closing container {}.", containerID);
+        HddsProtos.PipelineID pipelineID = closeContainerProto.getPipelineID();
+        HddsProtos.ReplicationType replicationType =
+            closeContainerProto.getReplicationType();
 
-      ContainerProtos.ContainerCommandRequestProto.Builder request =
-          ContainerProtos.ContainerCommandRequestProto.newBuilder();
-      request.setCmdType(ContainerProtos.Type.CloseContainer);
-      request.setContainerID(containerID);
-      request.setCloseContainer(
-          ContainerProtos.CloseContainerRequestProto.getDefaultInstance());
-      request.setTraceID(UUID.randomUUID().toString());
-      request.setDatanodeUuid(
-          context.getParent().getDatanodeDetails().getUuidString());
-      // submit the close container request for the XceiverServer to handle
-      container.submitContainerRequest(
-          request.build(), replicationType, pipelineID);
+        ContainerProtos.ContainerCommandRequestProto.Builder request =
+            ContainerProtos.ContainerCommandRequestProto.newBuilder();
+        request.setCmdType(ContainerProtos.Type.CloseContainer);
+        request.setContainerID(containerID);
+        request.setCloseContainer(
+            ContainerProtos.CloseContainerRequestProto.getDefaultInstance());
+        request.setTraceID(UUID.randomUUID().toString());
+        request.setDatanodeUuid(
+            context.getParent().getDatanodeDetails().getUuidString());
+        // submit the close container request for the XceiverServer to handle
+        container.submitContainerRequest(
+            request.build(), replicationType, pipelineID);
+        // Since the container is closed, we trigger an ICR
+        IncrementalContainerReportProto icr = IncrementalContainerReportProto
+            .newBuilder()
+            .addReport(container.getContainerSet()
+                .getContainer(containerID).getContainerReport())
+            .build();
+        context.addReport(icr);
+        context.getParent().triggerHeartbeat();
+      }
     } catch (Exception e) {
       if (e instanceof NotLeaderException) {
         // If the particular datanode is not the Ratis leader, the close
         // container command will not be executed by the follower but will be
         // executed by Ratis stateMachine transactions via leader to follower.
         // There can also be case where the datanode is in candidate state.
-        // In these situations, NotLeaderException is thrown. Remove the status
-        // from cmdStatus Map here so that it will be retried only by SCM if the
-        // leader could not not close the container after a certain time.
-        context.removeCommandStatus(containerID);
-        LOG.info(e.getLocalizedMessage());
+        // In these situations, NotLeaderException is thrown.
+        LOG.info("Follower cannot close the container {}.", containerID);
       } else {
         LOG.error("Can't close container " + containerID, e);
-        cmdExecuted = false;
       }
     } finally {
-      updateCommandStatus(context, command,
-          (cmdStatus) -> cmdStatus.setStatus(cmdExecuted), LOG);
       long endTime = Time.monotonicNow();
       totalTime += endTime - startTime;
     }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java
index 4fd72ec..0c0f1af 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java
@@ -141,7 +141,9 @@ public class HeartbeatEndpointTask
       rpcEndpoint.zeroMissedCount();
     } catch (IOException ex) {
       // put back the reports which failed to be sent
-      putBackReports(requestBuilder);
+      if (requestBuilder != null) {
+        putBackReports(requestBuilder);
+      }
       rpcEndpoint.logIfNeeded(ex);
     } finally {
       rpcEndpoint.unlock();
@@ -159,10 +161,10 @@ public class HeartbeatEndpointTask
       reports.add(requestBuilder.getNodeReport());
     }
     if (requestBuilder.getCommandStatusReportsCount() != 0) {
-      for (GeneratedMessage msg : requestBuilder
-          .getCommandStatusReportsList()) {
-        reports.add(msg);
-      }
+      reports.addAll(requestBuilder.getCommandStatusReportsList());
+    }
+    if (requestBuilder.getIncrementalContainerReportCount() != 0) {
+      reports.addAll(requestBuilder.getIncrementalContainerReportList());
     }
     context.putBackReports(reports);
   }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/container-service/src/main/proto/StorageContainerDatanodeProtocol.proto
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/proto/StorageContainerDatanodeProtocol.proto b/hadoop-hdds/container-service/src/main/proto/StorageContainerDatanodeProtocol.proto
index 4ddb7b2..33ea307 100644
--- a/hadoop-hdds/container-service/src/main/proto/StorageContainerDatanodeProtocol.proto
+++ b/hadoop-hdds/container-service/src/main/proto/StorageContainerDatanodeProtocol.proto
@@ -80,7 +80,7 @@ message SCMHeartbeatRequestProto {
   required DatanodeDetailsProto datanodeDetails = 1;
   optional NodeReportProto nodeReport = 2;
   optional ContainerReportsProto containerReport = 3;
-  optional IncrementalContainerReportProto incrementalContainerReport = 4;
+  repeated IncrementalContainerReportProto incrementalContainerReport = 4;
   repeated CommandStatusReportsProto commandStatusReports = 5;
   optional ContainerActionsProto containerActions = 6;
   optional PipelineActionsProto pipelineActions = 7;

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/command/CommandStatusReportHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/command/CommandStatusReportHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/command/CommandStatusReportHandler.java
index c0de382..53dfc5a 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/command/CommandStatusReportHandler.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/command/CommandStatusReportHandler.java
@@ -58,10 +58,6 @@ public class CommandStatusReportHandler implements
         publisher.fireEvent(SCMEvents.REPLICATION_STATUS, new
             ReplicationStatus(cmdStatus));
         break;
-      case closeContainerCommand:
-        publisher.fireEvent(SCMEvents.CLOSE_CONTAINER_STATUS, new
-            CloseContainerStatus(cmdStatus));
-        break;
       case deleteBlocksCommand:
         if (cmdStatus.getStatus() == CommandStatus.Status.EXECUTED) {
           publisher.fireEvent(SCMEvents.DELETE_BLOCK_STATUS,

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/CloseContainerEventHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/CloseContainerEventHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/CloseContainerEventHandler.java
index 69574a9..9796a96 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/CloseContainerEventHandler.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/CloseContainerEventHandler.java
@@ -22,14 +22,12 @@ import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.server.events.EventHandler;
 import org.apache.hadoop.hdds.server.events.EventPublisher;
-import org.apache.hadoop.hdds.server.events.IdentifiableEventPayload;
 import org.apache.hadoop.ozone.protocol.commands.CloseContainerCommand;
 import org.apache.hadoop.ozone.protocol.commands.CommandForDatanode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import static org.apache.hadoop.hdds.scm.events.SCMEvents.DATANODE_COMMAND;
-import static org.apache.hadoop.hdds.scm.events.SCMEvents.CLOSE_CONTAINER_RETRYABLE_REQ;
 
 /**
  * In case of a node failure, volume failure, volume out of spapce, node
@@ -128,32 +126,7 @@ public class CloseContainerEventHandler implements EventHandler<ContainerID> {
             new CommandForDatanode<>(node.getUuid(), closeContainerCommand))
         .forEach(command -> publisher.fireEvent(DATANODE_COMMAND, command));
 
-    publisher.fireEvent(CLOSE_CONTAINER_RETRYABLE_REQ,
-        new CloseContainerRetryableReq(containerID));
-
     LOG.trace("Issuing {} on Pipeline {} for container", closeContainerCommand,
         pipeline, containerID);
   }
-
-  /**
-   * Class to create retryable event. Prevents redundant requests for same
-   * container Id.
-   */
-  public static class CloseContainerRetryableReq implements
-      IdentifiableEventPayload {
-
-    private ContainerID containerID;
-    public CloseContainerRetryableReq(ContainerID containerID) {
-      this.containerID = containerID;
-    }
-
-    public ContainerID getContainerID() {
-      return containerID;
-    }
-
-    @Override
-    public long getId() {
-      return containerID.getId();
-    }
-  }
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/CloseContainerWatcher.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/CloseContainerWatcher.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/CloseContainerWatcher.java
deleted file mode 100644
index 4593c1f..0000000
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/CloseContainerWatcher.java
+++ /dev/null
@@ -1,101 +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>
- * <p>http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * <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.container;
-
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.CommandStatus.Status;
-import org.apache.hadoop.hdds.scm.command.CommandStatusReportHandler
-    .CloseContainerStatus;
-
-import org.apache.hadoop.hdds.scm.events.SCMEvents;
-import org.apache.hadoop.hdds.server.events.Event;
-import org.apache.hadoop.hdds.server.events.EventPublisher;
-import org.apache.hadoop.hdds.server.events.EventWatcher;
-import org.apache.hadoop.hdds.scm.container.CloseContainerEventHandler
-    .CloseContainerRetryableReq;
-import org.apache.hadoop.ozone.lease.LeaseManager;
-import org.apache.hadoop.ozone.lease.LeaseNotFoundException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-
-/**
- * This watcher will watch for CLOSE_CONTAINER_STATUS events fired from
- * CommandStatusReport. If required it will re-trigger CloseContainer command
- * for DataNodes to CloseContainerEventHandler.
- */
-public class CloseContainerWatcher extends
-    EventWatcher<CloseContainerRetryableReq, CloseContainerStatus> {
-
-  public static final Logger LOG =
-      LoggerFactory.getLogger(CloseContainerWatcher.class);
-  private final ContainerManager containerManager;
-
-  public CloseContainerWatcher(Event<CloseContainerRetryableReq> startEvent,
-      Event<CloseContainerStatus> completionEvent,
-      LeaseManager<Long> leaseManager, ContainerManager containerManager) {
-    super(startEvent, completionEvent, leaseManager);
-    this.containerManager = containerManager;
-  }
-
-  @Override
-  protected void onTimeout(EventPublisher publisher,
-      CloseContainerRetryableReq payload) {
-    // Let CloseContainerEventHandler handle this message.
-    this.resendEventToHandler(payload.getId(), publisher);
-  }
-
-  @Override
-  protected void onFinished(EventPublisher publisher,
-      CloseContainerRetryableReq payload) {
-    LOG.trace("CloseContainerCommand for containerId: {} executed ", payload
-        .getContainerID().getId());
-  }
-
-  @Override
-  protected synchronized void handleCompletion(CloseContainerStatus status,
-      EventPublisher publisher) throws LeaseNotFoundException {
-    // If status is PENDING then return without doing anything.
-    if(status.getCmdStatus().getStatus().equals(Status.PENDING)){
-      return;
-    }
-
-    CloseContainerRetryableReq closeCont = getTrackedEventbyId(status.getId());
-    super.handleCompletion(status, publisher);
-    // If status is FAILED then send a msg to Handler to resend the command.
-    if (status.getCmdStatus().getStatus().equals(Status.FAILED) && closeCont
-        != null) {
-      this.resendEventToHandler(closeCont.getId(), publisher);
-    }
-  }
-
-  private void resendEventToHandler(long containerID, EventPublisher
-      publisher) {
-    try {
-      // Check if container is still open
-      if (containerManager.getContainer(
-          ContainerID.valueof(containerID)).isOpen()) {
-        publisher.fireEvent(SCMEvents.CLOSE_CONTAINER,
-            ContainerID.valueof(containerID));
-      }
-    } catch (IOException e) {
-      LOG.warn("Error in CloseContainerWatcher while processing event " +
-          "for containerId {} ExceptionMsg: ", containerID, e.getMessage());
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerManager.java
index 0a48915..0906ca8 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerManager.java
@@ -16,14 +16,11 @@
  */
 package org.apache.hadoop.hdds.scm.container;
 
-import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
 import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.ContainerReportsProto;
 import org.apache.hadoop.hdds.scm.pipeline.PipelineNotFoundException;
 
 import java.io.Closeable;
@@ -123,14 +120,6 @@ public interface ContainerManager extends Closeable {
       HddsProtos.LifeCycleEvent event) throws IOException;
 
   /**
-   * Process container report from Datanode.
-   *
-   * @param reports Container report
-   */
-  void processContainerReports(DatanodeDetails datanodeDetails,
-      ContainerReportsProto reports) throws IOException;
-
-  /**
    * Returns the latest list of replicas for given containerId.
    *
    * @param containerID Container ID

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerReplica.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerReplica.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerReplica.java
index f2e80f4..9445fe8 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerReplica.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerReplica.java
@@ -22,6 +22,8 @@ import org.apache.commons.lang3.builder.CompareToBuilder;
 import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.commons.lang3.builder.HashCodeBuilder;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto
+    .StorageContainerDatanodeProtocolProtos.ContainerReplicaProto;
 
 import java.util.Optional;
 import java.util.UUID;
@@ -32,15 +34,18 @@ import java.util.UUID;
 public final class ContainerReplica implements Comparable<ContainerReplica> {
 
   final private ContainerID containerID;
+  final private ContainerReplicaProto.State state;
   final private DatanodeDetails datanodeDetails;
   final private UUID placeOfBirth;
 
   private Long sequenceId;
 
 
-  private ContainerReplica(ContainerID containerID, DatanodeDetails datanode,
-      UUID originNodeId) {
+  private ContainerReplica(final ContainerID containerID,
+      final ContainerReplicaProto.State state, final DatanodeDetails datanode,
+      final UUID originNodeId) {
     this.containerID = containerID;
+    this.state = state;
     this.datanodeDetails = datanode;
     this.placeOfBirth = originNodeId;
   }
@@ -68,6 +73,15 @@ public final class ContainerReplica implements Comparable<ContainerReplica> {
   }
 
   /**
+   * Returns the state of this replica.
+   *
+   * @return replica state
+   */
+  public ContainerReplicaProto.State getState() {
+    return state;
+  }
+
+  /**
    * Returns the Sequence Id of this replica.
    *
    * @return Sequence Id
@@ -126,6 +140,7 @@ public final class ContainerReplica implements Comparable<ContainerReplica> {
   public static class ContainerReplicaBuilder {
 
     private ContainerID containerID;
+    private ContainerReplicaProto.State state;
     private DatanodeDetails datanode;
     private UUID placeOfBirth;
     private Long sequenceId;
@@ -142,6 +157,12 @@ public final class ContainerReplica implements Comparable<ContainerReplica> {
       return this;
     }
 
+    public ContainerReplicaBuilder setContainerState(
+        final ContainerReplicaProto.State  containerState) {
+      state = containerState;
+      return this;
+    }
+
     /**
      * Set DatanodeDetails.
      *
@@ -184,9 +205,12 @@ public final class ContainerReplica implements Comparable<ContainerReplica> {
     public ContainerReplica build() {
       Preconditions.checkNotNull(containerID,
           "Container Id can't be null");
+      Preconditions.checkNotNull(state,
+          "Container state can't be null");
       Preconditions.checkNotNull(datanode,
           "DatanodeDetails can't be null");
-      ContainerReplica replica = new ContainerReplica(containerID, datanode,
+      ContainerReplica replica = new ContainerReplica(
+          containerID, state, datanode,
           Optional.ofNullable(placeOfBirth).orElse(datanode.getUuid()));
       Optional.ofNullable(sequenceId).ifPresent(replica::setSequenceId);
       return replica;

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerReportHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerReportHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerReportHandler.java
index 0cb2f81..c9fe9e4 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerReportHandler.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerReportHandler.java
@@ -17,19 +17,26 @@
  */
 package org.apache.hadoop.hdds.scm.container;
 
-import java.io.IOException;
+import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
 
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
-import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos;
-import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReportsProto;
-import org.apache.hadoop.hdds.scm.container.replication.ReplicationActivityStatus;
+import org.apache.hadoop.hdds.protocol.proto
+    .StorageContainerDatanodeProtocolProtos.ContainerReplicaProto;
+import org.apache.hadoop.hdds.protocol.proto
+    .StorageContainerDatanodeProtocolProtos.ContainerReportsProto;
+import org.apache.hadoop.hdds.scm.block.PendingDeleteStatusList;
+import org.apache.hadoop.hdds.scm.container.replication
+    .ReplicationActivityStatus;
 import org.apache.hadoop.hdds.scm.container.replication.ReplicationRequest;
 import org.apache.hadoop.hdds.scm.events.SCMEvents;
 import org.apache.hadoop.hdds.scm.node.NodeManager;
-import org.apache.hadoop.hdds.scm.node.states.ReportResult;
-import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher.ContainerReportFromDatanode;
+import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineManager;
+import org.apache.hadoop.hdds.scm.server
+    .SCMDatanodeHeartbeatDispatcher.ContainerReportFromDatanode;
 import org.apache.hadoop.hdds.server.events.EventHandler;
 import org.apache.hadoop.hdds.server.events.EventPublisher;
 
@@ -47,73 +54,160 @@ public class ContainerReportHandler implements
       LoggerFactory.getLogger(ContainerReportHandler.class);
 
   private final NodeManager nodeManager;
+  private final PipelineManager pipelineManager;
   private final ContainerManager containerManager;
-  private ReplicationActivityStatus replicationStatus;
+  private final ReplicationActivityStatus replicationStatus;
 
-  public ContainerReportHandler(ContainerManager containerManager,
-      NodeManager nodeManager,
-      ReplicationActivityStatus replicationActivityStatus) {
-    Preconditions.checkNotNull(containerManager);
+  public ContainerReportHandler(final NodeManager nodeManager,
+      final PipelineManager pipelineManager,
+      final ContainerManager containerManager,
+      final ReplicationActivityStatus replicationActivityStatus) {
     Preconditions.checkNotNull(nodeManager);
+    Preconditions.checkNotNull(pipelineManager);
+    Preconditions.checkNotNull(containerManager);
     Preconditions.checkNotNull(replicationActivityStatus);
     this.nodeManager = nodeManager;
+    this.pipelineManager = pipelineManager;
     this.containerManager = containerManager;
     this.replicationStatus = replicationActivityStatus;
   }
 
   @Override
-  public void onMessage(ContainerReportFromDatanode containerReportFromDatanode,
-      EventPublisher publisher) {
+  public void onMessage(final ContainerReportFromDatanode reportFromDatanode,
+      final EventPublisher publisher) {
+
+    final DatanodeDetails datanodeDetails =
+        reportFromDatanode.getDatanodeDetails();
 
-    DatanodeDetails datanodeOrigin =
-        containerReportFromDatanode.getDatanodeDetails();
+    final ContainerReportsProto containerReport =
+        reportFromDatanode.getReport();
 
-    ContainerReportsProto containerReport =
-        containerReportFromDatanode.getReport();
     try {
 
-      //update state in container db and trigger close container events
-      containerManager
-          .processContainerReports(datanodeOrigin, containerReport);
-
-      Set<ContainerID> containerIds = containerReport.getReportsList().stream()
-          .map(StorageContainerDatanodeProtocolProtos
-              .ContainerReplicaProto::getContainerID)
-          .map(ContainerID::new)
-          .collect(Collectors.toSet());
-
-      ReportResult<ContainerID> reportResult = nodeManager
-          .processContainerReport(datanodeOrigin.getUuid(), containerIds);
-
-      //we have the report, so we can update the states for the next iteration.
-      nodeManager
-          .setContainersForDatanode(datanodeOrigin.getUuid(), containerIds);
-
-      for (ContainerID containerID : reportResult.getMissingEntries()) {
-        final ContainerReplica replica = ContainerReplica.newBuilder()
-            .setContainerID(containerID)
-            .setDatanodeDetails(datanodeOrigin)
-            .build();
-        containerManager
-            .removeContainerReplica(containerID, replica);
-        checkReplicationState(containerID, publisher);
+      final List<ContainerReplicaProto> replicas = containerReport
+          .getReportsList();
+
+      // ContainerIDs which SCM expects this datanode to have.
+      final Set<ContainerID> expectedContainerIDs = nodeManager
+          .getContainers(datanodeDetails);
+
+      // ContainerIDs that this datanode actually has.
+      final Set<ContainerID> actualContainerIDs = replicas.parallelStream()
+          .map(ContainerReplicaProto::getContainerID)
+          .map(ContainerID::valueof).collect(Collectors.toSet());
+
+      // Container replicas which SCM is not aware of.
+      final  Set<ContainerID> newReplicas =
+          new HashSet<>(actualContainerIDs);
+      newReplicas.removeAll(expectedContainerIDs);
+
+      // Container replicas which are missing from datanode.
+      final Set<ContainerID> missingReplicas =
+          new HashSet<>(expectedContainerIDs);
+      missingReplicas.removeAll(actualContainerIDs);
+
+      processContainerReplicas(datanodeDetails, replicas, publisher);
+
+      // Remove missing replica from ContainerManager
+      for (ContainerID id : missingReplicas) {
+        try {
+          containerManager.getContainerReplicas(id)
+              .stream()
+              .filter(replica ->
+                  replica.getDatanodeDetails().equals(datanodeDetails))
+              .findFirst()
+              .ifPresent(replica -> {
+                try {
+                  containerManager.removeContainerReplica(id, replica);
+                } catch (ContainerNotFoundException |
+                    ContainerReplicaNotFoundException e) {
+                  // This should not happen, but even if it happens, not an
+                  // issue
+                }
+              });
+        } catch (ContainerNotFoundException e) {
+          LOG.warn("Cannot remove container replica, container {} not found",
+              id);
+        }
       }
 
-      for (ContainerID containerID : reportResult.getNewEntries()) {
-        final ContainerReplica replica = ContainerReplica.newBuilder()
-            .setContainerID(containerID)
-            .setDatanodeDetails(datanodeOrigin)
-            .build();
-        containerManager.updateContainerReplica(containerID, replica);
-        checkReplicationState(containerID, publisher);
-      }
+      // Update the latest set of containers for this datanode in NodeManager.
+      nodeManager.setContainers(datanodeDetails, actualContainerIDs);
+
+      // Replicate if needed.
+      newReplicas.forEach(id -> checkReplicationState(id, publisher));
+      missingReplicas.forEach(id -> checkReplicationState(id, publisher));
+
+    } catch (NodeNotFoundException ex) {
+      LOG.error("Received container report from unknown datanode {}",
+          datanodeDetails);
+    }
+
+  }
 
-    } catch (IOException e) {
-      //TODO: stop all the replication?
-      LOG.error("Error on processing container report from datanode {}",
-          datanodeOrigin, e);
+  private void processContainerReplicas(final DatanodeDetails datanodeDetails,
+      final List<ContainerReplicaProto> replicas,
+      final EventPublisher publisher) {
+    final PendingDeleteStatusList pendingDeleteStatusList =
+        new PendingDeleteStatusList(datanodeDetails);
+    for (ContainerReplicaProto replicaProto : replicas) {
+      try {
+        final ContainerID containerID = ContainerID.valueof(
+            replicaProto.getContainerID());
+        final ContainerInfo containerInfo = containerManager
+            .getContainer(containerID);
+        updateContainerState(datanodeDetails, containerInfo,
+            replicaProto, publisher);
+        if (containerInfo.getDeleteTransactionId() >
+            replicaProto.getDeleteTransactionId()) {
+          pendingDeleteStatusList
+              .addPendingDeleteStatus(replicaProto.getDeleteTransactionId(),
+                  containerInfo.getDeleteTransactionId(),
+                  containerInfo.getContainerID());
+        }
+      } catch (ContainerNotFoundException e) {
+        LOG.error("Received container report for an unknown container {}",
+            replicaProto.getContainerID());
+      }
     }
+    if (pendingDeleteStatusList.getNumPendingDeletes() > 0) {
+      publisher.fireEvent(SCMEvents.PENDING_DELETE_STATUS,
+          pendingDeleteStatusList);
+    }
+  }
 
+  private void updateContainerState(final DatanodeDetails datanodeDetails,
+      final ContainerInfo containerInfo,
+      final ContainerReplicaProto replicaProto,
+      final EventPublisher publisher)
+      throws ContainerNotFoundException {
+
+    final ContainerID id = containerInfo.containerID();
+    final ContainerReplica datanodeContainerReplica = ContainerReplica
+        .newBuilder()
+        .setContainerID(id)
+        .setContainerState(replicaProto.getState())
+        .setDatanodeDetails(datanodeDetails)
+        .build();
+    // TODO: Add bcsid and origin datanode to replica.
+
+    final ContainerReplica scmContainerReplica = containerManager
+        .getContainerReplicas(id)
+        .stream()
+        .filter(replica ->
+            replica.getDatanodeDetails().equals(datanodeDetails))
+        .findFirst().orElse(null);
+
+    // This is an in-memory update.
+    containerManager.updateContainerReplica(id, datanodeContainerReplica);
+    containerInfo.setUsedBytes(replicaProto.getUsed());
+    containerInfo.setNumberOfKeys(replicaProto.getKeyCount());
+
+    // Check if there is state change in container replica.
+    if (scmContainerReplica == null ||
+        scmContainerReplica.getState() != datanodeContainerReplica.getState()) {
+      //TODO: Handler replica state change.
+    }
   }
 
   private void checkReplicationState(ContainerID containerID,

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/IncrementalContainerReportHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/IncrementalContainerReportHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/IncrementalContainerReportHandler.java
new file mode 100644
index 0000000..e07ee73
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/IncrementalContainerReportHandler.java
@@ -0,0 +1,98 @@
+/**
+ * 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.container;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.protocol.proto
+    .StorageContainerDatanodeProtocolProtos.ContainerReplicaProto;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineManager;
+import org.apache.hadoop.hdds.scm.server
+    .SCMDatanodeHeartbeatDispatcher.IncrementalContainerReportFromDatanode;
+import org.apache.hadoop.hdds.server.events.EventHandler;
+import org.apache.hadoop.hdds.server.events.EventPublisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * Handles incremental container reports from datanode.
+ */
+public class IncrementalContainerReportHandler implements
+    EventHandler<IncrementalContainerReportFromDatanode> {
+
+  private static final Logger LOG =
+      LoggerFactory.getLogger(IncrementalContainerReportHandler.class);
+
+  private final PipelineManager pipelineManager;
+  private final ContainerManager containerManager;
+
+  public IncrementalContainerReportHandler(
+      final PipelineManager pipelineManager,
+      final ContainerManager containerManager)  {
+    Preconditions.checkNotNull(pipelineManager);
+    Preconditions.checkNotNull(containerManager);
+    this.pipelineManager = pipelineManager;
+    this.containerManager = containerManager;
+  }
+
+  @Override
+  public void onMessage(
+      final IncrementalContainerReportFromDatanode containerReportFromDatanode,
+      final EventPublisher publisher) {
+
+    for (ContainerReplicaProto replicaProto :
+         containerReportFromDatanode.getReport().getReportList()) {
+      try {
+        final DatanodeDetails datanodeDetails = containerReportFromDatanode
+            .getDatanodeDetails();
+        final ContainerID containerID = ContainerID
+            .valueof(replicaProto.getContainerID());
+        final ContainerInfo containerInfo = containerManager
+            .getContainer(containerID);
+
+        ContainerReplica replica = ContainerReplica.newBuilder()
+            .setContainerID(ContainerID.valueof(replicaProto.getContainerID()))
+            .setContainerState(replicaProto.getState())
+            .setDatanodeDetails(datanodeDetails)
+            .build();
+
+        containerManager.updateContainerReplica(containerID, replica);
+
+        // Check if the state of the container is changed.
+        if (replicaProto.getState() == ContainerReplicaProto.State.CLOSED &&
+            containerInfo.getState() == HddsProtos.LifeCycleState.CLOSING) {
+          containerManager.updateContainerState(containerID,
+              HddsProtos.LifeCycleEvent.CLOSE);
+        }
+
+        // TODO: Handler replica state change
+
+      } catch (ContainerNotFoundException e) {
+        LOG.warn("Container {} not found!", replicaProto.getContainerID());
+      } catch (IOException e) {
+        LOG.error("Exception while processing ICR for container {}",
+            replicaProto.getContainerID());
+      }
+    }
+
+  }
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/SCMContainerManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/SCMContainerManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/SCMContainerManager.java
index 0ea749f..86f1f9c 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/SCMContainerManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/SCMContainerManager.java
@@ -23,13 +23,10 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.StorageUnit;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ContainerInfoProto;
-import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleEvent;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
-import org.apache.hadoop.hdds.scm.block.PendingDeleteStatusList;
 import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
-import org.apache.hadoop.hdds.scm.events.SCMEvents;
 import org.apache.hadoop.hdds.scm.exceptions.SCMException;
 import org.apache.hadoop.hdds.scm.node.NodeManager;
 import org.apache.hadoop.hdds.scm.pipeline.PipelineManager;
@@ -37,10 +34,6 @@ import org.apache.hadoop.hdds.scm.pipeline.PipelineNotFoundException;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.ContainerReplicaProto;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.ContainerReportsProto;
 import org.apache.hadoop.hdds.server.ServerUtils;
 import org.apache.hadoop.hdds.server.events.EventPublisher;
 import org.apache.hadoop.ozone.OzoneConsts;
@@ -489,106 +482,6 @@ public class SCMContainerManager implements ContainerManager {
   }
 
   /**
-   * Process container report from Datanode.
-   * <p>
-   * Processing follows a very simple logic for time being.
-   * <p>
-   * 1. Datanodes report the current State -- denoted by the datanodeState
-   * <p>
-   * 2. We are the older SCM state from the Database -- denoted by
-   * the knownState.
-   * <p>
-   * 3. We copy the usage etc. from currentState to newState and log that
-   * newState to the DB. This allows us SCM to bootup again and read the
-   * state of the world from the DB, and then reconcile the state from
-   * container reports, when they arrive.
-   *
-   * @param reports Container report
-   */
-  @Override
-  public void processContainerReports(DatanodeDetails datanodeDetails,
-      ContainerReportsProto reports) throws IOException {
-    List<ContainerReplicaProto>
-        containerInfos = reports.getReportsList();
-    PendingDeleteStatusList pendingDeleteStatusList =
-        new PendingDeleteStatusList(datanodeDetails);
-    for (ContainerReplicaProto newInfo :
-        containerInfos) {
-      ContainerID id = ContainerID.valueof(newInfo.getContainerID());
-      ContainerReplica replica = ContainerReplica.newBuilder()
-          .setContainerID(id)
-          .setDatanodeDetails(datanodeDetails)
-          .setOriginNodeId(datanodeDetails.getUuid())
-          .build();
-      lock.lock();
-      try {
-        containerStateManager.updateContainerReplica(id, replica);
-        ContainerInfo currentInfo = containerStateManager.getContainer(id);
-        if (newInfo.getState() == ContainerReplicaProto.State.CLOSED
-            && currentInfo.getState() == LifeCycleState.CLOSING) {
-          currentInfo = updateContainerStateInternal(id, LifeCycleEvent.CLOSE);
-          if (!currentInfo.isOpen()) {
-            pipelineManager.removeContainerFromPipeline(
-                currentInfo.getPipelineID(), id);
-          }
-        }
-
-        ContainerInfoProto newState =
-            reconcileState(newInfo, currentInfo);
-
-        if (currentInfo.getDeleteTransactionId() >
-            newInfo.getDeleteTransactionId()) {
-          pendingDeleteStatusList
-                .addPendingDeleteStatus(newInfo.getDeleteTransactionId(),
-                    currentInfo.getDeleteTransactionId(),
-                    currentInfo.getContainerID());
-        }
-        containerStateManager.updateContainerInfo(
-            ContainerInfo.fromProtobuf(newState));
-        containerStore.put(id.getBytes(), newState.toByteArray());
-      } catch (ContainerNotFoundException e) {
-        LOG.error("Error while processing container report from datanode :" +
-                " {}, for container: {}, reason: container doesn't exist in" +
-                "container database.", datanodeDetails, id);
-      } finally {
-        lock.unlock();
-      }
-    }
-    if (pendingDeleteStatusList.getNumPendingDeletes() > 0) {
-      eventPublisher.fireEvent(SCMEvents.PENDING_DELETE_STATUS,
-          pendingDeleteStatusList);
-    }
-
-  }
-
-  /**
-   * Reconciles the state from Datanode with the state in SCM.
-   *
-   * @param datanodeState - State from the Datanode.
-   * @param knownState - State inside SCM.
-   * @return new SCM State for this container.
-   */
-  private HddsProtos.ContainerInfoProto reconcileState(
-      ContainerReplicaProto datanodeState,
-      ContainerInfo knownState) {
-    HddsProtos.ContainerInfoProto.Builder builder =
-        HddsProtos.ContainerInfoProto.newBuilder();
-    builder.setContainerID(knownState.getContainerID())
-        .setPipelineID(knownState.getPipelineID().getProtobuf())
-        .setState(knownState.getState())
-        .setReplicationType(knownState.getReplicationType())
-        .setReplicationFactor(knownState.getReplicationFactor())
-        .setUsedBytes(datanodeState.getUsed())
-        .setNumberOfKeys(datanodeState.getKeyCount())
-        .setStateEnterTime(knownState.getStateEnterTime())
-        .setDeleteTransactionId(knownState.getDeleteTransactionId());
-    if (knownState.getOwner() != null) {
-      builder.setOwner(knownState.getOwner());
-    }
-    return builder.build();
-  }
-
-  /**
    * Returns the latest list of DataNodes where replica for given containerId
    * exist. Throws an SCMException if no entry is found for given containerId.
    *

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/states/ContainerStateMap.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/states/ContainerStateMap.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/states/ContainerStateMap.java
index b8052a4..0c738b2 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/states/ContainerStateMap.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/states/ContainerStateMap.java
@@ -18,7 +18,6 @@
 
 package org.apache.hadoop.hdds.scm.container.states;
 
-import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 
 import org.apache.hadoop.hdds.scm.container.ContainerID;
@@ -263,12 +262,6 @@ public class ContainerStateMap {
     }
   }
 
-  @VisibleForTesting
-  // TODO: fix the test case and remove this method!
-  public static Logger getLOG() {
-    return LOG;
-  }
-
   /**
    * Just update the container State.
    * @param info ContainerInfo.

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/events/SCMEvents.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/events/SCMEvents.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/events/SCMEvents.java
index 30a7c34..72d416b 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/events/SCMEvents.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/events/SCMEvents.java
@@ -21,15 +21,13 @@ package org.apache.hadoop.hdds.scm.events;
 
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.scm.block.PendingDeleteStatusList;
-import org.apache.hadoop.hdds.scm.command.CommandStatusReportHandler
-    .CloseContainerStatus;
 import org.apache.hadoop.hdds.scm.command.CommandStatusReportHandler;
 import org.apache.hadoop.hdds.scm.command.CommandStatusReportHandler
     .ReplicationStatus;
-import org.apache.hadoop.hdds.scm.container.CloseContainerEventHandler
-        .CloseContainerRetryableReq;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher
+    .IncrementalContainerReportFromDatanode;
+import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher
         .PipelineReportFromDatanode;
 import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher
     .PipelineActionsFromDatanode;
@@ -79,6 +77,16 @@ public final class SCMEvents {
       new TypedEvent<>(ContainerReportFromDatanode.class, "Container_Report");
 
   /**
+   * IncrementalContainerReports are send out by Datanodes.
+   * This report is received by SCMDatanodeHeartbeatDispatcher and
+   * Incremental_Container_Report Event is generated.
+   */
+  public static final TypedEvent<IncrementalContainerReportFromDatanode>
+      INCREMENTAL_CONTAINER_REPORT = new TypedEvent<>(
+      IncrementalContainerReportFromDatanode.class,
+      "Incremental_Container_Report");
+
+  /**
    * ContainerActions are sent by Datanode. This event is received by
    * SCMDatanodeHeartbeatDispatcher and CONTAINER_ACTIONS event is generated.
    */
@@ -138,16 +146,6 @@ public final class SCMEvents {
       new TypedEvent<>(ContainerID.class, "Close_Container");
 
   /**
-   * A CLOSE_CONTAINER_RETRYABLE_REQ will be triggered by
-   * CloseContainerEventHandler after sending a SCMCommand to DataNode.
-   * CloseContainerWatcher will track this event. Watcher will be responsible
-   * for retrying it in event of failure or timeout.
-   */
-  public static final TypedEvent<CloseContainerRetryableReq>
-      CLOSE_CONTAINER_RETRYABLE_REQ = new TypedEvent<>(
-      CloseContainerRetryableReq.class, "Close_Container_Retryable");
-
-  /**
    * This event will be triggered whenever a new datanode is registered with
    * SCM.
    */
@@ -176,14 +174,6 @@ public final class SCMEvents {
       TypedEvent<>(ReplicationStatus.class, "Replicate_Command_Status");
   /**
    * This event will be triggered by CommandStatusReportHandler whenever a
-   * status for CloseContainer SCMCommand is received.
-   */
-  public static final Event<CloseContainerStatus>
-      CLOSE_CONTAINER_STATUS =
-      new TypedEvent<>(CloseContainerStatus.class,
-          "Close_Container_Command_Status");
-  /**
-   * This event will be triggered by CommandStatusReportHandler whenever a
    * status for DeleteBlock SCMCommand is received.
    */
   public static final TypedEvent<CommandStatusReportHandler.DeleteBlockStatus>

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DeadNodeHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DeadNodeHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DeadNodeHandler.java
index 1030428..43f0167 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DeadNodeHandler.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DeadNodeHandler.java
@@ -29,6 +29,7 @@ import org.apache.hadoop.hdds.scm.container.ContainerNotFoundException;
 import org.apache.hadoop.hdds.scm.container.ContainerReplica;
 import org.apache.hadoop.hdds.scm.container.replication.ReplicationRequest;
 import org.apache.hadoop.hdds.scm.events.SCMEvents;
+import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
 import org.apache.hadoop.hdds.server.events.EventHandler;
 import org.apache.hadoop.hdds.server.events.EventPublisher;
 
@@ -61,7 +62,14 @@ public class DeadNodeHandler implements EventHandler<DatanodeDetails> {
     // TODO: check if there are any pipeline on this node and fire close
     // pipeline event
     Set<ContainerID> ids =
-        nodeManager.getContainers(datanodeDetails.getUuid());
+        null;
+    try {
+      ids = nodeManager.getContainers(datanodeDetails);
+    } catch (NodeNotFoundException e) {
+      // This should not happen, we cannot get a dead node event for an
+      // unregistered node!
+      LOG.error("DeadNode event for a unregistered node: {}!", datanodeDetails);
+    }
     if (ids == null) {
       LOG.info("There's no containers in dead datanode {}, no replica will be"
           + " removed from the in-memory state.", datanodeDetails.getUuid());
@@ -72,18 +80,23 @@ public class DeadNodeHandler implements EventHandler<DatanodeDetails> {
     for (ContainerID id : ids) {
       try {
         final ContainerInfo container = containerManager.getContainer(id);
+        // TODO: For open containers, trigger close on other nodes
+        // TODO: Check replica count and call replication manager
+        // on these containers.
         if (!container.isOpen()) {
-          final ContainerReplica replica = ContainerReplica.newBuilder()
-              .setContainerID(id)
-              .setDatanodeDetails(datanodeDetails)
-              .build();
-          try {
-            containerManager.removeContainerReplica(id, replica);
-            replicateIfNeeded(container, publisher);
-          } catch (ContainerException ex) {
-            LOG.warn("Exception while removing container replica #{} for " +
-                "container #{}.", replica, container, ex);
-          }
+          Set<ContainerReplica> replicas = containerManager
+              .getContainerReplicas(id);
+          replicas.stream()
+              .filter(r -> r.getDatanodeDetails().equals(datanodeDetails))
+              .findFirst()
+              .ifPresent(replica -> {
+                try {
+                  containerManager.removeContainerReplica(id, replica);
+                } catch (ContainerException ex) {
+                  LOG.warn("Exception while removing container replica #{} " +
+                      "for container #{}.", replica, container, ex);
+                }
+              });
         }
       } catch (ContainerNotFoundException cnfe) {
         LOG.warn("Container Not found!", cnfe);

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NewNodeHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NewNodeHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NewNodeHandler.java
index 780aa2b..ed4fdba 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NewNodeHandler.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NewNodeHandler.java
@@ -19,31 +19,17 @@
 package org.apache.hadoop.hdds.scm.node;
 
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
-import org.apache.hadoop.hdds.scm.exceptions.SCMException;
 import org.apache.hadoop.hdds.server.events.EventHandler;
 import org.apache.hadoop.hdds.server.events.EventPublisher;
 
-import java.util.Collections;
-
 /**
  * Handles New Node event.
  */
 public class NewNodeHandler implements EventHandler<DatanodeDetails> {
 
-  private final NodeManager nodeManager;
-
-  public NewNodeHandler(NodeManager nodeManager) {
-    this.nodeManager = nodeManager;
-  }
-
   @Override
   public void onMessage(DatanodeDetails datanodeDetails,
                         EventPublisher publisher) {
-    try {
-      nodeManager.addDatanodeInContainerMap(datanodeDetails.getUuid(),
-          Collections.emptySet());
-    } catch (SCMException e) {
-      // TODO: log exception message.
-    }
+    // We currently have nothing to do when we receive new node event.
   }
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java
index d55ff98..e944634 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java
@@ -27,7 +27,6 @@ import org.apache.hadoop.hdds.scm.exceptions.SCMException;
 import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState;
-import org.apache.hadoop.hdds.scm.node.states.ReportResult;
 import org.apache.hadoop.hdds.server.events.EventHandler;
 import org.apache.hadoop.ozone.protocol.StorageContainerNodeProtocol;
 import org.apache.hadoop.ozone.protocol.commands.CommandForDatanode;
@@ -63,13 +62,6 @@ import java.util.UUID;
  */
 public interface NodeManager extends StorageContainerNodeProtocol,
     EventHandler<CommandForDatanode>, NodeManagerMXBean, Closeable {
-  /**
-   * Removes a data node from the management of this Node Manager.
-   *
-   * @param node - DataNode.
-   * @throws NodeNotFoundException
-   */
-  void removeNode(DatanodeDetails node) throws NodeNotFoundException;
 
   /**
    * Gets all Live Datanodes that is currently communicating with SCM.
@@ -102,6 +94,7 @@ public interface NodeManager extends StorageContainerNodeProtocol,
    * Return a map of node stats.
    * @return a map of individual node stats (live/stale but not dead).
    */
+  // TODO: try to change the return type to Map<DatanodeDetails, SCMNodeStat>
   Map<UUID, SCMNodeStat> getNodeStats();
 
   /**
@@ -121,10 +114,10 @@ public interface NodeManager extends StorageContainerNodeProtocol,
 
   /**
    * Get set of pipelines a datanode is part of.
-   * @param dnId - datanodeID
+   * @param datanodeDetails DatanodeDetails
    * @return Set of PipelineID
    */
-  Set<PipelineID> getPipelineByDnID(UUID dnId);
+  Set<PipelineID> getPipelines(DatanodeDetails datanodeDetails);
 
   /**
    * Add pipeline information in the NodeManager.
@@ -139,40 +132,22 @@ public interface NodeManager extends StorageContainerNodeProtocol,
   void removePipeline(Pipeline pipeline);
 
   /**
-   * Update set of containers available on a datanode.
-   * @param uuid - DatanodeID
+   * Remaps datanode to containers mapping to the new set of containers.
+   * @param datanodeDetails - DatanodeDetails
    * @param containerIds - Set of containerIDs
    * @throws SCMException - if datanode is not known. For new datanode use
    *                        addDatanodeInContainerMap call.
    */
-  void setContainersForDatanode(UUID uuid, Set<ContainerID> containerIds)
-      throws SCMException;
-
-  /**
-   * Process containerReport received from datanode.
-   * @param uuid - DataonodeID
-   * @param containerIds - Set of containerIDs
-   * @return The result after processing containerReport
-   */
-  ReportResult<ContainerID> processContainerReport(UUID uuid,
-      Set<ContainerID> containerIds);
+  void setContainers(DatanodeDetails datanodeDetails,
+      Set<ContainerID> containerIds) throws NodeNotFoundException;
 
   /**
    * Return set of containerIDs available on a datanode.
-   * @param uuid - DatanodeID
-   * @return - set of containerIDs
-   */
-  Set<ContainerID> getContainers(UUID uuid);
-
-  /**
-   * Insert a new datanode with set of containerIDs for containers available
-   * on it.
-   * @param uuid - DatanodeID
-   * @param containerIDs - Set of ContainerIDs
-   * @throws SCMException - if datanode already exists
+   * @param datanodeDetails DatanodeDetails
+   * @return set of containerIDs
    */
-  void addDatanodeInContainerMap(UUID uuid, Set<ContainerID> containerIDs)
-      throws SCMException;
+  Set<ContainerID> getContainers(DatanodeDetails datanodeDetails)
+      throws NodeNotFoundException;
 
   /**
    * Add a {@link SCMCommand} to the command queue, which are
@@ -188,7 +163,7 @@ public interface NodeManager extends StorageContainerNodeProtocol,
    * @param dnUuid
    * @param nodeReport
    */
-  void processNodeReport(UUID dnUuid, NodeReportProto nodeReport);
+  void processNodeReport(DatanodeDetails dnUuid, NodeReportProto nodeReport);
 
   /**
    * Process a dead node event in this Node Manager.
@@ -202,5 +177,6 @@ public interface NodeManager extends StorageContainerNodeProtocol,
    * @param dnID - Datanode uuid.
    * @return list of commands
    */
+  // TODO: We can give better name to this method!
   List<SCMCommand> getCommandQueue(UUID dnID);
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeReportHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeReportHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeReportHandler.java
index 331bfed..f419764 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeReportHandler.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeReportHandler.java
@@ -50,6 +50,6 @@ public class NodeReportHandler implements EventHandler<NodeReportFromDatanode> {
         + "missing DatanodeDetails.");
     LOGGER.trace("Processing node report for dn: {}", dn);
     nodeManager
-        .processNodeReport(dn.getUuid(), nodeReportFromDatanode.getReport());
+        .processNodeReport(dn, nodeReportFromDatanode.getReport());
   }
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeStateManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeStateManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeStateManager.java
index 588756c..a459519 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeStateManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeStateManager.java
@@ -29,7 +29,6 @@ import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
 import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeStat;
 import org.apache.hadoop.hdds.scm.events.SCMEvents;
-import org.apache.hadoop.hdds.scm.exceptions.SCMException;
 import org.apache.hadoop.hdds.scm.node.states.*;
 import org.apache.hadoop.hdds.scm.node.states.Node2PipelineMap;
 import org.apache.hadoop.hdds.server.events.Event;
@@ -94,11 +93,6 @@ public class NodeStateManager implements Runnable, Closeable {
    */
   private final Node2PipelineMap node2PipelineMap;
   /**
-   * Maintains the map from node to ContainerIDs for the containers
-   * available on the node.
-   */
-  private final Node2ContainerMap node2ContainerMap;
-  /**
    * Used for publishing node state change events.
    */
   private final EventPublisher eventPublisher;
@@ -131,7 +125,6 @@ public class NodeStateManager implements Runnable, Closeable {
   public NodeStateManager(Configuration conf, EventPublisher eventPublisher) {
     this.nodeStateMap = new NodeStateMap();
     this.node2PipelineMap = new Node2PipelineMap();
-    this.node2ContainerMap = new Node2ContainerMap();
     this.eventPublisher = eventPublisher;
     this.state2EventMap = new HashMap<>();
     initialiseState2EventMap();
@@ -431,18 +424,6 @@ public class NodeStateManager implements Runnable, Closeable {
   }
 
   /**
-   * Removes a node from NodeStateManager.
-   *
-   * @param datanodeDetails DatanodeDetails
-   *
-   * @throws NodeNotFoundException if the node is not present
-   */
-  public void removeNode(DatanodeDetails datanodeDetails)
-      throws NodeNotFoundException {
-    nodeStateMap.removeNode(datanodeDetails.getUuid());
-  }
-
-  /**
    * Returns the current stats of the node.
    *
    * @param uuid node id
@@ -475,19 +456,6 @@ public class NodeStateManager implements Runnable, Closeable {
   }
 
   /**
-   * Remove the current stats of the specify node.
-   *
-   * @param uuid node id
-   *
-   * @return SCMNodeStat the stat removed from the node.
-   *
-   * @throws NodeNotFoundException if the node is not present.
-   */
-  public SCMNodeStat removeNodeStat(UUID uuid) throws NodeNotFoundException {
-    return nodeStateMap.removeNodeStat(uuid);
-  }
-
-  /**
    * Removes a pipeline from the node2PipelineMap.
    * @param pipeline - Pipeline to be removed
    */
@@ -498,23 +466,11 @@ public class NodeStateManager implements Runnable, Closeable {
    * Update set of containers available on a datanode.
    * @param uuid - DatanodeID
    * @param containerIds - Set of containerIDs
-   * @throws SCMException - if datanode is not known. For new datanode use
-   *                        addDatanodeInContainerMap call.
-   */
-  public void setContainersForDatanode(UUID uuid, Set<ContainerID> containerIds)
-      throws SCMException {
-    node2ContainerMap.setContainersForDatanode(uuid, containerIds);
-  }
-
-  /**
-   * Process containerReport received from datanode.
-   * @param uuid - DataonodeID
-   * @param containerIds - Set of containerIDs
-   * @return The result after processing containerReport
+   * @throws NodeNotFoundException - if datanode is not known.
    */
-  public ReportResult<ContainerID> processContainerReport(UUID uuid,
-      Set<ContainerID> containerIds) {
-    return node2ContainerMap.processReport(uuid, containerIds);
+  public void setContainers(UUID uuid, Set<ContainerID> containerIds)
+      throws NodeNotFoundException {
+    nodeStateMap.setContainers(uuid, containerIds);
   }
 
   /**
@@ -522,20 +478,9 @@ public class NodeStateManager implements Runnable, Closeable {
    * @param uuid - DatanodeID
    * @return - set of containerIDs
    */
-  public Set<ContainerID> getContainers(UUID uuid) {
-    return node2ContainerMap.getContainers(uuid);
-  }
-
-  /**
-   * Insert a new datanode with set of containerIDs for containers available
-   * on it.
-   * @param uuid - DatanodeID
-   * @param containerIDs - Set of ContainerIDs
-   * @throws SCMException - if datanode already exists
-   */
-  public void addDatanodeInContainerMap(UUID uuid,
-      Set<ContainerID> containerIDs) throws SCMException {
-    node2ContainerMap.insertNewDatanode(uuid, containerIDs);
+  public Set<ContainerID> getContainers(UUID uuid)
+      throws NodeNotFoundException {
+    return nodeStateMap.getContainers(uuid);
   }
 
   /**

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/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 35c22f3..c42ef66 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
@@ -27,7 +27,6 @@ import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
 import org.apache.hadoop.hdds.scm.exceptions.SCMException;
 import org.apache.hadoop.hdds.scm.node.states.NodeAlreadyExistsException;
 import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
-import org.apache.hadoop.hdds.scm.node.states.ReportResult;
 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;
@@ -129,16 +128,6 @@ public class SCMNodeManager
     }
   }
 
-  /**
-   * Removes a data node from the management of this Node Manager.
-   *
-   * @param node - DataNode.
-   * @throws NodeNotFoundException
-   */
-  @Override
-  public void removeNode(DatanodeDetails node) throws NodeNotFoundException {
-    nodeStateManager.removeNode(node);
-  }
 
   /**
    * Gets all datanodes that are in a certain state. This function works by
@@ -270,10 +259,8 @@ public class SCMNodeManager
       datanodeDetails.setHostName(dnAddress.getHostName());
       datanodeDetails.setIpAddress(dnAddress.getHostAddress());
     }
-    UUID dnId = datanodeDetails.getUuid();
     try {
       nodeStateManager.addNode(datanodeDetails);
-      nodeStateManager.setNodeStat(dnId, new SCMNodeStat());
       // Updating Node Report, as registration is successful
       updateNodeStat(datanodeDetails.getUuid(), nodeReport);
       LOG.info("Registered Data node : {}", datanodeDetails);
@@ -326,8 +313,9 @@ public class SCMNodeManager
    * @param nodeReport
    */
   @Override
-  public void processNodeReport(UUID dnUuid, NodeReportProto nodeReport) {
-    this.updateNodeStat(dnUuid, nodeReport);
+  public void processNodeReport(DatanodeDetails dnUuid,
+                                NodeReportProto nodeReport) {
+    this.updateNodeStat(dnUuid.getUuid(), nodeReport);
   }
 
   /**
@@ -377,12 +365,12 @@ public class SCMNodeManager
 
   /**
    * Get set of pipelines a datanode is part of.
-   * @param dnId - datanodeID
+   * @param datanodeDetails - datanodeID
    * @return Set of PipelineID
    */
   @Override
-  public Set<PipelineID> getPipelineByDnID(UUID dnId) {
-    return nodeStateManager.getPipelineByDnID(dnId);
+  public Set<PipelineID> getPipelines(DatanodeDetails datanodeDetails) {
+    return nodeStateManager.getPipelineByDnID(datanodeDetails.getUuid());
   }
 
 
@@ -406,50 +394,27 @@ public class SCMNodeManager
 
   /**
    * Update set of containers available on a datanode.
-   * @param uuid - DatanodeID
+   * @param datanodeDetails - DatanodeID
    * @param containerIds - Set of containerIDs
    * @throws SCMException - if datanode is not known. For new datanode use
    *                        addDatanodeInContainerMap call.
    */
   @Override
-  public void setContainersForDatanode(UUID uuid,
-      Set<ContainerID> containerIds) throws SCMException {
-    nodeStateManager.setContainersForDatanode(uuid, containerIds);
-  }
-
-  /**
-   * Process containerReport received from datanode.
-   * @param uuid - DataonodeID
-   * @param containerIds - Set of containerIDs
-   * @return The result after processing containerReport
-   */
-  @Override
-  public ReportResult<ContainerID> processContainerReport(UUID uuid,
-      Set<ContainerID> containerIds) {
-    return nodeStateManager.processContainerReport(uuid, containerIds);
+  public void setContainers(DatanodeDetails datanodeDetails,
+      Set<ContainerID> containerIds) throws NodeNotFoundException {
+    nodeStateManager.setContainers(datanodeDetails.getUuid(),
+        containerIds);
   }
 
   /**
    * Return set of containerIDs available on a datanode.
-   * @param uuid - DatanodeID
+   * @param datanodeDetails - DatanodeID
    * @return - set of containerIDs
    */
   @Override
-  public Set<ContainerID> getContainers(UUID uuid) {
-    return nodeStateManager.getContainers(uuid);
-  }
-
-  /**
-   * Insert a new datanode with set of containerIDs for containers available
-   * on it.
-   * @param uuid - DatanodeID
-   * @param containerIDs - Set of ContainerIDs
-   * @throws SCMException - if datanode already exists
-   */
-  @Override
-  public void addDatanodeInContainerMap(UUID uuid,
-      Set<ContainerID> containerIDs) throws SCMException {
-    nodeStateManager.addDatanodeInContainerMap(uuid, containerIDs);
+  public Set<ContainerID> getContainers(DatanodeDetails datanodeDetails)
+      throws NodeNotFoundException {
+    return nodeStateManager.getContainers(datanodeDetails.getUuid());
   }
 
   // TODO:
@@ -481,6 +446,7 @@ public class SCMNodeManager
    * @param dnUuid datanode uuid.
    */
   @Override
+  // TODO: This should be removed.
   public void processDeadNode(UUID dnUuid) {
     try {
       SCMNodeStat stat = nodeStateManager.getNodeStat(dnUuid);

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/StaleNodeHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/StaleNodeHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/StaleNodeHandler.java
index 9df9dff..268fe5b 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/StaleNodeHandler.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/StaleNodeHandler.java
@@ -49,7 +49,7 @@ public class StaleNodeHandler implements EventHandler<DatanodeDetails> {
   public void onMessage(DatanodeDetails datanodeDetails,
                         EventPublisher publisher) {
     Set<PipelineID> pipelineIds =
-        nodeManager.getPipelineByDnID(datanodeDetails.getUuid());
+        nodeManager.getPipelines(datanodeDetails);
     for (PipelineID pipelineID : pipelineIds) {
       try {
         pipelineManager.finalizePipeline(pipelineID);

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/NodeStateMap.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/NodeStateMap.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/NodeStateMap.java
index 774ced1..a917e79 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/NodeStateMap.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/NodeStateMap.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.hdds.scm.node.states;
 
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState;
+import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeStat;
 import org.apache.hadoop.hdds.scm.node.DatanodeInfo;
 
@@ -48,6 +49,10 @@ public class NodeStateMap {
    * Represents the current stats of node.
    */
   private final ConcurrentHashMap<UUID, SCMNodeStat> nodeStats;
+  /**
+   * Node to set of containers on the node.
+   */
+  private final ConcurrentHashMap<UUID, Set<ContainerID>> nodeToContainer;
 
   private final ReadWriteLock lock;
 
@@ -59,6 +64,7 @@ public class NodeStateMap {
     nodeMap = new ConcurrentHashMap<>();
     stateMap = new ConcurrentHashMap<>();
     nodeStats = new ConcurrentHashMap<>();
+    nodeToContainer = new ConcurrentHashMap<>();
     initStateMap();
   }
 
@@ -88,6 +94,8 @@ public class NodeStateMap {
         throw new NodeAlreadyExistsException("Node UUID: " + id);
       }
       nodeMap.put(id, new DatanodeInfo(datanodeDetails));
+      nodeStats.put(id, new SCMNodeStat());
+      nodeToContainer.put(id, Collections.emptySet());
       stateMap.get(nodeState).add(id);
     } finally {
       lock.writeLock().unlock();
@@ -238,30 +246,6 @@ public class NodeStateMap {
   }
 
   /**
-   * Removes the node from NodeStateMap.
-   *
-   * @param uuid node id
-   *
-   * @throws NodeNotFoundException if the node is not found
-   */
-  public void removeNode(UUID uuid) throws NodeNotFoundException {
-    lock.writeLock().lock();
-    try {
-      if (nodeMap.containsKey(uuid)) {
-        for (Map.Entry<NodeState, Set<UUID>> entry : stateMap.entrySet()) {
-          if(entry.getValue().remove(uuid)) {
-            break;
-          }
-          nodeMap.remove(uuid);
-        }
-        throw new NodeNotFoundException("Node UUID: " + uuid);
-      }
-    } finally {
-      lock.writeLock().unlock();
-    }
-  }
-
-  /**
    * Returns the current stats of the node.
    *
    * @param uuid node id
@@ -298,21 +282,30 @@ public class NodeStateMap {
     nodeStats.put(uuid, newstat);
   }
 
-  /**
-   * Remove the current stats of the specify node.
-   *
-   * @param uuid node id
-   *
-   * @return SCMNodeStat the stat removed from the node.
-   *
-   * @throws NodeNotFoundException if the node is not found
-   */
-  public SCMNodeStat removeNodeStat(UUID uuid) throws NodeNotFoundException {
-    SCMNodeStat stat = nodeStats.remove(uuid);
-    if (stat == null) {
+  public void setContainers(UUID uuid, Set<ContainerID> containers)
+      throws NodeNotFoundException{
+    if (!nodeToContainer.containsKey(uuid)) {
       throw new NodeNotFoundException("Node UUID: " + uuid);
     }
-    return stat;
+    nodeToContainer.put(uuid, containers);
+  }
+
+  public Set<ContainerID> getContainers(UUID uuid)
+      throws NodeNotFoundException {
+    Set<ContainerID> containers = nodeToContainer.get(uuid);
+    if (containers == null) {
+      throw new NodeNotFoundException("Node UUID: " + uuid);
+    }
+    return Collections.unmodifiableSet(containers);
+  }
+
+  public void removeContainer(UUID uuid, ContainerID containerID) throws
+      NodeNotFoundException {
+    Set<ContainerID> containers = nodeToContainer.get(uuid);
+    if (containers == null) {
+      throw new NodeNotFoundException("Node UUID: " + uuid);
+    }
+    containers.remove(containerID);
   }
 
   /**

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeHeartbeatDispatcher.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeHeartbeatDispatcher.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeHeartbeatDispatcher.java
index 7c7df27..ede8b4f 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeHeartbeatDispatcher.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeHeartbeatDispatcher.java
@@ -20,7 +20,9 @@ package org.apache.hadoop.hdds.scm.server;
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.proto
-        .StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
+    .StorageContainerDatanodeProtocolProtos.IncrementalContainerReportProto;
+import org.apache.hadoop.hdds.protocol.proto
+    .StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
 import org.apache.hadoop.hdds.protocol.proto
     .StorageContainerDatanodeProtocolProtos.PipelineActionsProto;
 import org.apache.hadoop.hdds.protocol.proto
@@ -210,6 +212,19 @@ public final class SCMDatanodeHeartbeatDispatcher {
   }
 
   /**
+   * Incremental Container report event payload with origin.
+   */
+  public static class IncrementalContainerReportFromDatanode
+      extends ReportFromDatanode<IncrementalContainerReportProto> {
+
+    public IncrementalContainerReportFromDatanode(
+        DatanodeDetails datanodeDetails,
+        IncrementalContainerReportProto report) {
+      super(datanodeDetails, report);
+    }
+  }
+
+  /**
    * Container action event payload with origin.
    */
   public static class ContainerActionsFromDatanode

http://git-wip-us.apache.org/repos/asf/hadoop/blob/c80f753b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeProtocolServer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeProtocolServer.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeProtocolServer.java
index 37c7386..0beceab 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeProtocolServer.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeProtocolServer.java
@@ -97,6 +97,7 @@ import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DATANODE_ADDRES
 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.scm.events.SCMEvents.CONTAINER_REPORT;
 import static org.apache.hadoop.hdds.scm.events.SCMEvents.PIPELINE_REPORT;
 import static org.apache.hadoop.hdds.scm.server.StorageContainerManager.startRpcServer;
 import static org.apache.hadoop.hdds.server.ServerUtils.updateRPCListenAddress;
@@ -196,8 +197,9 @@ public class SCMDatanodeProtocolServer implements
         .register(datanodeDetails, nodeReport, pipelineReportsProto);
     if (registeredCommand.getError()
         == SCMRegisteredResponseProto.ErrorCode.success) {
-      scm.getContainerManager().processContainerReports(datanodeDetails,
-          containerReportsProto);
+      eventPublisher.fireEvent(CONTAINER_REPORT,
+          new SCMDatanodeHeartbeatDispatcher.ContainerReportFromDatanode(
+              datanodeDetails, containerReportsProto));
       eventPublisher.fireEvent(SCMEvents.NODE_REGISTRATION_CONT_REPORT,
           new NodeRegistrationContainerReport(datanodeDetails,
               containerReportsProto));


---------------------------------------------------------------------
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