From a881db0b6ebc298bee7626c34f3d411442920cdb Mon Sep 17 00:00:00 2001 From: ashishk Date: Fri, 12 Jun 2026 13:32:52 +0530 Subject: [PATCH 1/4] HDDS-15541. PlacementPolicy to use PendingContainerTracker to check space availability. --- .../hdds/scm/SCMCommonPlacementPolicy.java | 29 +++++---------- .../hadoop/hdds/scm/node/NodeManager.java | 9 +++++ .../scm/node/PendingContainerTracker.java | 34 +++++++++++++++-- .../hadoop/hdds/scm/node/SCMNodeManager.java | 13 +++++-- .../hdds/scm/pipeline/PipelineProvider.java | 3 +- .../scm/TestSCMCommonPlacementPolicy.java | 37 +++---------------- .../hdds/scm/container/MockNodeManager.java | 13 +++++++ .../scm/container/SimpleMockNodeManager.java | 5 +++ .../TestContainerPlacementFactory.java | 2 + .../TestSCMContainerPlacementCapacity.java | 4 ++ .../TestSCMContainerPlacementRackAware.java | 2 + .../TestSCMContainerPlacementRackScatter.java | 5 +++ .../TestSCMContainerPlacementRandom.java | 9 +++++ .../scm/node/TestPendingContainerTracker.java | 34 +++++++++++++++++ .../TestPipelinePlacementFactory.java | 3 ++ .../pipeline/TestPipelinePlacementPolicy.java | 1 + .../pipeline/TestRatisPipelineProvider.java | 5 +++ 17 files changed, 150 insertions(+), 58 deletions(-) diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMCommonPlacementPolicy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMCommonPlacementPolicy.java index 93940b7770ff..60443e013daf 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMCommonPlacementPolicy.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMCommonPlacementPolicy.java @@ -37,7 +37,6 @@ import org.apache.hadoop.hdds.conf.ConfigurationSource; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.MetadataStorageReportProto; -import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.StorageReportProto; import org.apache.hadoop.hdds.scm.container.ContainerReplica; import org.apache.hadoop.hdds.scm.container.placement.algorithms.ContainerPlacementStatusDefault; import org.apache.hadoop.hdds.scm.exceptions.SCMException; @@ -47,7 +46,6 @@ import org.apache.hadoop.hdds.scm.node.NodeManager; import org.apache.hadoop.hdds.scm.node.NodeStatus; import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException; -import org.apache.hadoop.ozone.container.common.volume.VolumeUsage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -273,7 +271,7 @@ public List filterNodesWithSpace(List nodes, int nodesRequired, long metadataSizeRequired, long dataSizeRequired) throws SCMException { List nodesWithSpace = nodes.stream().filter(d -> - hasEnoughSpace(d, metadataSizeRequired, dataSizeRequired)) + hasEnoughSpace(d, metadataSizeRequired, dataSizeRequired, nodeManager)) .collect(Collectors.toList()); if (nodesWithSpace.size() < nodesRequired) { @@ -298,29 +296,21 @@ public List filterNodesWithSpace(List nodes, */ public static boolean hasEnoughSpace(DatanodeDetails datanodeDetails, long metadataSizeRequired, - long dataSizeRequired) { + long dataSizeRequired, + NodeManager nodeManager) { Preconditions.checkArgument(datanodeDetails instanceof DatanodeInfo); - boolean enoughForData = false; boolean enoughForMeta = false; DatanodeInfo datanodeInfo = (DatanodeInfo) datanodeDetails; + // Data-space check: use PendingContainerTracker slot availability. + // This accounts for both current disk usage and in-flight allocations. if (dataSizeRequired > 0) { - for (StorageReportProto reportProto : datanodeInfo.getStorageReports()) { - if (VolumeUsage.getUsableSpace(reportProto) > dataSizeRequired) { - enoughForData = true; - break; - } + if (!nodeManager.hasAvailableSpace(datanodeInfo)) { + LOG.debug("Datanode {} has no available container slots.", datanodeDetails); + return false; } - } else { - enoughForData = true; - } - - if (!enoughForData) { - LOG.debug("Datanode {} has no volumes with enough space to allocate {} " + - "bytes for data.", datanodeDetails, dataSizeRequired); - return false; } if (metadataSizeRequired > 0) { @@ -539,7 +529,8 @@ public boolean isValidNode(DatanodeDetails datanodeDetails, return false; } NodeStatus nodeStatus = datanodeInfo.getNodeStatus(); - if (nodeStatus.isNodeWritable() && (hasEnoughSpace(datanodeInfo, metadataSizeRequired, dataSizeRequired))) { + if (nodeStatus.isNodeWritable() && (hasEnoughSpace(datanodeInfo, metadataSizeRequired, + dataSizeRequired, nodeManager))) { LOG.debug("Datanode {} is chosen. Required metadata size is {} and " + "required data size is {} and NodeStatus is {}", datanodeDetails, metadataSizeRequired, dataSizeRequired, nodeStatus); 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 6a2d92b64a5e..3da84aad3a73 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 @@ -195,6 +195,15 @@ default int getAllNodeCount() { */ boolean checkSpaceAndRecordAllocation(DatanodeInfo datanodeInfo, ContainerID containerID); + /** + * Returns true if the datanode has at least one available container slot considering + * in-flight allocations tracked by PendingContainerTracker. + * + * @param datanodeInfo the datanode to check + * @return true if at least one slot is free + */ + boolean hasAvailableSpace(DatanodeInfo datanodeInfo); + /** * Removes a pending container allocation from a datanode. * diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/PendingContainerTracker.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/PendingContainerTracker.java index 3821727ed971..f4e866972f6a 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/PendingContainerTracker.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/PendingContainerTracker.java @@ -208,12 +208,38 @@ public boolean checkSpaceAndRecordAllocation(DatanodeInfo datanodeInfo, Containe } /** - * Remove a pending container allocation from a specific DataNode. - * Removes from both current and previous windows. - * Called when container is confirmed. + * Returns true if the given datanode has at least one allocatable container slot + * available, accounting for pending in-flight allocations. * - * @param containerID The container to remove from pending + * Uses the same slot-counting logic as {@link TwoWindowBucket#checkSpaceAndAdd} + * + *

This method does not consume a slot — it is a read-only check intended + * for the placement policy + * + * @param datanodeInfo the datanode to check + * @return true if at least one container slot is available */ + public boolean hasAvailableSpace(DatanodeInfo datanodeInfo) { + Objects.requireNonNull(datanodeInfo, "datanodeInfo == null"); + List storageReports = datanodeInfo.getStorageReports(); + if (storageReports.isEmpty()) { + return false; + } + TwoWindowBucket bucket = datanodeInfo.getPendingContainerAllocations(); + bucket.rollIfNeeded(); + final int pendingCount = bucket.getCount(); + long allocatableCount = 0; + for (StorageReportProto report : storageReports) { + allocatableCount += VolumeUsage.getUsableSpace(report) / maxContainerSize; + if (allocatableCount > pendingCount) { + return true; + } + } + LOG.debug("Datanode {} has no available container slots. Pending: {}, Allocatable: {}", + datanodeInfo.getID(), pendingCount, allocatableCount); + return false; + } + public void removePendingAllocation(TwoWindowBucket bucket, ContainerID containerID) { Objects.requireNonNull(containerID, "containerID == null"); 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 4ead7b1b1a39..40a7ec943a56 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 @@ -213,7 +213,7 @@ public SCMNodeManager( ScmConfigKeys.OZONE_SCM_PIPELINE_OWNER_CONTAINER_COUNT_DEFAULT); this.scmContext = scmContext; this.sendCommandNotifyMap = new HashMap<>(); - this.nonWritableNodeFilter = new NonWritableNodeFilter(conf); + this.nonWritableNodeFilter = new NonWritableNodeFilter(conf, this); } @Override @@ -1084,6 +1084,11 @@ public boolean checkSpaceAndRecordAllocation(DatanodeInfo datanodeInfo, Containe return pendingContainerTracker.checkSpaceAndRecordAllocation(datanodeInfo, containerID); } + @Override + public boolean hasAvailableSpace(DatanodeInfo datanodeInfo) { + return pendingContainerTracker.hasAvailableSpace(datanodeInfo); + } + @Override public void removePendingAllocationForDatanode(DatanodeInfo datanodeInfo, ContainerID containerID) { pendingContainerTracker.removePendingAllocation( @@ -1468,8 +1473,9 @@ static class NonWritableNodeFilter implements Predicate { private final long blockSize; private final long minRatisVolumeSizeBytes; private final long containerSize; + private final NodeManager nodeManager; - NonWritableNodeFilter(ConfigurationSource conf) { + NonWritableNodeFilter(ConfigurationSource conf, NodeManager nodeManager) { blockSize = (long) conf.getStorageSize( OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE, OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE_DEFAULT, @@ -1482,12 +1488,13 @@ static class NonWritableNodeFilter implements Predicate { ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE, ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE_DEFAULT, StorageUnit.BYTES); + this.nodeManager = nodeManager; } @Override public boolean test(DatanodeInfo dn) { return !dn.getNodeStatus().isNodeWritable() - || (!hasEnoughSpace(dn, minRatisVolumeSizeBytes, containerSize) + || (!hasEnoughSpace(dn, minRatisVolumeSizeBytes, containerSize, nodeManager) && !hasEnoughCommittedVolumeSpace(dn)); } diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineProvider.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineProvider.java index b1b2d7349066..0248d012eade 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineProvider.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineProvider.java @@ -87,7 +87,8 @@ List pickNodesNotUsed(REPLICATION_CONFIG replicationConfig, int nodesRequired = replicationConfig.getRequiredNodes(); List healthyDNs = pickAllNodesNotUsed(replicationConfig); List healthyDNsWithSpace = healthyDNs.stream() - .filter(dn -> SCMCommonPlacementPolicy.hasEnoughSpace(dn, metadataSizeRequired, dataSizeRequired)) + .filter(dn -> SCMCommonPlacementPolicy.hasEnoughSpace( + dn, metadataSizeRequired, dataSizeRequired, nodeManager)) .limit(nodesRequired) .collect(Collectors.toList()); diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/TestSCMCommonPlacementPolicy.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/TestSCMCommonPlacementPolicy.java index 5ecd3fa1c74c..9290e8d83bed 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/TestSCMCommonPlacementPolicy.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/TestSCMCommonPlacementPolicy.java @@ -17,7 +17,6 @@ package org.apache.hadoop.hdds.scm; -import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.StorageTypeProto.DISK; import static org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto.State.CLOSED; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -474,7 +473,7 @@ protected List chooseDatanodesInternal( } @Test - public void testDatanodeIsInvalidInCaseOfIncreasingCommittedBytes() { + public void testDatanodeIsInvalidWhenNoSlotsAvailable() { NodeManager nodeMngr = mock(NodeManager.class); final DatanodeID datanodeID = DatanodeID.of(UUID.randomUUID()); DummyPlacementPolicy placementPolicy = @@ -488,43 +487,19 @@ public void testDatanodeIsInvalidInCaseOfIncreasingCommittedBytes() { when(datanodeInfo.getNodeStatus()).thenReturn(nodeStatus); when(nodeMngr.getNode(eq(datanodeID))).thenReturn(datanodeInfo); - // capacity = 200000, used = 90000, remaining = 101000, committed = 500 - StorageContainerDatanodeProtocolProtos.StorageReportProto storageReport1 = - HddsTestUtils.createStorageReport(DatanodeID.randomID(), "/data/hdds", - 200000, 90000, 101000, DISK).toBuilder() - .setCommitted(500) - .setFreeSpaceToSpare(10000) - .build(); - // capacity = 200000, used = 90000, remaining = 101000, committed = 1000 - StorageContainerDatanodeProtocolProtos.StorageReportProto storageReport2 = - HddsTestUtils.createStorageReport(DatanodeID.randomID(), "/data/hdds", - 200000, 90000, 101000, DISK).toBuilder() - .setCommitted(1000) - .setFreeSpaceToSpare(100000) - .build(); StorageContainerDatanodeProtocolProtos.MetadataStorageReportProto metaReport = HddsTestUtils.createMetadataStorageReport("/data/metadata", 200); - when(datanodeInfo.getStorageReports()) - .thenReturn(Collections.singletonList(storageReport1)) - .thenReturn(Collections.singletonList(storageReport2)); when(datanodeInfo.getMetadataStorageReports()) .thenReturn(Collections.singletonList(metaReport)); - - // 500 committed bytes: - // - // 101000 500 - // | | - // (remaining - committed) > Math.max(4000, freeSpaceToSpare) - // | - // 100000 - // - // Summary: 101000 - 500 > 100000 == true + // Space check now uses PendingContainerTracker.hasAvailableSpace: + // slot available → isValidNode returns true + when(nodeMngr.hasAvailableSpace(datanodeInfo)).thenReturn(true); assertTrue(placementPolicy.isValidNode(datanodeDetails, 100, 4000)); - // 1000 committed bytes: - // Summary: 101000 - 1000 > 100000 == false + // No slot available (all pending) → isValidNode returns false + when(nodeMngr.hasAvailableSpace(datanodeInfo)).thenReturn(false); assertFalse(placementPolicy.isValidNode(datanodeDetails, 100, 4000)); } diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java index 1b8c4bbf384d..bd6d5cc1d93a 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java @@ -461,6 +461,14 @@ public boolean checkSpaceAndRecordAllocation(DatanodeInfo datanodeInfo, Containe return pendingContainerTracker.checkSpaceAndRecordAllocation(datanodeInfo, containerID); } + @Override + public boolean hasAvailableSpace(DatanodeInfo datanodeInfo) { + if (datanodeInfo == null) { + return false; + } + return pendingContainerTracker.hasAvailableSpace(datanodeInfo); + } + @Override public void removePendingAllocationForDatanode(DatanodeInfo datanodeInfo, ContainerID containerID) { if (datanodeInfo != null) { @@ -955,6 +963,11 @@ public PendingContainerTracker getPendingContainerTracker() { return pendingContainerTracker; } + public void setPendingContainerMaxSize(long maxContainerSize) { + this.pendingContainerTracker = new PendingContainerTracker(maxContainerSize, + HddsTestUtils.ROLL_INTERVAL_MS_DEFAULT, null); + } + @Override public long getLastHeartbeat(DatanodeDetails datanodeDetails) { return -1; diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/SimpleMockNodeManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/SimpleMockNodeManager.java index e47b979b4750..ec481f079736 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/SimpleMockNodeManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/SimpleMockNodeManager.java @@ -256,6 +256,11 @@ public boolean checkSpaceAndRecordAllocation(DatanodeInfo datanodeInfo, Containe return true; } + @Override + public boolean hasAvailableSpace(DatanodeInfo datanodeInfo) { + return true; + } + @Override public void removePendingAllocationForDatanode(DatanodeInfo datanodeInfo, ContainerID containerID) { } diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestContainerPlacementFactory.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestContainerPlacementFactory.java index ed4e96b8de56..5e6f0ef39592 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestContainerPlacementFactory.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestContainerPlacementFactory.java @@ -26,6 +26,7 @@ import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -146,6 +147,7 @@ public void testRackAwarePolicy() throws IOException { when(nodeManager.getNode(dn.getID())) .thenReturn(dn); } + when(nodeManager.hasAvailableSpace(any(DatanodeInfo.class))).thenReturn(true); PlacementPolicy policy = ContainerPlacementPolicyFactory .getPolicy(conf, nodeManager, cluster, true, diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementCapacity.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementCapacity.java index 1fb3f53504d3..b86bf3e58d04 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementCapacity.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementCapacity.java @@ -118,6 +118,10 @@ public void chooseDatanodes() throws SCMException { .filter(dn -> dn.getID().equals(invocation.getArgument(0))) .findFirst() .orElse(null)); + when(mockNodeManager.hasAvailableSpace(any(DatanodeInfo.class))).thenAnswer(invocation -> { + DatanodeInfo di = invocation.getArgument(0); + return di.getStorageReports().stream().anyMatch(r -> r.getRemaining() >= 15L); + }); SCMContainerPlacementCapacity scmContainerPlacementRandom = new SCMContainerPlacementCapacity(mockNodeManager, conf, null, true, diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackAware.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackAware.java index dd068f55cdfe..5b1961df1071 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackAware.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackAware.java @@ -32,6 +32,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assumptions.assumeTrue; +import static org.mockito.Mockito.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -180,6 +181,7 @@ private void setup(int datanodeCount) { } when(nodeManager.getClusterNetworkTopologyMap()) .thenReturn(cluster); + when(nodeManager.hasAvailableSpace(any(DatanodeInfo.class))).thenReturn(true); // create placement policy instances policy = new SCMContainerPlacementRackAware( diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackScatter.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackScatter.java index e015b93c1e31..cbbfae952980 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackScatter.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackScatter.java @@ -33,6 +33,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assumptions.assumeTrue; +import static org.mockito.Mockito.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -249,6 +250,10 @@ private void createMocksAndUpdateStorageReports(int datanodeCount) { } when(nodeManager.getClusterNetworkTopologyMap()) .thenReturn(cluster); + when(nodeManager.hasAvailableSpace(any(DatanodeInfo.class))).thenAnswer(invocation -> { + DatanodeInfo di = invocation.getArgument(0); + return di.getStorageReports().stream().anyMatch(r -> r.getRemaining() > 1L); + }); // create placement policy instances policy = new SCMContainerPlacementRackScatter( diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRandom.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRandom.java index 47602a385fd6..b8c1a0f4c4f0 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRandom.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRandom.java @@ -22,6 +22,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -89,6 +90,10 @@ public void chooseDatanodes() throws SCMException { NodeManager mockNodeManager = mock(NodeManager.class); when(mockNodeManager.getNodes(NodeStatus.inServiceHealthy())) .thenReturn(new ArrayList<>(datanodes)); + when(mockNodeManager.hasAvailableSpace(any(DatanodeInfo.class))).thenAnswer(invocation -> { + DatanodeInfo di = invocation.getArgument(0); + return di.getStorageReports().stream().anyMatch(r -> r.getRemaining() >= 15L); + }); SCMContainerPlacementRandom scmContainerPlacementRandom = new SCMContainerPlacementRandom(mockNodeManager, conf, null, true, @@ -209,6 +214,10 @@ public void testIsValidNode() throws SCMException { .thenReturn(datanodes.get(1)); when(mockNodeManager.getNode(datanodes.get(2).getID())) .thenReturn(datanodes.get(2)); + when(mockNodeManager.hasAvailableSpace(any(DatanodeInfo.class))).thenAnswer(invocation -> { + DatanodeInfo di = invocation.getArgument(0); + return di.getStorageReports().stream().anyMatch(r -> r.getRemaining() >= 15L); + }); SCMContainerPlacementRandom scmContainerPlacementRandom = new SCMContainerPlacementRandom(mockNodeManager, conf, null, true, diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestPendingContainerTracker.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestPendingContainerTracker.java index b486715deb01..a2c6882eada1 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestPendingContainerTracker.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestPendingContainerTracker.java @@ -376,6 +376,40 @@ public void testMultiVolumeWithCommittedBytes() { assertFalse(tracker.checkSpaceAndRecordAllocation(dnInfo, containers.get(1))); } + /** + * Pending in-flight replications recorded via checkSpaceAndRecordAllocation count against + * slots, same as write-path containers. hasAvailableSpace reflects the combined total. + */ + @Test + public void testInFlightReplicationCountsAgainstAvailableSlots() { + long containerSize = MAX_CONTAINER_SIZE; + DatanodeInfo dnInfo = datanodes.get(0); + + // Two slots of usable space + List twoSlotReports = new ArrayList<>(); + twoSlotReports.add(createStorageReport(dnInfo, 10 * containerSize, 2 * containerSize, 0)); + dnInfo.updateStorageReports(twoSlotReports); + + assertTrue(tracker.hasAvailableSpace(dnInfo)); // 2 slots free + assertTrue(tracker.checkSpaceAndRecordAllocation(dnInfo, containers.get(0))); // slot 1 used + assertTrue(tracker.hasAvailableSpace(dnInfo)); // 1 slot free + assertTrue(tracker.checkSpaceAndRecordAllocation(dnInfo, containers.get(1))); // slot 2 used + assertFalse(tracker.hasAvailableSpace(dnInfo)); // 0 slots free + assertFalse(tracker.checkSpaceAndRecordAllocation(dnInfo, containers.get(2))); // rejected + } + + /** + * hasAvailableSpace on a DN with no storage reports returns false. + */ + @Test + public void testHasAvailableSpaceWithNoStorageReports() { + DatanodeInfo emptyDn = new DatanodeInfo( + MockDatanodeDetails.randomLocalDatanodeDetails(), NodeStatus.inServiceHealthy(), null, + HddsTestUtils.ROLL_INTERVAL_MS_DEFAULT); + // No storage reports set + assertFalse(tracker.hasAvailableSpace(emptyDn)); + } + private StorageReportProto createStorageReport(DatanodeInfo dn, long capacity, long remaining, long committed) { return HddsTestUtils.createStorageReports(dn.getID(), capacity, remaining, committed).get(0); } diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementFactory.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementFactory.java index 84afb74f0a5a..3ddb0361e08a 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementFactory.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementFactory.java @@ -27,6 +27,8 @@ import static org.junit.jupiter.api.Assertions.assertNotSame; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.when; @@ -128,6 +130,7 @@ private void setupRacks(int datanodeCount, int nodesPerRack, when(nodeManager.getNode(dn.getID())) .thenReturn(dn); } + doReturn(true).when(nodeManager).hasAvailableSpace(any(DatanodeInfo.class)); DBStore dbStore = DBStoreBuilder.createDBStore(conf, SCMDBDefinition.get()); SCMHAManager scmhaManager = SCMHAManagerStub.getInstance(true); diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementPolicy.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementPolicy.java index 90b7747350bf..d76bf22b3fe7 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementPolicy.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementPolicy.java @@ -258,6 +258,7 @@ public void testChooseNodeNotEnoughSpace() throws IOException { "the space requirement"; // A huge container size + localNodeManager.setPendingContainerMaxSize(200 * OzoneConsts.TB); SCMException ex = assertThrows(SCMException.class, () -> localPlacementPolicy.chooseDatanodes(new ArrayList<>(datanodes.size()), diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java index ca9d1f5a6c3c..bf352f2051ef 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java @@ -41,6 +41,7 @@ import org.apache.hadoop.hdds.client.RatisReplicationConfig; import org.apache.hadoop.hdds.client.ReplicationConfig; import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.conf.StorageUnit; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.DatanodeID; import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; @@ -97,6 +98,10 @@ public void init(int maxPipelinePerNode, OzoneConfiguration conf, File dir) thro dbStore = DBStoreBuilder.createDBStore(conf, SCMDBDefinition.get()); nodeManager = new MockNodeManager(true, nodeCount); nodeManager.setNumPipelinePerDatanode(maxPipelinePerNode); + long containerSize = (long) conf.getStorageSize( + ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE, + ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE_DEFAULT, StorageUnit.BYTES); + nodeManager.setPendingContainerMaxSize(containerSize); SCMHAManager scmhaManager = SCMHAManagerStub.getInstance(true); conf.setInt(ScmConfigKeys.OZONE_DATANODE_PIPELINE_LIMIT, maxPipelinePerNode); From 11d32b9d035982396b75645b580e4b16390c4995 Mon Sep 17 00:00:00 2001 From: ashishk Date: Wed, 17 Jun 2026 17:48:13 +0530 Subject: [PATCH 2/4] Fix review comments --- .../hdds/scm/SCMCommonPlacementPolicy.java | 28 +++++++++------ .../scm/node/PendingContainerTracker.java | 27 ++++++++++---- .../hadoop/hdds/scm/node/SCMNodeManager.java | 36 ++++++++++++------- .../hdds/scm/pipeline/PipelineProvider.java | 2 +- 4 files changed, 62 insertions(+), 31 deletions(-) diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMCommonPlacementPolicy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMCommonPlacementPolicy.java index 60443e013daf..861578255857 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMCommonPlacementPolicy.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMCommonPlacementPolicy.java @@ -271,7 +271,7 @@ public List filterNodesWithSpace(List nodes, int nodesRequired, long metadataSizeRequired, long dataSizeRequired) throws SCMException { List nodesWithSpace = nodes.stream().filter(d -> - hasEnoughSpace(d, metadataSizeRequired, dataSizeRequired, nodeManager)) + hasEnoughSpace(d, metadataSizeRequired, nodeManager)) .collect(Collectors.toList()); if (nodesWithSpace.size() < nodesRequired) { @@ -289,14 +289,21 @@ public List filterNodesWithSpace(List nodes, } /** - * Returns true if this node has enough space to meet our requirement. + * Returns true if this node has enough space to satisfy the placement request. * - * @param datanodeDetails DatanodeDetails - * @return true if we have enough space. + *

Data-space is checked via {@link NodeManager#hasAvailableSpace}, which + * delegates to {@link org.apache.hadoop.hdds.scm.node.PendingContainerTracker} + * and accounts for both current disk usage and in-flight allocations. + * The check always uses {@code maxContainerSize} as the unit of allocation, + * regardless of the actual container's used bytes. + * + * @param datanodeDetails the datanode to evaluate + * @param metadataSizeRequired minimum metadata volume space required in bytes + * @param nodeManager used to check slot availability via PendingContainerTracker + * @return true if the datanode has both an available data slot and enough metadata space */ public static boolean hasEnoughSpace(DatanodeDetails datanodeDetails, long metadataSizeRequired, - long dataSizeRequired, NodeManager nodeManager) { Preconditions.checkArgument(datanodeDetails instanceof DatanodeInfo); @@ -306,11 +313,10 @@ public static boolean hasEnoughSpace(DatanodeDetails datanodeDetails, // Data-space check: use PendingContainerTracker slot availability. // This accounts for both current disk usage and in-flight allocations. - if (dataSizeRequired > 0) { - if (!nodeManager.hasAvailableSpace(datanodeInfo)) { - LOG.debug("Datanode {} has no available container slots.", datanodeDetails); - return false; - } + // Always slot-based (maxContainerSize unit). + if (!nodeManager.hasAvailableSpace(datanodeInfo)) { + LOG.debug("Datanode {} has no available container slots.", datanodeDetails); + return false; } if (metadataSizeRequired > 0) { @@ -530,7 +536,7 @@ public boolean isValidNode(DatanodeDetails datanodeDetails, } NodeStatus nodeStatus = datanodeInfo.getNodeStatus(); if (nodeStatus.isNodeWritable() && (hasEnoughSpace(datanodeInfo, metadataSizeRequired, - dataSizeRequired, nodeManager))) { + nodeManager))) { LOG.debug("Datanode {} is chosen. Required metadata size is {} and " + "required data size is {} and NodeStatus is {}", datanodeDetails, metadataSizeRequired, dataSizeRequired, nodeStatus); diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/PendingContainerTracker.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/PendingContainerTracker.java index f4e866972f6a..9af6c76bc851 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/PendingContainerTracker.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/PendingContainerTracker.java @@ -105,7 +105,11 @@ synchronized void rollIfNeeded() { previousWindow.clear(); currentWindow.clear(); lastRollTime = now; - LOG.debug("Double roll interval elapsed ({}ms): dropped {} pending containers", elapsed, dropped); + if (dropped > 0) { + LOG.warn("PendingContainerTracker: force-dropped {} unconfirmed pending containers " + + "on DN {} after {}ms (2x rollInterval). " + + "Container reports may have been lost.", dropped, datanodeID, elapsed); + } } else if (elapsed >= rollIntervalMs) { previousWindow.clear(); final Set tmp = previousWindow; @@ -156,7 +160,8 @@ synchronized boolean checkSpaceAndAdd( final int pendingAllocationCount = getCount(); long allocatableCount = 0; for (StorageReportProto report : storageReports) { - final long allocatableCountOnThisDisk = VolumeUsage.getUsableSpace(report) / maxContainerSize; + final long allocatableCountOnThisDisk = + Math.max(0L, VolumeUsage.getUsableSpace(report)) / maxContainerSize; allocatableCount += allocatableCountOnThisDisk; if (allocatableCount > pendingAllocationCount) { final boolean added = currentWindow.add(containerID); @@ -211,10 +216,9 @@ public boolean checkSpaceAndRecordAllocation(DatanodeInfo datanodeInfo, Containe * Returns true if the given datanode has at least one allocatable container slot * available, accounting for pending in-flight allocations. * - * Uses the same slot-counting logic as {@link TwoWindowBucket#checkSpaceAndAdd} - * - *

This method does not consume a slot — it is a read-only check intended - * for the placement policy + *

Slot availability is based on {@code maxContainerSize}: a slot exists for each + * {@code maxContainerSize}-worth of usable space on any volume. This read-only check + * is intended for the placement policy and does not consume a slot. * * @param datanodeInfo the datanode to check * @return true if at least one container slot is available @@ -230,16 +234,25 @@ public boolean hasAvailableSpace(DatanodeInfo datanodeInfo) { final int pendingCount = bucket.getCount(); long allocatableCount = 0; for (StorageReportProto report : storageReports) { - allocatableCount += VolumeUsage.getUsableSpace(report) / maxContainerSize; + allocatableCount += Math.max(0L, VolumeUsage.getUsableSpace(report)) / maxContainerSize; if (allocatableCount > pendingCount) { return true; } } LOG.debug("Datanode {} has no available container slots. Pending: {}, Allocatable: {}", datanodeInfo.getID(), pendingCount, allocatableCount); + if (metrics != null) { + metrics.incNumSkippedFullNodeContainerAllocation(); + } return false; } + /** + * Remove pending allocation from the bucket for the given container. + * + * @param bucket TWO window bucket of the datanode + * @param containerID containerID + */ public void removePendingAllocation(TwoWindowBucket bucket, ContainerID containerID) { Objects.requireNonNull(containerID, "containerID == null"); 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 40a7ec943a56..27f5099b5847 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 @@ -22,7 +22,6 @@ import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState.IN_SERVICE; import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState.HEALTHY; import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState.HEALTHY_READONLY; -import static org.apache.hadoop.hdds.scm.SCMCommonPlacementPolicy.hasEnoughSpace; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; @@ -213,7 +212,7 @@ public SCMNodeManager( ScmConfigKeys.OZONE_SCM_PIPELINE_OWNER_CONTAINER_COUNT_DEFAULT); this.scmContext = scmContext; this.sendCommandNotifyMap = new HashMap<>(); - this.nonWritableNodeFilter = new NonWritableNodeFilter(conf, this); + this.nonWritableNodeFilter = new NonWritableNodeFilter(conf, pendingContainerTracker); } @Override @@ -1472,10 +1471,9 @@ static class NonWritableNodeFilter implements Predicate { private final long blockSize; private final long minRatisVolumeSizeBytes; - private final long containerSize; - private final NodeManager nodeManager; + private final PendingContainerTracker tracker; - NonWritableNodeFilter(ConfigurationSource conf, NodeManager nodeManager) { + NonWritableNodeFilter(ConfigurationSource conf, PendingContainerTracker tracker) { blockSize = (long) conf.getStorageSize( OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE, OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE_DEFAULT, @@ -1484,18 +1482,32 @@ static class NonWritableNodeFilter implements Predicate { ScmConfigKeys.OZONE_DATANODE_RATIS_VOLUME_FREE_SPACE_MIN, ScmConfigKeys.OZONE_DATANODE_RATIS_VOLUME_FREE_SPACE_MIN_DEFAULT, StorageUnit.BYTES); - containerSize = (long) conf.getStorageSize( - ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE, - ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE_DEFAULT, - StorageUnit.BYTES); - this.nodeManager = nodeManager; + this.tracker = tracker; } @Override public boolean test(DatanodeInfo dn) { return !dn.getNodeStatus().isNodeWritable() - || (!hasEnoughSpace(dn, minRatisVolumeSizeBytes, containerSize, nodeManager) - && !hasEnoughCommittedVolumeSpace(dn)); + || (!hasEnoughSpaceForNode(dn) && !hasEnoughCommittedVolumeSpace(dn)); + } + + /** + * Returns true if the datanode has both an available data slot (via + * {@link PendingContainerTracker}) and sufficient metadata volume space. + */ + private boolean hasEnoughSpaceForNode(DatanodeInfo dn) { + if (!tracker.hasAvailableSpace(dn)) { + return false; + } + if (minRatisVolumeSizeBytes <= 0) { + return true; + } + for (StorageReportProto report : dn.getStorageReports()) { + if (report.getRemaining() > minRatisVolumeSizeBytes) { + return true; + } + } + return false; } /** diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineProvider.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineProvider.java index 0248d012eade..75a7c05554c7 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineProvider.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineProvider.java @@ -88,7 +88,7 @@ List pickNodesNotUsed(REPLICATION_CONFIG replicationConfig, List healthyDNs = pickAllNodesNotUsed(replicationConfig); List healthyDNsWithSpace = healthyDNs.stream() .filter(dn -> SCMCommonPlacementPolicy.hasEnoughSpace( - dn, metadataSizeRequired, dataSizeRequired, nodeManager)) + dn, metadataSizeRequired, nodeManager)) .limit(nodesRequired) .collect(Collectors.toList()); From 8303848671c93e80338885ae4342d9b2f76e52fc Mon Sep 17 00:00:00 2001 From: ashishk Date: Wed, 17 Jun 2026 21:41:19 +0530 Subject: [PATCH 3/4] Fix comments --- .../hadoop/hdds/scm/SCMCommonPlacementPolicy.java | 11 +++++------ .../apache/hadoop/hdds/scm/node/SCMNodeManager.java | 5 +++-- .../hdds/scm/pipeline/PipelinePlacementPolicy.java | 2 +- 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMCommonPlacementPolicy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMCommonPlacementPolicy.java index 861578255857..0deece79c6d8 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMCommonPlacementPolicy.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMCommonPlacementPolicy.java @@ -250,7 +250,7 @@ protected List chooseDatanodesInternal( } return filterNodesWithSpace(healthyNodes, nodesRequired, - metadataSizeRequired, dataSizeRequired); + metadataSizeRequired); } /** @@ -268,7 +268,7 @@ protected boolean usedNodesPassed(List list) { } public List filterNodesWithSpace(List nodes, - int nodesRequired, long metadataSizeRequired, long dataSizeRequired) + int nodesRequired, long metadataSizeRequired) throws SCMException { List nodesWithSpace = nodes.stream().filter(d -> hasEnoughSpace(d, metadataSizeRequired, nodeManager)) @@ -276,10 +276,9 @@ public List filterNodesWithSpace(List nodes, if (nodesWithSpace.size() < nodesRequired) { String msg = String.format("Unable to find enough nodes that meet the " + - "space requirement of %d bytes for metadata and %d bytes for " + - "data in healthy node set. Required %d. Found %d.", - metadataSizeRequired, dataSizeRequired, nodesRequired, - nodesWithSpace.size()); + "space requirement of %d bytes for metadata and an available " + + "container slot in healthy node set. Required %d. Found %d.", + metadataSizeRequired, nodesRequired, nodesWithSpace.size()); LOG.warn(msg); throw new SCMException(msg, SCMException.ResultCodes.FAILED_TO_FIND_NODES_WITH_SPACE); 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 27f5099b5847..5ff1fe59312e 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 @@ -62,6 +62,7 @@ import org.apache.hadoop.hdds.protocol.proto.HddsProtos.StorageTypeProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandQueueReportProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.LayoutVersionProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.MetadataStorageReportProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeReportProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReportsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto; @@ -1493,7 +1494,7 @@ public boolean test(DatanodeInfo dn) { /** * Returns true if the datanode has both an available data slot (via - * {@link PendingContainerTracker}) and sufficient metadata volume space. + * {@link PendingContainerTracker}) and sufficient Ratis metadata volume space. */ private boolean hasEnoughSpaceForNode(DatanodeInfo dn) { if (!tracker.hasAvailableSpace(dn)) { @@ -1502,7 +1503,7 @@ private boolean hasEnoughSpaceForNode(DatanodeInfo dn) { if (minRatisVolumeSizeBytes <= 0) { return true; } - for (StorageReportProto report : dn.getStorageReports()) { + for (MetadataStorageReportProto report : dn.getMetadataStorageReports()) { if (report.getRemaining() > minRatisVolumeSizeBytes) { return true; } diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java index 8e1437d474c3..615d466f6e4c 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java @@ -156,7 +156,7 @@ List filterViableNodes( } healthyNodes = filterNodesWithSpace(healthyNodes, nodesRequired, - metadataSizeRequired, dataSizeRequired); + metadataSizeRequired); boolean multipleRacks = multipleRacksAvailable(healthyNodes); int excludedNodesSize = 0; if (excludedNodes != null) { From 91be97b59f45312c1c46167bd2b0324dadee442c Mon Sep 17 00:00:00 2001 From: ashishk Date: Fri, 19 Jun 2026 20:49:54 +0530 Subject: [PATCH 4/4] Fix review comments --- .../scm/node/PendingContainerTracker.java | 13 +++++--- .../hadoop/hdds/scm/node/SCMNodeManager.java | 3 ++ .../hadoop/hdds/scm/node/SCMNodeMetrics.java | 10 ++++++ .../hdds/scm/container/MockNodeManager.java | 3 -- .../scm/node/TestPendingContainerTracker.java | 31 +++++++++++++++++++ 5 files changed, 52 insertions(+), 8 deletions(-) diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/PendingContainerTracker.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/PendingContainerTracker.java index 9af6c76bc851..229dc5fad929 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/PendingContainerTracker.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/PendingContainerTracker.java @@ -160,6 +160,9 @@ synchronized boolean checkSpaceAndAdd( final int pendingAllocationCount = getCount(); long allocatableCount = 0; for (StorageReportProto report : storageReports) { + if (report.hasFailed() && report.getFailed()) { + continue; + } final long allocatableCountOnThisDisk = Math.max(0L, VolumeUsage.getUsableSpace(report)) / maxContainerSize; allocatableCount += allocatableCountOnThisDisk; @@ -217,8 +220,8 @@ public boolean checkSpaceAndRecordAllocation(DatanodeInfo datanodeInfo, Containe * available, accounting for pending in-flight allocations. * *

Slot availability is based on {@code maxContainerSize}: a slot exists for each - * {@code maxContainerSize}-worth of usable space on any volume. This read-only check - * is intended for the placement policy and does not consume a slot. + * {@code maxContainerSize}-worth of usable space on any volume. This check is intended for the placement policy. + * This rolls expired-window entries but does not consume a slot. * * @param datanodeInfo the datanode to check * @return true if at least one container slot is available @@ -234,6 +237,9 @@ public boolean hasAvailableSpace(DatanodeInfo datanodeInfo) { final int pendingCount = bucket.getCount(); long allocatableCount = 0; for (StorageReportProto report : storageReports) { + if (report.hasFailed() && report.getFailed()) { + continue; + } allocatableCount += Math.max(0L, VolumeUsage.getUsableSpace(report)) / maxContainerSize; if (allocatableCount > pendingCount) { return true; @@ -241,9 +247,6 @@ public boolean hasAvailableSpace(DatanodeInfo datanodeInfo) { } LOG.debug("Datanode {} has no available container slots. Pending: {}, Allocatable: {}", datanodeInfo.getID(), pendingCount, allocatableCount); - if (metrics != null) { - metrics.incNumSkippedFullNodeContainerAllocation(); - } return false; } 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 5ff1fe59312e..f492b2e3dcab 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 @@ -1435,7 +1435,9 @@ private void nodeSpaceStatistics(Map nodeStatics) { long capacityByte = 0; long scmUsedByte = 0; long remainingByte = 0; + long totalPending = 0; for (DatanodeInfo dni : nodeStateManager.getAllNodes()) { + totalPending += dni.getPendingContainerAllocations().getCount(); List storageReports = dni.getStorageReports(); if (storageReports != null && !storageReports.isEmpty()) { for (StorageReportProto storageReport : storageReports) { @@ -1445,6 +1447,7 @@ private void nodeSpaceStatistics(Map nodeStatics) { } } } + metrics.setTotalPendingContainerSlots(totalPending); long nonScmUsedByte = capacityByte - scmUsedByte - remainingByte; if (nonScmUsedByte < 0) { diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeMetrics.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeMetrics.java index 0b90247ae8ab..ee78ea3dd2f6 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeMetrics.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeMetrics.java @@ -30,6 +30,7 @@ import org.apache.hadoop.metrics2.lib.Interns; import org.apache.hadoop.metrics2.lib.MetricsRegistry; import org.apache.hadoop.metrics2.lib.MutableCounterLong; +import org.apache.hadoop.metrics2.lib.MutableGaugeLong; import org.apache.hadoop.ozone.OzoneConsts; import org.apache.hadoop.util.StringUtils; @@ -54,6 +55,7 @@ public final class SCMNodeMetrics implements MetricsSource { private @Metric MutableCounterLong numPendingContainersAdded; private @Metric MutableCounterLong numPendingContainersRemoved; private @Metric MutableCounterLong numSkippedFullNodeContainerAllocation; + private @Metric MutableGaugeLong totalPendingContainerSlots; private final MetricsRegistry registry; private final NodeManagerMXBean managerMXBean; @@ -148,6 +150,14 @@ void incNumSkippedFullNodeContainerAllocation() { numSkippedFullNodeContainerAllocation.incr(); } + void setTotalPendingContainerSlots(long value) { + totalPendingContainerSlots.set(value); + } + + public long getTotalPendingContainerSlots() { + return totalPendingContainerSlots.value(); + } + /** * Get aggregated counter and gauge metrics. */ diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java index bd6d5cc1d93a..b9aa2930f013 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java @@ -463,9 +463,6 @@ public boolean checkSpaceAndRecordAllocation(DatanodeInfo datanodeInfo, Containe @Override public boolean hasAvailableSpace(DatanodeInfo datanodeInfo) { - if (datanodeInfo == null) { - return false; - } return pendingContainerTracker.hasAvailableSpace(datanodeInfo); } diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestPendingContainerTracker.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestPendingContainerTracker.java index a2c6882eada1..ff789aba141b 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestPendingContainerTracker.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestPendingContainerTracker.java @@ -23,6 +23,7 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.StorageReportProto; @@ -410,7 +411,37 @@ public void testHasAvailableSpaceWithNoStorageReports() { assertFalse(tracker.hasAvailableSpace(emptyDn)); } + /** + * A failed volume has remaining=0 by DN convention, but the tracker should + * explicitly skip it (report.getFailed() == true) so a stale non-zero + * remaining value on a failed volume can never grant spurious slots. + */ + @Test + public void testFailedVolumeNotCountedAsAllocatableSlot() { + StorageReportProto failed = createFailedStorageReport(dn1); + StorageReportProto healthy = createStorageReport(dn1, + 10 * MAX_CONTAINER_SIZE, MAX_CONTAINER_SIZE, 0); // 1 real slot + dn1.updateStorageReports(new ArrayList<>(Arrays.asList(failed, healthy))); + assertTrue(tracker.hasAvailableSpace(dn1)); // healthy vol → 1 slot + assertTrue(tracker.checkSpaceAndRecordAllocation(dn1, container1)); // consumes it + assertFalse(tracker.hasAvailableSpace(dn1)); // 0 slots left + assertFalse(tracker.checkSpaceAndRecordAllocation(dn1, container2)); // rejected + } + + @Test + public void testAllVolumesFailedReturnsFalse() { + dn1.updateStorageReports(new ArrayList<>(Arrays.asList(( + createFailedStorageReport(dn1)), + createFailedStorageReport(dn1)))); + assertFalse(tracker.hasAvailableSpace(dn1)); + assertFalse(tracker.checkSpaceAndRecordAllocation(dn1, container1)); + } + private StorageReportProto createStorageReport(DatanodeInfo dn, long capacity, long remaining, long committed) { return HddsTestUtils.createStorageReports(dn.getID(), capacity, remaining, committed).get(0); } + + private StorageReportProto createFailedStorageReport(DatanodeInfo dn) { + return HddsTestUtils.createStorageReport(dn.getID(), "", 0, 0, 0, null, true); + } }