Rebase and fix review comments

Author: ashishkr200

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfig.java

@@ -138,31 +138,10 @@ public class ScmConfig extends ReconfigurableConfig {
   )
   private int transactionToDNsCommitMapLimit = 5000000;
 
-  @Config(key = "hdds.scm.container.pending.allocation.roll.interval",
-      defaultValue = "5m",
-      type = ConfigType.TIME,
-      tags = { ConfigTag.SCM, ConfigTag.CONTAINER },
-      description =
-          "Time interval for rolling the pending container allocation window. " +
-              "Pending container allocations are tracked in a two-window tumbling bucket " +
-              "pattern. Each window has this duration. " +
-              "After 2x this interval, allocations that haven't been confirmed via " +
-              "container reports will automatically age out. Default is 5 minutes."
-  )
-  private Duration pendingContainerAllocationRollInterval = Duration.ofMinutes(5);
-
   public int getTransactionToDNsCommitMapLimit() {
     return transactionToDNsCommitMapLimit;
   }
 
-  public Duration getPendingContainerAllocationRollInterval() {
-    return pendingContainerAllocationRollInterval;
-  }
-
-  public void setPendingContainerAllocationRollInterval(Duration duration) {
-    this.pendingContainerAllocationRollInterval = duration;
-  }
-
   public Duration getBlockDeletionInterval() {
     return blockDeletionInterval;
   }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java

@@ -269,6 +269,11 @@ public final class ScmConfigKeys {
   public static final String OZONE_SCM_STALENODE_INTERVAL_DEFAULT =
       "5m";
 
+  public static final String OZONE_SCM_PENDING_CONTAINER_ROLL_INTERVAL =
+      "ozone.scm.pending.container.roll.interval";
+  public static final String OZONE_SCM_PENDING_CONTAINER_ROLL_INTERVAL_DEFAULT =
+      "5m";
+
   public static final String OZONE_SCM_HEARTBEAT_RPC_TIMEOUT =
       "ozone.scm.heartbeat.rpc-timeout";
   public static final String OZONE_SCM_HEARTBEAT_RPC_TIMEOUT_DEFAULT =

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -1037,6 +1037,17 @@
       balances the amount of metadata.
     </description>
   </property>
+  <property>
+    <name>ozone.scm.pending.container.roll.interval</name>
+    <value>5m</value>
+    <tag>OZONE, SCM, PERFORMANCE, MANAGEMENT</tag>
+    <description>
+      The interval at which the two-window tumbling bucket for pending
+      container allocations rolls over per DataNode. Pending containers
+      that have not been confirmed within two intervals are automatically
+      aged out. Default is 5 minutes.
+    </description>
+  </property>
   <property>
     <name>ozone.scm.container.lock.stripes</name>
     <value>512</value>

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

@@ -136,15 +136,15 @@ public List<ContainerInfo> getContainers(ReplicationType type) {
 
   @Override
   public List<ContainerID> getContainerIDs(final ContainerID startID,
-      final int count,
-      final LifeCycleState state) {
+                                           final int count,
+                                           final LifeCycleState state) {
     scmContainerManagerMetrics.incNumListContainersOps();
     return containerStateManager.getContainerIDs(state, startID, count);
   }
 
   @Override
   public List<ContainerInfo> getContainers(final ContainerID startID,
-      final int count) {
+                                           final int count) {
     scmContainerManagerMetrics.incNumListContainersOps();
     return containerStateManager.getContainerInfos(startID, count);
   }
@@ -157,8 +157,8 @@ public List<ContainerInfo> getContainers(final LifeCycleState state) {
 
   @Override
   public List<ContainerInfo> getContainers(final ContainerID startID,
-      final int count,
-      final LifeCycleState state) {
+                                           final int count,
+                                           final LifeCycleState state) {
     scmContainerManagerMetrics.incNumListContainersOps();
     return containerStateManager.getContainerInfos(state, startID, count);
   }
@@ -271,14 +271,12 @@ private ContainerInfo allocateContainer(final Pipeline pipeline,
     containerStateManager.addContainer(containerInfoBuilder.build());
     pipelineManager.recordPendingAllocation(pipeline, containerID);
     scmContainerManagerMetrics.incNumSuccessfulCreateContainers();
-    pipelineManager.recordPendingAllocation(pipeline, containerID);
-
     return containerStateManager.getContainer(containerID);
   }
 
   @Override
   public void updateContainerState(final ContainerID cid,
-      final LifeCycleEvent event)
+                                   final LifeCycleEvent event)
       throws IOException, InvalidStateTransitionException {
     HddsProtos.ContainerID protoId = cid.getProtobuf();
     lock.lock();
@@ -314,7 +312,7 @@ public void updateContainerInfo(final ContainerID cid, ContainerInfoProto contai
 
   @Override
   public void transitionDeletingOrDeletedToTargetState(ContainerID containerID, LifeCycleState targetState)
-      throws IOException {
+          throws IOException {
     HddsProtos.ContainerID proto = containerID.getProtobuf();
     lock.lock();
     try {
@@ -340,7 +338,7 @@ public Set<ContainerReplica> getContainerReplicas(final ContainerID id)
 
   @Override
   public void updateContainerReplica(final ContainerID cid,
-      final ContainerReplica replica)
+                                     final ContainerReplica replica)
       throws ContainerNotFoundException {
     if (containerExist(cid)) {
       containerStateManager.updateContainerReplica(replica);
@@ -351,7 +349,7 @@ public void updateContainerReplica(final ContainerID cid,
 
   @Override
   public void removeContainerReplica(final ContainerID cid,
-      final ContainerReplica replica)
+                                     final ContainerReplica replica)
       throws ContainerNotFoundException, ContainerReplicaNotFoundException {
     if (containerExist(cid)) {
       containerStateManager.removeContainerReplica(replica);
@@ -422,7 +420,7 @@ private NavigableSet<ContainerID> getContainersForOwner(
 
   @Override
   public void notifyContainerReportProcessing(final boolean isFullReport,
-      final boolean success) {
+                                              final boolean success) {
     if (isFullReport) {
       if (success) {
         scmContainerManagerMetrics.incNumContainerReportsProcessedSuccessful();

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

@@ -176,17 +176,11 @@ public void onMessage(final ContainerReportFromDatanode reportFromDatanode,
           if (!alreadyInDn) {
             // This is a new Container not in the nodeManager -> dn map yet
             getNodeManager().addContainer(datanodeDetails, cid);
-            
             // Remove from pending tracker when container is added to DN
             // This container was just confirmed for the first time on this DN
-            // No need to remove on subsequent reports (it's already been removed)
-            if (container != null) {
-              PendingContainerTracker tracker =
-                  getNodeManager().getPendingContainerTracker();
-              if (tracker != null) {
-                tracker.removePendingAllocation(datanodeDetails, cid);
-              }
-            }
+            PendingContainerTracker tracker = getNodeManager().getPendingContainerTracker();
+            tracker.removePendingAllocation(getNodeManager().getDatanodeInfo(datanodeDetails).
+                getPendingContainerAllocations(), cid);
           }
           if (container == null || ContainerReportValidator
                   .validate(container, datanodeDetails, replica)) {

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

@@ -104,12 +104,8 @@ protected void processICR(IncrementalContainerReportFromDatanode report,
           }
           if (ContainerReportValidator.validate(container, dd, replicaProto)) {
             processContainerReplica(dd, container, replicaProto, publisher, detailsForLogging);
-
-            PendingContainerTracker tracker =
-                getNodeManager().getPendingContainerTracker();
-            if (tracker != null) {
-              tracker.removePendingAllocation(dd, id);
-            }
+            PendingContainerTracker tracker = getNodeManager().getPendingContainerTracker();
+            tracker.removePendingAllocation(getNodeManager().getDatanodeInfo(dd).getPendingContainerAllocations(), id);
           }
           success = true;
         } catch (ContainerNotFoundException e) {

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java

@@ -440,23 +440,4 @@ default void removeNode(DatanodeDetails datanodeDetails) throws NodeNotFoundExce
    * SCM-side tracker for container allocations not yet reported by datanodes.
    */
   PendingContainerTracker getPendingContainerTracker();
-
-  /**
-   * True if the node can accept another container of the given size, accounting for
-   * {@link #getPendingContainerTracker()}.
-   */
-  boolean hasSpaceForNewContainerAllocation(DatanodeDetails node);
-
-  /**
-   * Records a pending container allocation for {@code node} so that subsequent
-   * space checks via {@link #hasSpaceForNewContainerAllocation} account for the
-   * in-flight allocation before the datanode sends an ICR.
-   *
-   * @param node        the datanode that will receive the new container replica
-   * @param containerID the container being allocated
-   */
-  default void recordPendingAllocationForDatanode(DatanodeDetails node,
-      ContainerID containerID) {
-    getPendingContainerTracker().recordPendingAllocationForDatanode(node, containerID);
-  }
 }

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeStateManager.java

@@ -47,6 +47,7 @@
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.LayoutVersionProto;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.hdds.scm.events.SCMEvents;
 import org.apache.hadoop.hdds.scm.ha.SCMContext;
@@ -124,7 +125,7 @@ public class NodeStateManager implements Runnable, Closeable {
    */
   private final long deadNodeIntervalMs;
 
-  private final long containerRollIntervalMs = 5 * 60 * 1000;  //TODO
+  private final long containerRollIntervalMs;  //TODO
 
   /**
    * The future is used to pause/unpause the scheduled checks.
@@ -214,6 +215,11 @@ public NodeStateManager(ConfigurationSource conf,
             scmContext.getFinalizationCheckpoint()) &&
             !layoutMatchCondition.test(layout);
 
+    containerRollIntervalMs = conf.getTimeDuration(
+        ScmConfigKeys.OZONE_SCM_PENDING_CONTAINER_ROLL_INTERVAL,
+        ScmConfigKeys.OZONE_SCM_PENDING_CONTAINER_ROLL_INTERVAL_DEFAULT,
+        TimeUnit.MILLISECONDS);
+
     scheduleNextHealthCheck();
   }
 

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/PendingContainerTracker.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.hdds.scm.node;
 
+import com.google.common.annotations.VisibleForTesting;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
@@ -222,4 +223,9 @@ public void removePendingAllocation(TwoWindowBucket bucket, ContainerID containe
       metrics.incNumPendingContainersRemoved();
     }
   }
+
+  @VisibleForTesting
+  public SCMNodeMetrics getMetrics() {
+    return metrics;
+  }
 }

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java

@@ -69,7 +69,6 @@
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMRegisteredResponseProto.ErrorCode;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMVersionRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.StorageReportProto;
-import org.apache.hadoop.hdds.scm.ScmConfig;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.scm.VersionInfo;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
@@ -145,13 +144,6 @@ public class SCMNodeManager implements NodeManager {
   private final NonWritableNodeFilter nonWritableNodeFilter;
   private final int numContainerPerVolume;
 
-  /**
-   * SCM-side pending container allocations per datanode (not yet in container reports).
-   */
-  private final PendingContainerTracker pendingContainerTracker;
-
-  private final long maxContainerSizeBytes;
-
   /**
    * Lock used to synchronize some operation in Node manager to ensure a
    * consistent view of the node state.
@@ -200,7 +192,10 @@ public SCMNodeManager(
     this.pendingContainerTracker = new PendingContainerTracker(
         (long) conf.getStorageSize(ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE,
             ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE_DEFAULT, StorageUnit.BYTES),
-        5 * 60 * 1000, // TODO
+        conf.getTimeDuration(
+            ScmConfigKeys.OZONE_SCM_PENDING_CONTAINER_ROLL_INTERVAL,
+            ScmConfigKeys.OZONE_SCM_PENDING_CONTAINER_ROLL_INTERVAL_DEFAULT,
+            TimeUnit.MILLISECONDS),
         this.metrics);
     this.clusterMap = networkTopology;
     this.nodeResolver = nodeResolver;
@@ -219,14 +214,6 @@ public SCMNodeManager(
     this.scmContext = scmContext;
     this.sendCommandNotifyMap = new HashMap<>();
     this.nonWritableNodeFilter = new NonWritableNodeFilter(conf);
-
-    this.maxContainerSizeBytes = (long) conf.getStorageSize(
-        ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE,
-        ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE_DEFAULT, StorageUnit.BYTES);
-    ScmConfig scmConfig = conf.getObject(ScmConfig.class);
-    long rollIntervalMs = scmConfig.getPendingContainerAllocationRollInterval().toMillis();
-    this.pendingContainerTracker = new PendingContainerTracker(
-        maxContainerSizeBytes, rollIntervalMs, this.metrics);
   }
 
   @Override
@@ -252,35 +239,6 @@ public PendingContainerTracker getPendingContainerTracker() {
     return pendingContainerTracker;
   }
 
-  /**
-   * Effective space check aligned with container allocation: per-disk slot model minus
-   * SCM pending allocations.
-   */
-  @Override
-  public boolean hasSpaceForNewContainerAllocation(DatanodeDetails node) {
-    Objects.requireNonNull(node, "node==null");
-    try {
-      DatanodeInfo datanodeInfo = getDatanodeInfo(node);
-      if (datanodeInfo == null) {
-        LOG.warn("DatanodeInfo not found for node {}", node.getUuidString());
-        return false;
-      }
-      return pendingContainerTracker.hasEffectiveAllocatableSpaceForNewContainer(
-          node, datanodeInfo);
-    } catch (Exception e) {
-      LOG.warn("Error checking allocatable space for node {}", node.getUuidString(), e);
-      return false;
-    }
-  }
-
-  @Override
-  public void recordPendingAllocationForDatanode(DatanodeDetails node,
-      ContainerID containerID) {
-    Objects.requireNonNull(node, "node==null");
-    Objects.requireNonNull(containerID, "containerID==null");
-    pendingContainerTracker.recordPendingAllocationForDatanode(node, containerID);
-  }
-
   protected NodeStateManager getNodeStateManager() {
     return nodeStateManager;
   }
@@ -762,7 +720,6 @@ public void processNodeReport(DatanodeDetails datanodeDetails,
         datanodeInfo.updateStorageReports(nodeReport.getStorageReportList());
         datanodeInfo.updateMetaDataStorageReports(nodeReport.
             getMetadataStorageReportList());
-        pendingContainerTracker.rollWindowsIfNeeded(datanodeDetails);
         metrics.incNumNodeReportProcessed();
       }
     } catch (NodeNotFoundException e) {