You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by de...@apache.org on 2026/08/10 07:34:19 UTC
(ozone) 06/10: HDDS-15250. PipelineProvider support create pipeline by StorageTier (#10410)
This is an automated email from the ASF dual-hosted git repository.
devmadhuu pushed a commit to branch HDDS-11233
in repository https://gitbox.apache.org/repos/asf/ozone.git
commit 9d6319c30dfa3b185ff4fdf3643906bef419409e
Author: XiChen <32...@users.noreply.github.com>
AuthorDate: Wed Jun 17 20:04:04 2026 +0800
HDDS-15250. PipelineProvider support create pipeline by StorageTier (#10410)
Co-authored-by: Xi Chen <ch...@apache.org>
---
.../org/apache/hadoop/hdds/client/StorageTier.java | 82 +++--------
.../apache/hadoop/hdds/client/StorageTierUtil.java | 66 +++++++++
.../hadoop/hdds/client/StorageTypeUtils.java | 5 +
.../hadoop/hdds/scm/exceptions/SCMException.java | 4 +-
.../common/impl/StorageLocationReport.java | 32 +----
.../interface-client/src/main/proto/hdds.proto | 1 +
.../src/main/proto/ScmServerProtocol.proto | 2 +
.../apache/hadoop/hdds/scm/PlacementPolicy.java | 27 ++--
.../hadoop/hdds/scm/SCMCommonPlacementPolicy.java | 92 +++++++++----
.../algorithms/SCMContainerPlacementCapacity.java | 20 +--
.../algorithms/SCMContainerPlacementRackAware.java | 127 +++++++++--------
.../SCMContainerPlacementRackScatter.java | 17 +--
.../algorithms/SCMContainerPlacementRandom.java | 22 +--
.../replication/ECUnderReplicationHandler.java | 21 +--
.../replication/MisReplicationHandler.java | 4 +-
.../QuasiClosedStuckUnderReplicationHandler.java | 5 +-
.../replication/RatisUnderReplicationHandler.java | 8 +-
.../replication/ReplicationManagerUtil.java | 25 ++--
.../hdds/scm/pipeline/ECPipelineProvider.java | 13 +-
.../hadoop/hdds/scm/pipeline/PipelineFactory.java | 4 +-
.../hdds/scm/pipeline/PipelinePlacementPolicy.java | 25 ++--
.../hadoop/hdds/scm/pipeline/PipelineProvider.java | 16 ++-
.../hdds/scm/pipeline/RatisPipelineProvider.java | 23 ++--
.../hdds/scm/pipeline/SimplePipelineProvider.java | 23 +++-
.../hdds/scm/TestSCMCommonPlacementPolicy.java | 14 +-
.../hadoop/hdds/scm/container/MockNodeManager.java | 19 ++-
.../algorithms/TestContainerPlacementFactory.java | 5 +-
.../TestSCMContainerPlacementCapacity.java | 3 +-
.../TestSCMContainerPlacementRackAware.java | 134 ++++++++++--------
.../TestSCMContainerPlacementRackScatter.java | 109 +++++++++------
.../TestSCMContainerPlacementRandom.java | 9 +-
.../replication/MisReplicationHandlerTests.java | 3 +-
.../container/replication/ReplicationTestUtil.java | 9 +-
.../replication/TestECMisReplicationHandler.java | 7 +-
.../replication/TestECUnderReplicationHandler.java | 12 +-
.../TestRatisMisReplicationHandler.java | 5 +-
.../TestRatisUnderReplicationHandler.java | 6 +-
.../replication/TestReplicationManager.java | 2 +-
.../scm/pipeline/MockRatisPipelineProvider.java | 7 +-
.../hdds/scm/pipeline/TestECPipelineProvider.java | 11 +-
.../TestPipelineDatanodesIntersection.java | 3 +-
.../scm/pipeline/TestPipelinePlacementFactory.java | 19 +--
.../scm/pipeline/TestPipelinePlacementPolicy.java | 36 ++---
.../scm/pipeline/TestRatisPipelineProvider.java | 151 ++++++++++++++-------
.../scm/pipeline/TestSimplePipelineProvider.java | 64 +++++++--
.../placement/TestContainerPlacement.java | 5 +-
.../hadoop/hdds/protocol/StorageTierTest.java | 12 +-
.../ozone/recon/scm/ReconPipelineFactory.java | 8 +-
48 files changed, 806 insertions(+), 511 deletions(-)
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTier.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTier.java
index c9215831497..1e5c40dfe64 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTier.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTier.java
@@ -17,15 +17,9 @@
package org.apache.hadoop.hdds.client;
-import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.ONE;
-import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.THREE;
-
import java.util.Arrays;
import java.util.Collections;
-import java.util.EnumMap;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.StorageTierProto;
@@ -40,21 +34,20 @@ public enum StorageTier {
private final String tierName;
private final List<StorageType> storageTypes;
- private final boolean uniformStorageType;
- private static final Map<StorageTier, Map<ReplicationConfig, List<StorageType>>>
- CACHE = new EnumMap<>(StorageTier.class);
+ private final boolean isUniform;
+ private static StorageTier defaultTier = DISK;
StorageTier(String tierName) {
this.tierName = tierName;
this.storageTypes = Collections.emptyList();
- this.uniformStorageType = true;
+ this.isUniform = true;
}
// Constructor for uniform storage tiers
StorageTier(String tierName, StorageType uniformStorageType) {
this.tierName = tierName;
this.storageTypes = Collections.singletonList(uniformStorageType);
- this.uniformStorageType = true;
+ this.isUniform = true;
}
// Constructor for non-uniform storage tiers
@@ -68,25 +61,11 @@ public enum StorageTier {
" StorageType were provided.");
}
this.storageTypes = Arrays.asList(storageTypes);
- this.uniformStorageType = false;
+ this.isUniform = false;
}
- static {
- // Precompute storage type mappings for each replication config
- for (StorageTier tier : StorageTier.values()) {
- Map<ReplicationConfig, List<StorageType>> tierCache = new HashMap<>();
- List<ReplicationConfig> replicationConfigs = Arrays.asList(
- RatisReplicationConfig.getInstance(ONE),
- RatisReplicationConfig.getInstance(THREE),
- StandaloneReplicationConfig.getInstance(ONE),
- StandaloneReplicationConfig.getInstance(THREE)
- );
-
- for (ReplicationConfig config : replicationConfigs) {
- tierCache.put(config, tier.computeStorageTypes(config));
- }
- CACHE.put(tier, tierCache);
- }
+ public static StorageTier getDefaultTier() {
+ return defaultTier;
}
public StorageTierProto toProto() {
@@ -121,50 +100,31 @@ public String getTierName() {
return tierName;
}
- public boolean isUniformStorageType() {
- return uniformStorageType;
- }
-
- /**
- * Computes the list of StorageTypes based on replication configuration.
- *
- * @param replicationConfig The replication configuration.
- * @return The list of StorageTypes for the given tier and replication configuration.
- */
- private List<StorageType> computeStorageTypes(
- ReplicationConfig replicationConfig) {
- if (isUniformStorageType()) {
- int numberOfNodes = replicationConfig.getRequiredNodes();
- if (storageTypes.isEmpty()) {
- return Collections.emptyList();
- }
- return Collections.nCopies(numberOfNodes, storageTypes.get(0));
- } else {
- throw new UnsupportedOperationException(
- "Unsupported not UniformStorage Storage Tier: " + replicationConfig);
- }
+ public boolean isUniform() {
+ return isUniform;
}
/**
* Maps a StorageTier to its corresponding StorageType based on replication type.
*
- * @param replicationConfig The replication configuration.
+ * @param nodeCount The node count of the storageTier required.
* @return The list of StorageTypes corresponding to the given tier and replication configuration.
* @throws IllegalArgumentException if the replication configuration is not supported.
*/
- public List<StorageType> getStorageTypes(
- ReplicationConfig replicationConfig) {
- Map<ReplicationConfig, List<StorageType>> tierCache = CACHE.get(this);
+ public List<StorageType> getStorageTypes(int nodeCount) {
+ if (nodeCount < 0) {
+ throw new IllegalArgumentException("Unsupported node count: " +
+ nodeCount + " for StorageTier: " + getTierName());
+ }
- if (tierCache != null) {
- List<StorageType> cachedStorageType = tierCache.get(replicationConfig);
- if (cachedStorageType != null) {
- return cachedStorageType;
+ if (isUniform()) {
+ if (storageTypes.isEmpty()) {
+ return Collections.emptyList();
}
+ return Collections.nCopies(nodeCount, storageTypes.get(0));
}
- throw new IllegalArgumentException("Unsupported ReplicationConfig: " +
- replicationConfig + " for StorageTier: " + getTierName());
+ throw new UnsupportedOperationException(
+ "Unsupported not uniform StorageTier: " + this);
}
-
}
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTierUtil.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTierUtil.java
new file mode 100644
index 00000000000..d68c71a31da
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTierUtil.java
@@ -0,0 +1,66 @@
+/*
+ * 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.client;
+
+import java.util.List;
+import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdds.scm.exceptions.SCMException;
+
+/**
+ * Utility class for managing StorageTier operations.
+ */
+public final class StorageTierUtil {
+ private StorageTierUtil() {
+ }
+
+ /**
+ * Validates the given StorageTier and throws an exception if it is empty.
+ *
+ * @param storageTier the StorageTier to check
+ * @throws IllegalArgumentException if the StorageTier is null
+ * @throws SCMException if the StorageTier is empty
+ */
+ public static void validateNotEmpty(StorageTier storageTier) throws SCMException {
+ if (storageTier == null) {
+ throw new IllegalArgumentException("storageTier must not be null");
+ }
+ if (storageTier.equals(StorageTier.EMPTY)) {
+ throw new SCMException("Cannot create Pipeline for empty tier",
+ SCMException.ResultCodes.CANNOT_CREATE_PIPELINE_FOR_EMPTY_TIER);
+ }
+ }
+
+ /**
+ * Returns the StorageType for uniform StorageTier.
+ *
+ * @param storageTier the StorageTier to get StorageType from
+ * @return The uniform StorageTier corresponding StorageType
+ * @throws SCMException if the StorageTier is non-uniform or the EMPTY StorageTier
+ */
+ public static StorageType getStorageTypeForUniformStorageTier(StorageTier storageTier, ReplicationConfig config)
+ throws SCMException {
+ validateNotEmpty(storageTier);
+ List<StorageType> storageTypes = storageTier.getStorageTypes(config.getRequiredNodes());
+ if (storageTier.isUniform()) {
+ return storageTypes.get(0);
+ } else {
+ throw new SCMException("Unsupported non-uniform storage tier " + storageTier,
+ SCMException.ResultCodes.UNSUPPORTED_NON_UNIFORM_STORAGE_TIER);
+ }
+ }
+}
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTypeUtils.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTypeUtils.java
index 0956581ff87..6b94d16e57b 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTypeUtils.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTypeUtils.java
@@ -43,6 +43,8 @@ public static HddsProtos.StorageTypeProto getStorageTypeProto(@Nonnull StorageTy
return HddsProtos.StorageTypeProto.PROVIDED;
case RAM_DISK:
return HddsProtos.StorageTypeProto.RAM_DISK;
+ case NVDIMM:
+ return HddsProtos.StorageTypeProto.NVDIMM;
default:
throw new IllegalArgumentException("Illegal Storage Type specified");
}
@@ -61,6 +63,8 @@ public static StorageType getFromProtobuf(@Nonnull HddsProtos.StorageTypeProto p
return StorageType.PROVIDED;
case RAM_DISK:
return StorageType.RAM_DISK;
+ case NVDIMM:
+ return StorageType.NVDIMM;
default:
throw new IllegalArgumentException("Illegal Storage Type specified");
}
@@ -93,6 +97,7 @@ public static int getIDFromProtobuf(@Nonnull HddsProtos.StorageTypeProto proto)
case ARCHIVE:
case PROVIDED:
case RAM_DISK:
+ case NVDIMM:
return proto.getNumber();
default:
throw new IllegalArgumentException("Illegal Storage Type specified");
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/exceptions/SCMException.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/exceptions/SCMException.java
index 66df2b91501..24964d0c21d 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/exceptions/SCMException.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/exceptions/SCMException.java
@@ -157,6 +157,8 @@ public enum ResultCodes {
CA_ROTATION_IN_POST_PROGRESS,
CONTAINER_ALREADY_CLOSED,
CONTAINER_ALREADY_CLOSING,
- UNSUPPORTED_OPERATION
+ UNSUPPORTED_OPERATION,
+ CANNOT_CREATE_PIPELINE_FOR_EMPTY_TIER,
+ UNSUPPORTED_NON_UNIFORM_STORAGE_TIER,
}
}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/StorageLocationReport.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/StorageLocationReport.java
index a32cdce53a1..75c2f973222 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/StorageLocationReport.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/StorageLocationReport.java
@@ -20,6 +20,7 @@
import java.io.IOException;
import net.jcip.annotations.Immutable;
import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdds.client.StorageTypeUtils;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.StorageTypeProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.MetadataStorageReportProto;
@@ -116,32 +117,7 @@ public StorageType getStorageType() {
}
private StorageTypeProto getStorageTypeProto() throws IllegalArgumentException {
- return getStorageTypeProto(getStorageType());
- }
-
- public static StorageTypeProto getStorageTypeProto(StorageType type)
- throws IllegalArgumentException {
- StorageTypeProto storageTypeProto;
- switch (type) {
- case SSD:
- storageTypeProto = StorageTypeProto.SSD;
- break;
- case DISK:
- storageTypeProto = StorageTypeProto.DISK;
- break;
- case ARCHIVE:
- storageTypeProto = StorageTypeProto.ARCHIVE;
- break;
- case PROVIDED:
- storageTypeProto = StorageTypeProto.PROVIDED;
- break;
- case RAM_DISK:
- storageTypeProto = StorageTypeProto.RAM_DISK;
- break;
- default:
- throw new IllegalArgumentException("Illegal Storage Type specified");
- }
- return storageTypeProto;
+ return StorageTypeUtils.getStorageTypeProto(getStorageType());
}
public long getReserved() {
@@ -176,7 +152,7 @@ public static StorageType getStorageType(StorageTypeProto proto) throws
storageType = StorageType.RAM_DISK;
break;
default:
- throw new IllegalArgumentException("Illegal Storage Type specified");
+ throw new IllegalArgumentException("Illegal Storage Type specified: " + proto);
}
return storageType;
}
@@ -247,7 +223,7 @@ public static StorageLocationReport getFromProtobuf(StorageReportProto report)
builder.setScmUsed(report.getScmUsed());
}
if (report.hasStorageType()) {
- builder.setStorageType(getStorageType(report.getStorageType()));
+ builder.setStorageType(StorageTypeUtils.getFromProtobuf(report.getStorageType()));
}
if (report.hasRemaining()) {
builder.setRemaining(report.getRemaining());
diff --git a/hadoop-hdds/interface-client/src/main/proto/hdds.proto b/hadoop-hdds/interface-client/src/main/proto/hdds.proto
index cd55fb82d0e..67eab258084 100644
--- a/hadoop-hdds/interface-client/src/main/proto/hdds.proto
+++ b/hadoop-hdds/interface-client/src/main/proto/hdds.proto
@@ -159,6 +159,7 @@ enum StorageTypeProto {
ARCHIVE = 3;
RAM_DISK = 4;
PROVIDED = 5;
+ NVDIMM = 6;
}
/**
diff --git a/hadoop-hdds/interface-server/src/main/proto/ScmServerProtocol.proto b/hadoop-hdds/interface-server/src/main/proto/ScmServerProtocol.proto
index 4de70addfcf..1acec652062 100644
--- a/hadoop-hdds/interface-server/src/main/proto/ScmServerProtocol.proto
+++ b/hadoop-hdds/interface-server/src/main/proto/ScmServerProtocol.proto
@@ -143,6 +143,8 @@ enum Status {
CONTAINER_ALREADY_CLOSED = 45;
CONTAINER_ALREADY_CLOSING = 46;
UNSUPPORTED_OPERATION = 47;
+ CANNOT_CREATE_PIPELINE_FOR_EMPTY_TIER = 48;
+ UNSUPPORTED_NON_UNIFORM_STORAGE_TIER = 49;
}
/**
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/PlacementPolicy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/PlacementPolicy.java
index 917a29b2a8e..75926f04406 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/PlacementPolicy.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/PlacementPolicy.java
@@ -22,34 +22,37 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.scm.container.ContainerReplica;
+import org.apache.hadoop.hdds.scm.exceptions.SCMException;
/**
* A PlacementPolicy support choosing datanodes to build
* pipelines or containers with specified constraints.
*/
public interface PlacementPolicy {
-
default List<DatanodeDetails> chooseDatanodes(
- List<DatanodeDetails> excludedNodes,
- List<DatanodeDetails> favoredNodes, int nodesRequired,
- long metadataSizeRequired, long dataSizeRequired) throws IOException {
+ List<DatanodeDetails> excludedNodes,
+ List<DatanodeDetails> favoredNodes, int nodesRequired,
+ long metadataSizeRequired, long dataSizeRequired,
+ StorageType storageType) throws SCMException {
return this.chooseDatanodes(Collections.emptyList(), excludedNodes,
- favoredNodes, nodesRequired, metadataSizeRequired,
- dataSizeRequired);
+ favoredNodes, nodesRequired, metadataSizeRequired,
+ dataSizeRequired, storageType);
}
/**
* Given an initial set of datanodes and the size required,
* return set of datanodes that satisfy the nodes and size requirement.
*
- * @param usedNodes - List of nodes already chosen for pipeline
- * @param excludedNodes - list of nodes to be excluded.
- * @param favoredNodes - list of nodes preferred.
- * @param nodesRequired - number of datanodes required.
- * @param dataSizeRequired - size required for the container.
+ * @param usedNodes - List of nodes already chosen for pipeline
+ * @param excludedNodes - list of nodes to be excluded.
+ * @param favoredNodes - list of nodes preferred.
+ * @param nodesRequired - number of datanodes required.
* @param metadataSizeRequired - size required for Ratis metadata.
+ * @param dataSizeRequired - size required for the container.
+ * @param storageType - StorageType required for the container.
* @return list of datanodes chosen.
* @throws IOException
*/
@@ -57,7 +60,7 @@ List<DatanodeDetails> chooseDatanodes(List<DatanodeDetails> usedNodes,
List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> favoredNodes,
int nodesRequired, long metadataSizeRequired,
- long dataSizeRequired) throws IOException;
+ long dataSizeRequired, StorageType storageType) throws SCMException;
/**
* Given a list of datanode and the number of replicas required, return
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 0deece79c6d..cbec57bc33f 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
@@ -34,9 +34,12 @@
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
+import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdds.client.StorageTypeUtils;
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;
@@ -46,6 +49,7 @@
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;
@@ -85,7 +89,7 @@ public abstract class SCMCommonPlacementPolicy implements
* RackAwarePlacementPolicy that needs to know if the old or new interface is
* used to be able to use the node lists correctly.
*/
- private static final List<DatanodeDetails> UNSET_USED_NODES
+ public static final List<DatanodeDetails> UNSET_USED_NODES
= Collections.unmodifiableList(new ArrayList<>());
/**
@@ -133,10 +137,10 @@ public final List<DatanodeDetails> chooseDatanodes(
List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> favoredNodes, int nodesRequired,
long metadataSizeRequired,
- long dataSizeRequired) throws SCMException {
+ long dataSizeRequired, StorageType storageType) throws SCMException {
return this.chooseDatanodes(UNSET_USED_NODES, excludedNodes,
favoredNodes, nodesRequired, metadataSizeRequired,
- dataSizeRequired);
+ dataSizeRequired, storageType);
}
/**
@@ -173,12 +177,14 @@ private List<DatanodeDetails> validateDatanodes(List<DatanodeDetails> dns) {
* 2. We place containers on nodes with enough space for that container.
* 3. if a set of containers are requested, we either meet the required
* number of nodes or we fail that request.
- * @param usedNodes - datanodes with existing replicas
- * @param excludedNodes - datanodes with failures
- * @param favoredNodes - list of nodes preferred.
- * @param nodesRequired - number of datanodes required.
- * @param dataSizeRequired - size required for the container.
+ *
+ * @param usedNodes - datanodes with existing replicas
+ * @param excludedNodes - datanodes with failures
+ * @param favoredNodes - list of nodes preferred.
+ * @param nodesRequired - number of datanodes required.
* @param metadataSizeRequired - size required for Ratis metadata.
+ * @param dataSizeRequired - size required for the container.
+ * @param storageType - StorageType required for the container.
* @return list of datanodes chosen.
* @throws SCMException SCM exception.
*/
@@ -187,7 +193,8 @@ public final List<DatanodeDetails> chooseDatanodes(
List<DatanodeDetails> usedNodes,
List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> favoredNodes,
- int nodesRequired, long metadataSizeRequired, long dataSizeRequired)
+ int nodesRequired, long metadataSizeRequired, long dataSizeRequired,
+ StorageType storageType)
throws SCMException {
/*
This method calls the chooseDatanodeInternal after fixing
@@ -204,25 +211,28 @@ object of DatanodeDetails(with Topology Information) while trying to get the
*/
return chooseDatanodesInternal(validateDatanodes(usedNodes),
validateDatanodes(excludedNodes), favoredNodes, nodesRequired,
- metadataSizeRequired, dataSizeRequired);
+ metadataSizeRequired, dataSizeRequired, storageType);
}
/**
* Pipeline placement choose datanodes to join the pipeline.
- * @param usedNodes - list of the datanodes to already chosen in the
- * pipeline.
- * @param excludedNodes - excluded nodes
- * @param favoredNodes - list of nodes preferred.
- * @param nodesRequired - number of datanodes required.
- * @param dataSizeRequired - size required for the container.
+ *
+ * @param usedNodes - list of the datanodes to already chosen in the
+ * pipeline.
+ * @param excludedNodes - excluded nodes
+ * @param favoredNodes - list of nodes preferred.
+ * @param nodesRequired - number of datanodes required.
* @param metadataSizeRequired - size required for Ratis metadata.
+ * @param dataSizeRequired - size required for the container.
+ * @param storageType - StorageType required for the container.
* @return a list of chosen datanodeDetails
* @throws SCMException when chosen nodes are not enough in numbers
*/
protected List<DatanodeDetails> chooseDatanodesInternal(
List<DatanodeDetails> usedNodes, List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> favoredNodes,
- int nodesRequired, long metadataSizeRequired, long dataSizeRequired)
+ int nodesRequired, long metadataSizeRequired, long dataSizeRequired,
+ StorageType storageType)
throws SCMException {
List<DatanodeDetails> healthyNodes =
nodeManager.getNodes(NodeStatus.inServiceHealthy());
@@ -249,8 +259,8 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
}
- return filterNodesWithSpace(healthyNodes, nodesRequired,
- metadataSizeRequired);
+ return filterNodesWithSpaceAndStorageType(healthyNodes, nodesRequired,
+ metadataSizeRequired, dataSizeRequired, storageType);
}
/**
@@ -267,18 +277,20 @@ protected boolean usedNodesPassed(List<DatanodeDetails> list) {
return list != UNSET_USED_NODES;
}
- public List<DatanodeDetails> filterNodesWithSpace(List<DatanodeDetails> nodes,
- int nodesRequired, long metadataSizeRequired)
+ public List<DatanodeDetails> filterNodesWithSpaceAndStorageType(List<DatanodeDetails> nodes,
+ int nodesRequired, long metadataSizeRequired, long dataSizeRequired,
+ StorageType storageType)
throws SCMException {
List<DatanodeDetails> nodesWithSpace = nodes.stream().filter(d ->
- hasEnoughSpace(d, metadataSizeRequired, nodeManager))
+ hasEnoughSpace(d, metadataSizeRequired, dataSizeRequired, storageType, nodeManager))
.collect(Collectors.toList());
if (nodesWithSpace.size() < nodesRequired) {
String msg = String.format("Unable to find enough nodes that meet the " +
- "space requirement of %d bytes for metadata and an available " +
- "container slot in healthy node set. Required %d. Found %d.",
- metadataSizeRequired, nodesRequired, nodesWithSpace.size());
+ "space requirement of %d bytes for metadata and %d bytes in storageType %s for " +
+ "data in healthy node set. Required %d. Found %d.",
+ metadataSizeRequired, dataSizeRequired, storageType, nodesRequired,
+ nodesWithSpace.size());
LOG.warn(msg);
throw new SCMException(msg,
SCMException.ResultCodes.FAILED_TO_FIND_NODES_WITH_SPACE);
@@ -303,6 +315,7 @@ public List<DatanodeDetails> filterNodesWithSpace(List<DatanodeDetails> nodes,
*/
public static boolean hasEnoughSpace(DatanodeDetails datanodeDetails,
long metadataSizeRequired,
+ long dataSizeRequired, StorageType storageType,
NodeManager nodeManager) {
Preconditions.checkArgument(datanodeDetails instanceof DatanodeInfo);
@@ -318,6 +331,29 @@ public static boolean hasEnoughSpace(DatanodeDetails datanodeDetails,
return false;
}
+ // Tier-aware data check (StorageType feature): when a specific storageType is
+ // requested, ensure the node has a volume of that tier with enough usable space.
+ // When storageType is null (any tier), rely on the slot check above (master behavior).
+ boolean enoughForData = false;
+ if (dataSizeRequired > 0 && storageType != null) {
+ for (StorageReportProto reportProto : datanodeInfo.getStorageReports()) {
+ boolean matchesTier = StorageTypeUtils.getFromProtobuf(
+ reportProto.getStorageType()).equals(storageType);
+ if (matchesTier && VolumeUsage.getUsableSpace(reportProto) > dataSizeRequired) {
+ enoughForData = true;
+ break;
+ }
+ }
+ } 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) {
for (MetadataStorageReportProto reportProto
: datanodeInfo.getMetadataStorageReports()) {
@@ -525,7 +561,7 @@ public void removePeers(DatanodeDetails dn,
* @return true if the datanode is available.
*/
public boolean isValidNode(DatanodeDetails datanodeDetails,
- long metadataSizeRequired, long dataSizeRequired) {
+ long metadataSizeRequired, long dataSizeRequired, StorageType storageType) {
DatanodeInfo datanodeInfo = (DatanodeInfo)getNodeManager()
.getNode(datanodeDetails.getID());
if (datanodeInfo == null) {
@@ -534,8 +570,8 @@ public boolean isValidNode(DatanodeDetails datanodeDetails,
return false;
}
NodeStatus nodeStatus = datanodeInfo.getNodeStatus();
- if (nodeStatus.isNodeWritable() && (hasEnoughSpace(datanodeInfo, metadataSizeRequired,
- nodeManager))) {
+ if (nodeStatus.isNodeWritable() && (hasEnoughSpace(datanodeInfo,
+ metadataSizeRequired, dataSizeRequired, storageType, 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/container/placement/algorithms/SCMContainerPlacementCapacity.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementCapacity.java
index e135f3de7fb..7de1e8f48a5 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementCapacity.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementCapacity.java
@@ -19,6 +19,7 @@
import com.google.common.annotations.VisibleForTesting;
import java.util.List;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.scm.SCMCommonPlacementPolicy;
@@ -89,26 +90,27 @@ public SCMContainerPlacementCapacity(final NodeManager nodeManager,
/**
* Called by SCM to choose datanodes.
*
- * @param usedNodes - list of the datanodes to already chosen in the
- * pipeline.
- * @param excludedNodes - list of the datanodes to exclude.
- * @param favoredNodes - list of nodes preferred.
- * @param nodesRequired - number of datanodes required.
- * @param dataSizeRequired - size required for the container.
+ * @param usedNodes - list of the datanodes to already chosen in the
+ * pipeline.
+ * @param excludedNodes - list of the datanodes to exclude.
+ * @param favoredNodes - list of nodes preferred.
+ * @param nodesRequired - number of datanodes required.
* @param metadataSizeRequired - size required for Ratis metadata.
+ * @param dataSizeRequired - size required for the container.
+ * @param storageType - StorageType required for the container.
* @return List of datanodes.
- * @throws SCMException SCMException
+ * @throws SCMException SCMException
*/
@Override
protected List<DatanodeDetails> chooseDatanodesInternal(
List<DatanodeDetails> usedNodes, List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> favoredNodes,
final int nodesRequired, long metadataSizeRequired,
- long dataSizeRequired) throws SCMException {
+ long dataSizeRequired, StorageType storageType) throws SCMException {
metrics.incrDatanodeRequestCount(nodesRequired);
List<DatanodeDetails> healthyNodes = super.chooseDatanodesInternal(
usedNodes, excludedNodes, favoredNodes, nodesRequired,
- metadataSizeRequired, dataSizeRequired);
+ metadataSizeRequired, dataSizeRequired, storageType);
if (healthyNodes.size() == nodesRequired) {
return healthyNodes;
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackAware.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackAware.java
index 7bcc5fb8926..fc3ea2cafd0 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackAware.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackAware.java
@@ -25,6 +25,7 @@
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.scm.SCMCommonPlacementPolicy;
@@ -87,25 +88,26 @@ public SCMContainerPlacementRackAware(final NodeManager nodeManager,
* There are two scenarios, one is choosing all nodes for a new pipeline.
* Another is choosing node to meet replication requirement.
*
- * @param usedNodes - list of the datanodes to already chosen in the
- * pipeline.
- * @param excludedNodes - list of the datanodes to exclude.
- * @param favoredNodes - list of nodes preferred. This is a hint to the
- * allocator, whether the favored nodes will be used
- * depends on whether the nodes meets the allocator's
- * requirement.
- * @param nodesRequired - number of datanodes required.
- * @param dataSizeRequired - size required for the container.
+ * @param usedNodes - list of the datanodes to already chosen in the
+ * pipeline.
+ * @param excludedNodes - list of the datanodes to exclude.
+ * @param favoredNodes - list of nodes preferred. This is a hint to the
+ * allocator, whether the favored nodes will be used
+ * depends on whether the nodes meets the allocator's
+ * requirement.
+ * @param nodesRequired - number of datanodes required.
* @param metadataSizeRequired - size required for Ratis metadata.
+ * @param dataSizeRequired - size required for the container.
+ * @param storageType - StorageType required for the container.
* @return List of datanodes.
- * @throws SCMException SCMException
+ * @throws SCMException SCMException
*/
@Override
protected List<DatanodeDetails> chooseDatanodesInternal(
List<DatanodeDetails> usedNodes,
List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> favoredNodes, int nodesRequired,
- long metadataSizeRequired, long dataSizeRequired)
+ long metadataSizeRequired, long dataSizeRequired, StorageType storageType)
throws SCMException {
Map<String, Long> mapSizeRequired = new HashMap<>();
mapSizeRequired.put(META_DATA_SIZE_REQUIRED, metadataSizeRequired);
@@ -115,7 +117,7 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
// If interface is called without used nodes
// In this case consider only exclude nodes to determine racks
return chooseDatanodesInternalLegacy(excludedNodes,
- favoredNodes, nodesRequired, mapSizeRequired);
+ favoredNodes, nodesRequired, mapSizeRequired, storageType);
}
Preconditions.checkArgument(nodesRequired > 0);
metrics.incrDatanodeRequestCount(nodesRequired);
@@ -126,7 +128,7 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
throw new SCMException("No enough datanodes to choose. " +
"TotalNode = " + datanodeCount + " RequiredNode = " + nodesRequired +
" ExcludedNode = " + excludedNodesCount +
- " UsedNode = " + usedNodesCount, null);
+ " UsedNode = " + usedNodesCount + " StorageType = " + storageType, null);
}
List<DatanodeDetails> mutableFavoredNodes = favoredNodes;
// sanity check of favoredNodes
@@ -158,7 +160,7 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
} else {
firstNode = chooseNode(mutableExcludedNodes, null,
null, metadataSizeRequired,
- dataSizeRequired);
+ dataSizeRequired, storageType);
}
chosenNodes.add(firstNode);
nodesRequired--;
@@ -176,7 +178,7 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
} else {
mutableExcludedNodes.add(firstNode);
secondNode = chooseNode(mutableExcludedNodes, Arrays.asList(firstNode),
- Arrays.asList(firstNode), metadataSizeRequired, dataSizeRequired);
+ Arrays.asList(firstNode), metadataSizeRequired, dataSizeRequired, storageType);
}
chosenNodes.add(secondNode);
nodesRequired--;
@@ -201,7 +203,7 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
favorIndex++;
} else {
firstNode = chooseNode(mutableExcludedNodes, mutableUsedNodes,
- mutableUsedNodes, metadataSizeRequired, dataSizeRequired);
+ mutableUsedNodes, metadataSizeRequired, dataSizeRequired, storageType);
}
chosenNodes.add(firstNode);
nodesRequired--;
@@ -225,7 +227,7 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
mutableExcludedNodes.addAll(mutableUsedNodes);
return chooseNodes(mutableExcludedNodes, chosenNodes,
mutableFavoredNodes, mutableUsedNodes, favorIndex,
- nodesRequired, mapSizeRequired);
+ nodesRequired, mapSizeRequired, storageType);
}
}
}
@@ -240,7 +242,7 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
} else {
secondNode =
chooseNode(mutableExcludedNodes, mutableUsedNodes,
- mutableUsedNodes, metadataSizeRequired, dataSizeRequired);
+ mutableUsedNodes, metadataSizeRequired, dataSizeRequired, storageType);
}
chosenNodes.add(secondNode);
mutableExcludedNodes.add(secondNode);
@@ -254,7 +256,7 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
}
// choose remaining nodes on different racks
return chooseNodes(mutableExcludedNodes, chosenNodes, mutableFavoredNodes,
- mutableUsedNodes, favorIndex, nodesRequired, mapSizeRequired);
+ mutableUsedNodes, favorIndex, nodesRequired, mapSizeRequired, storageType);
}
/**
@@ -262,20 +264,21 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
* There are two scenarios, one is choosing all nodes for a new pipeline.
* Another is choosing node to meet replication requirement.
*
- * @param excludedNodes - list of the datanodes to exclude.
- * @param favoredNodes - list of nodes preferred. This is a hint to the
- * allocator, whether the favored nodes will be used
- * depends on whether the nodes meets the allocator's
- * requirement.
- * @param nodesRequired - number of datanodes required.
+ * @param excludedNodes - list of the datanodes to exclude.
+ * @param favoredNodes - list of nodes preferred. This is a hint to the
+ * allocator, whether the favored nodes will be used
+ * depends on whether the nodes meets the allocator's
+ * requirement.
+ * @param nodesRequired - number of datanodes required.
* @param mapSizeRequired - size required for the container, Ratis metadata.
+ * @param storageType - StorageType required for the container.
* @return List of datanodes.
- * @throws SCMException SCMException
+ * @throws SCMException SCMException
*/
protected List<DatanodeDetails> chooseDatanodesInternalLegacy(
List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> favoredNodes, int nodesRequired,
- Map<String, Long> mapSizeRequired)
+ Map<String, Long> mapSizeRequired, StorageType storageType)
throws SCMException {
Preconditions.checkArgument(nodesRequired > 0);
metrics.incrDatanodeRequestCount(nodesRequired);
@@ -285,7 +288,8 @@ protected List<DatanodeDetails> chooseDatanodesInternalLegacy(
throw new SCMException("No enough datanodes to choose. " +
"TotalNode = " + datanodeCount +
" RequiredNode = " + nodesRequired +
- " ExcludedNode = " + excludedNodesCount, null);
+ " ExcludedNode = " + excludedNodesCount +
+ " StorageType = " + storageType, null);
}
long metadataSizeRequired = mapSizeRequired.get(META_DATA_SIZE_REQUIRED);
long dataSizeRequired = mapSizeRequired.get(DATA_SIZE_REQUIRED);
@@ -313,7 +317,7 @@ protected List<DatanodeDetails> chooseDatanodesInternalLegacy(
favorIndex++;
} else {
firstNode = chooseNode(null, null, null, metadataSizeRequired,
- dataSizeRequired);
+ dataSizeRequired, storageType);
}
chosenNodes.add(firstNode);
nodesRequired--;
@@ -331,7 +335,7 @@ protected List<DatanodeDetails> chooseDatanodesInternalLegacy(
favorIndex++;
} else {
secondNode = chooseNode(chosenNodes, Arrays.asList(firstNode),
- Arrays.asList(firstNode), metadataSizeRequired, dataSizeRequired);
+ Arrays.asList(firstNode), metadataSizeRequired, dataSizeRequired, storageType);
}
chosenNodes.add(secondNode);
nodesRequired--;
@@ -342,7 +346,7 @@ protected List<DatanodeDetails> chooseDatanodesInternalLegacy(
mutableUsedNodes.addAll(chosenNodes);
// choose remaining datanodes on different rack as first and second
return chooseNodes(null, chosenNodes, mutableFavoredNodes,
- mutableUsedNodes, favorIndex, nodesRequired, mapSizeRequired);
+ mutableUsedNodes, favorIndex, nodesRequired, mapSizeRequired, storageType);
} else {
List<DatanodeDetails> mutableExcludedNodes = new ArrayList<>(excludedNodes);
// choose node to meet replication requirement
@@ -359,7 +363,7 @@ protected List<DatanodeDetails> chooseDatanodesInternalLegacy(
favorIndex++;
} else {
firstNode = chooseNode(mutableExcludedNodes, excludedNodes,
- excludedNodes, metadataSizeRequired, dataSizeRequired);
+ excludedNodes, metadataSizeRequired, dataSizeRequired, storageType);
}
chosenNodes.add(firstNode);
nodesRequired--;
@@ -370,7 +374,7 @@ protected List<DatanodeDetails> chooseDatanodesInternalLegacy(
mutableUsedNodes.addAll(chosenNodes);
mutableUsedNodes.addAll(mutableExcludedNodes);
return chooseNodes(null, chosenNodes, mutableFavoredNodes,
- mutableUsedNodes, favorIndex, nodesRequired, mapSizeRequired);
+ mutableUsedNodes, favorIndex, nodesRequired, mapSizeRequired, storageType);
}
// case 2: two or more excluded nodes, if these two nodes are
// in the same rack, then choose nodes on different racks, otherwise,
@@ -385,7 +389,7 @@ protected List<DatanodeDetails> chooseDatanodesInternalLegacy(
mutableUsedNodes.addAll(mutableExcludedNodes);
return chooseNodes(mutableExcludedNodes, chosenNodes,
mutableFavoredNodes, mutableUsedNodes, favorIndex,
- nodesRequired, mapSizeRequired);
+ nodesRequired, mapSizeRequired, storageType);
}
}
}
@@ -400,7 +404,7 @@ protected List<DatanodeDetails> chooseDatanodesInternalLegacy(
} else {
secondNode =
chooseNode(chosenNodes, mutableExcludedNodes, mutableExcludedNodes,
- metadataSizeRequired, dataSizeRequired);
+ metadataSizeRequired, dataSizeRequired, storageType);
}
chosenNodes.add(secondNode);
mutableExcludedNodes.add(secondNode);
@@ -413,7 +417,7 @@ protected List<DatanodeDetails> chooseDatanodesInternalLegacy(
mutableUsedNodes.addAll(mutableExcludedNodes);
return chooseNodes(mutableExcludedNodes, chosenNodes, mutableFavoredNodes,
mutableUsedNodes,
- favorIndex, nodesRequired, mapSizeRequired);
+ favorIndex, nodesRequired, mapSizeRequired, storageType);
}
}
@@ -427,21 +431,21 @@ public DatanodeDetails chooseNode(List<DatanodeDetails> healthyNodes) {
* meets all the requirements, there is fallback chosen process depending on
* whether fallback is allowed when this class is instantiated.
*
- *
- * @param excludedNodes - list of the datanodes to excluded. Can be null.
- * @param affinityNodes - the chosen nodes should be on the same rack as
- * affinityNodes. Can be null.
- * @param usedNodes - the chosen nodes should be on the different rack
- * than usedNodes rack when affinityNode is null.
- * @param dataSizeRequired - size required for the container.
+ * @param excludedNodes - list of the datanodes to excluded. Can be null.
+ * @param affinityNodes - the chosen nodes should be on the same rack as
+ * affinityNodes. Can be null.
+ * @param usedNodes - the chosen nodes should be on the different rack
+ * than usedNodes rack when affinityNode is null.
* @param metadataSizeRequired - size required for Ratis metadata.
+ * @param dataSizeRequired - size required for the container.
+ * @param storageType - StorageType required for the container.
* @return List of chosen datanodes.
- * @throws SCMException SCMException
+ * @throws SCMException SCMException
*/
private DatanodeDetails chooseNode(List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> affinityNodes, List<DatanodeDetails> usedNodes,
long metadataSizeRequired,
- long dataSizeRequired) throws SCMException {
+ long dataSizeRequired, StorageType storageType) throws SCMException {
int ancestorGen = RACK_LEVEL;
int maxRetry = MAX_RETRY;
List<String> excludedNodesForCapacity = null;
@@ -499,7 +503,7 @@ private DatanodeDetails chooseNode(List<DatanodeDetails> excludedNodes,
}
// there is no constrains to reduce or fallback is true
throw new SCMException("No satisfied datanode to meet the" +
- " excludedNodes and affinityNode constrains.", null);
+ " excludedNodes and affinityNode constrains in required StorageType = " + storageType, null);
}
if (usedNodes != null && usedNodes.contains(node)) {
@@ -510,7 +514,7 @@ private DatanodeDetails chooseNode(List<DatanodeDetails> excludedNodes,
continue;
}
- if (isValidNode(node, metadataSizeRequired, dataSizeRequired)) {
+ if (isValidNode(node, metadataSizeRequired, dataSizeRequired, storageType)) {
metrics.incrDatanodeChooseSuccessCount();
if (isFallbacked) {
metrics.incrDatanodeChooseFallbackCount();
@@ -523,7 +527,7 @@ private DatanodeDetails chooseNode(List<DatanodeDetails> excludedNodes,
// avoid the infinite loop
String errMsg = "No satisfied datanode to meet the space constrains. "
+ "metadata size required: " + metadataSizeRequired +
- " data size required: " + dataSizeRequired;
+ " data size required: " + dataSizeRequired + " StorageType required " + storageType;
LOG.info(errMsg);
throw new SCMException(errMsg, null);
}
@@ -538,24 +542,27 @@ private DatanodeDetails chooseNode(List<DatanodeDetails> excludedNodes,
* Choose a batch of datanodes on different rack than excludedNodes or
* chosenNodes.
*
- * @param excludedNodes - list of the datanodes to excluded. Can be null.
- * @param chosenNodes - list of nodes already chosen. These nodes should also
- * be excluded. Cannot be null.
- * @param favoredNodes - list of favoredNodes. It's a hint. Whether the nodes
- * are chosen depends on whether they meet the constrains.
- * Can be null.
- * @param usedNodes - list of the nodes that are already used.
- * @param favorIndex - the node index of favoredNodes which is not chosen yet.
- * @param nodesRequired - number of datanodes required.
+ * @param excludedNodes - list of the datanodes to excluded. Can be null.
+ * @param chosenNodes - list of nodes already chosen. These nodes should also
+ * be excluded. Cannot be null.
+ * @param favoredNodes - list of favoredNodes. It's a hint. Whether the nodes
+ * are chosen depends on whether they meet the constrains.
+ * Can be null.
+ * @param usedNodes - list of the nodes that are already used.
+ * @param favorIndex - the node index of favoredNodes which is not chosen yet.
+ * @param nodesRequired - number of datanodes required.
* @param mapSizeRequired - size required for the container, Ratis metadata.
+ * @param storageType - StorageType required for the container.
* @return List of chosen datanodes.
- * @throws SCMException SCMException
+ * @throws SCMException SCMException
*/
+ @SuppressWarnings("checkstyle:ParameterNumber")
private List<DatanodeDetails> chooseNodes(List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> chosenNodes, List<DatanodeDetails> favoredNodes,
List<DatanodeDetails> usedNodes,
int favorIndex, int nodesRequired,
- Map<String, Long> mapSizeRequired) throws SCMException {
+ Map<String, Long> mapSizeRequired,
+ StorageType storageType) throws SCMException {
Preconditions.checkArgument(chosenNodes != null);
List<DatanodeDetails> excludedNodeList = excludedNodes != null ?
excludedNodes : chosenNodes;
@@ -571,7 +578,7 @@ private List<DatanodeDetails> chooseNodes(List<DatanodeDetails> excludedNodes,
} else {
chosenNode = chooseNode(excludedNodeList, null, usedNodes,
mapSizeRequired.get(META_DATA_SIZE_REQUIRED),
- mapSizeRequired.get(DATA_SIZE_REQUIRED));
+ mapSizeRequired.get(DATA_SIZE_REQUIRED), storageType);
}
excludedNodeList.add(chosenNode);
usedNodes.add(chosenNode);
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackScatter.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackScatter.java
index 13bb9487501..54262e5b5c6 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackScatter.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackScatter.java
@@ -29,6 +29,7 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.scm.ContainerPlacementStatus;
@@ -107,7 +108,7 @@ private Set<DatanodeDetails> chooseNodesFromRacks(List<Node> racks,
List<DatanodeDetails> mutableFavoredNodes,
int nodesRequired, long metadataSizeRequired, long dataSizeRequired,
int maxOuterLoopIterations, Map<Node, Integer> rackCntMap,
- int maxReplicasPerRack) {
+ int maxReplicasPerRack, StorageType storageType) {
if (nodesRequired <= 0) {
return Collections.emptySet();
}
@@ -165,7 +166,7 @@ private Set<DatanodeDetails> chooseNodesFromRacks(List<Node> racks,
continue;
}
Node node = chooseNode(rack.getNetworkFullPath(), unavailableNodes,
- metadataSizeRequired, dataSizeRequired);
+ metadataSizeRequired, dataSizeRequired, storageType);
if (node != null) {
chosenNodes.add((DatanodeDetails) node);
rackCntMap.merge(rack, 1, Math::addExact);
@@ -218,7 +219,7 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
final List<DatanodeDetails> excludedNodes,
final List<DatanodeDetails> favoredNodes,
final int nodesRequired, final long metadataSizeRequired,
- final long dataSizeRequired) throws SCMException {
+ final long dataSizeRequired, StorageType storageType) throws SCMException {
if (nodesRequired <= 0) {
String errorMsg = "num of nodes required to choose should bigger" +
"than 0, but the given num is " + nodesRequired;
@@ -252,7 +253,7 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
// Generate mutableFavoredNodes, only stores valid favoredNodes
for (DatanodeDetails datanodeDetails : favoredNodes) {
if (isValidNode(datanodeDetails, metadataSizeRequired,
- dataSizeRequired)) {
+ dataSizeRequired, storageType)) {
mutableFavoredNodes.add(datanodeDetails);
}
}
@@ -318,7 +319,7 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
chooseNodesFromRacks(racks, unavailableNodes,
mutableFavoredNodes, additionalRacksRequired,
metadataSizeRequired, dataSizeRequired, maxReplicasPerRack,
- usedRacksCntMap, maxReplicasPerRack));
+ usedRacksCntMap, maxReplicasPerRack, storageType));
if (chosenNodes.size() < additionalRacksRequired) {
String reason = "Chosen nodes size from Unique Racks: " + chosenNodes
@@ -341,7 +342,7 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
chosenNodes.addAll(chooseNodesFromRacks(racks, unavailableNodes,
mutableFavoredNodes, nodesRequired - chosenNodes.size(),
metadataSizeRequired, dataSizeRequired,
- Integer.MAX_VALUE, usedRacksCntMap, maxReplicasPerRack));
+ Integer.MAX_VALUE, usedRacksCntMap, maxReplicasPerRack, storageType));
}
List<DatanodeDetails> result = new ArrayList<>(chosenNodes);
if (nodesRequired != chosenNodes.size()) {
@@ -446,7 +447,7 @@ public DatanodeDetails chooseNode(List<DatanodeDetails> healthyNodes) {
* @return the chosen datanode.
*/
private Node chooseNode(String scope, List<Node> excludedNodes,
- long metadataSizeRequired, long dataSizeRequired) {
+ long metadataSizeRequired, long dataSizeRequired, StorageType storageType) {
int maxRetry = INNER_LOOP_MAX_RETRY;
while (true) {
if (metrics != null) {
@@ -468,7 +469,7 @@ private Node chooseNode(String scope, List<Node> excludedNodes,
if (node != null) {
DatanodeDetails datanodeDetails = (DatanodeDetails) node;
if (isValidNode(datanodeDetails, metadataSizeRequired,
- dataSizeRequired)) {
+ dataSizeRequired, storageType)) {
if (metrics != null) {
metrics.incrDatanodeChooseSuccessCount();
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRandom.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRandom.java
index 65ca4f3914a..846dd5bbe4f 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRandom.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRandom.java
@@ -19,6 +19,7 @@
import com.google.common.annotations.VisibleForTesting;
import java.util.List;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.scm.PlacementPolicy;
@@ -62,27 +63,30 @@ public SCMContainerPlacementRandom(final NodeManager nodeManager,
/**
* Choose datanodes called by the SCM to choose the datanode.
*
- * @param usedNodes - list of the datanodes to already chosen in the
- * pipeline.
- * @param excludedNodes - list of the datanodes to exclude.
- * @param favoredNodes - list of nodes preferred.
- * @param nodesRequired - number of datanodes required.
- * @param dataSizeRequired - size required for the container.
+ * @param usedNodes - list of the datanodes to already chosen in the
+ * pipeline.
+ * @param excludedNodes - list of the datanodes to exclude.
+ * @param favoredNodes - list of nodes preferred.
+ * @param nodesRequired - number of datanodes required.
* @param metadataSizeRequired - size required for Ratis metadata.
+ * @param dataSizeRequired - size required for the container.
+ * @param storageType - StorageType required for the container.
* @return List of Datanodes.
- * @throws SCMException SCMException
+ * @throws SCMException SCMException
*/
@Override
protected List<DatanodeDetails> chooseDatanodesInternal(
List<DatanodeDetails> usedNodes,
List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> favoredNodes, final int nodesRequired,
- long metadataSizeRequired, long dataSizeRequired)
+ long metadataSizeRequired, long dataSizeRequired,
+ StorageType storageType)
throws SCMException {
metrics.incrDatanodeRequestCount(nodesRequired);
List<DatanodeDetails> healthyNodes =
super.chooseDatanodesInternal(usedNodes, excludedNodes, favoredNodes,
- nodesRequired, metadataSizeRequired, dataSizeRequired);
+ nodesRequired, metadataSizeRequired, dataSizeRequired,
+ storageType);
if (healthyNodes.size() == nodesRequired) {
return healthyNodes;
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ECUnderReplicationHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ECUnderReplicationHandler.java
index 5d41952b58e..de5151a40b8 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ECUnderReplicationHandler.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ECUnderReplicationHandler.java
@@ -32,6 +32,7 @@
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.StorageUnit;
@@ -301,9 +302,9 @@ private int processMissingIndexes(
: excludedNodes;
// placement with overloaded nodes excluded
+ // TODO StoragePolicy replace this StorageType with container actual StorageType
final List<DatanodeDetails> selectedDatanodes = getTargetDatanodes(
- container, expectedTargetCount, usedNodes, excludedOrOverloadedNodes
- );
+ container, expectedTargetCount, usedNodes, excludedOrOverloadedNodes, StorageType.DEFAULT);
final int targetCount = selectedDatanodes.size();
if (hasOverloaded &&
@@ -313,8 +314,9 @@ private int processMissingIndexes(
!recoveryIsCritical) {
// check if placement exists when overloaded nodes are not excluded
+ // TODO StoragePolicy replace this StorageType with container actual StorageType
final List<DatanodeDetails> targetsMaybeOverloaded = getTargetDatanodes(
- container, expectedTargetCount, usedNodes, excludedNodes);
+ container, expectedTargetCount, usedNodes, excludedNodes, StorageType.DEFAULT);
if (targetsMaybeOverloaded.size() == expectedTargetCount) {
final int overloadedCount = expectedTargetCount - targetCount;
@@ -406,12 +408,12 @@ private int processMissingIndexes(
private List<DatanodeDetails> getTargetDatanodes(
ContainerInfo container, int requiredNodes,
List<DatanodeDetails> usedNodes,
- List<DatanodeDetails> excludedNodes
- ) throws SCMException {
+ List<DatanodeDetails> excludedNodes,
+ StorageType storageType) throws SCMException {
return ReplicationManagerUtil.getTargetDatanodes(
containerPlacement, requiredNodes,
usedNodes, excludedNodes,
- currentContainerSize, container);
+ currentContainerSize, container, storageType);
}
/**
@@ -431,8 +433,9 @@ private int processDecommissioningIndexes(
if (!decomIndexes.isEmpty()) {
LOG.debug("Processing decommissioning indexes {} for container {}.",
decomIndexes, container.containerID());
+ // TODO StoragePolicy replace this StorageType with container actual StorageType
final List<DatanodeDetails> selectedDatanodes = getTargetDatanodes(
- container, decomIndexes.size(), usedNodes, excludedNodes);
+ container, decomIndexes.size(), usedNodes, excludedNodes, StorageType.DEFAULT);
ContainerPlacementStatus placementStatusWithSelectedTargets =
validatePlacement(container, availableSourceNodes, selectedDatanodes);
@@ -528,9 +531,9 @@ private int processMaintenanceOnlyIndexes(
LOG.debug("Number of maintenance replicas of container {} that need " +
"additional copies: {}.", container.containerID(),
additionalMaintenanceCopiesNeeded);
+ // TODO StoragePolicy replace this StorageType with container actual StorageType
List<DatanodeDetails> targets = getTargetDatanodes(
- container, maintIndexes.size(), usedNodes, excludedNodes
- );
+ container, maintIndexes.size(), usedNodes, excludedNodes, StorageType.DEFAULT);
usedNodes.addAll(targets);
Iterator<DatanodeDetails> iterator = targets.iterator();
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandler.java
index c4be97384e8..2c4538a7daa 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandler.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandler.java
@@ -24,6 +24,7 @@
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.StorageUnit;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -153,11 +154,12 @@ public int processAndSendCommands(
int requiredNodes = replicasToBeReplicated.size();
+ // TODO StoragePolicy replace this StorageType with container actual StorageType
List<DatanodeDetails> targetDatanodes = ReplicationManagerUtil
.getTargetDatanodes(containerPlacement, requiredNodes,
excludedAndUsedNodes.getUsedNodes(),
excludedAndUsedNodes.getExcludedNodes(), currentContainerSize,
- container);
+ container, StorageType.DEFAULT);
List<DatanodeDetails> availableSources = sources.stream()
.map(ContainerReplica::getDatanodeDetails)
.collect(Collectors.toList());
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/QuasiClosedStuckUnderReplicationHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/QuasiClosedStuckUnderReplicationHandler.java
index 91ab321859a..e9d7f5c1fbc 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/QuasiClosedStuckUnderReplicationHandler.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/QuasiClosedStuckUnderReplicationHandler.java
@@ -23,6 +23,7 @@
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.StorageUnit;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -164,9 +165,9 @@ private List<DatanodeDetails> getTargets(ContainerInfo containerInfo,
LOG.debug("UsedList: {}, size {}. ExcludeList: {}, size: {}. ",
used, used.size(), excluded, excluded.size());
-
+ // TODO StoragePolicy replace this StorageType with container actual StorageType
return ReplicationManagerUtil.getTargetDatanodes(placementPolicy,
- additionalRequired, used, excluded, currentContainerSize, containerInfo);
+ additionalRequired, used, excluded, currentContainerSize, containerInfo, StorageType.DEFAULT);
}
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/RatisUnderReplicationHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/RatisUnderReplicationHandler.java
index 8853e23e652..84fc2ea80f9 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/RatisUnderReplicationHandler.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/RatisUnderReplicationHandler.java
@@ -27,6 +27,7 @@
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.StorageUnit;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -237,9 +238,10 @@ of other replicas, including UNHEALTHY replicas that are not pending delete (bec
int numCommandsSent = 0;
for (ContainerReplica replica : sources) {
// find a target for each source and send replicate command
+ // TODO StoragePolicy replace this StorageType with container actual StorageType
final List<DatanodeDetails> target =
ReplicationManagerUtil.getTargetDatanodes(placementPolicy, 1, excludedAndUsedNodes.getUsedNodes(),
- excludedAndUsedNodes.getExcludedNodes(), currentContainerSize, container);
+ excludedAndUsedNodes.getExcludedNodes(), currentContainerSize, container, StorageType.DEFAULT);
int count = 0;
try {
count = sendReplicationCommands(container, ImmutableList.of(replica.getDatanodeDetails()), target);
@@ -459,10 +461,10 @@ private List<DatanodeDetails> getTargets(
LOG.debug("UsedList: {}, size {}. ExcludeList: {}, size: {}. ",
used, used.size(), excluded, excluded.size());
-
+ // TODO StoragePolicy replace this StorageType with container actual StorageType
return ReplicationManagerUtil.getTargetDatanodes(placementPolicy,
replicaCount.additionalReplicaNeeded(), used, excluded,
- currentContainerSize, replicaCount.getContainer());
+ currentContainerSize, replicaCount.getContainer(), StorageType.DEFAULT);
}
private int sendReplicationCommands(
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManagerUtil.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManagerUtil.java
index b08bdafd788..18685f72bf4 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManagerUtil.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManagerUtil.java
@@ -28,6 +28,7 @@
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.DatanodeID;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
@@ -71,22 +72,24 @@ private static String formatDatanodeDetails(List<DatanodeDetails> dns) {
* policy will be called again. This will continue until the placement policy
* is able to select enough nodes or the number of nodes requested is reduced
* to zero when an exception will be thrown.
- * @param policy The placement policy to use to select nodes.
- * @param requiredNodes The number of nodes required
- * @param usedNodes Any nodes already used by the container
- * @param excludedNodes Any Excluded nodes which cannot be selected
+ *
+ * @param policy The placement policy to use to select nodes.
+ * @param requiredNodes The number of nodes required
+ * @param usedNodes Any nodes already used by the container
+ * @param excludedNodes Any Excluded nodes which cannot be selected
* @param defaultContainerSize The cluster default max container size
- * @param container The container to select new replicas for
+ * @param container The container to select new replicas for
+ * @param storageType The storage type supported by target nodes
* @return A list of up to requiredNodes datanodes to use as targets for new
- * replicas. Note the number of nodes returned may be less than the
- * number of nodes requested if the placement policy is unable to
- * return enough nodes.
+ * replicas. Note the number of nodes returned may be less than the
+ * number of nodes requested if the placement policy is unable to
+ * return enough nodes.
* @throws SCMException If no nodes can be selected.
*/
public static List<DatanodeDetails> getTargetDatanodes(PlacementPolicy policy,
int requiredNodes, List<DatanodeDetails> usedNodes,
List<DatanodeDetails> excludedNodes, long defaultContainerSize,
- ContainerInfo container) throws SCMException {
+ ContainerInfo container, StorageType storageType) throws SCMException {
// Ensure that target datanodes have enough space to hold a complete
// container.
@@ -98,10 +101,10 @@ public static List<DatanodeDetails> getTargetDatanodes(PlacementPolicy policy,
try {
if (usedNodes == null) {
return policy.chooseDatanodes(excludedNodes, null,
- mutableRequiredNodes, 0, dataSizeRequired);
+ mutableRequiredNodes, 0, dataSizeRequired, storageType);
} else {
return policy.chooseDatanodes(usedNodes, excludedNodes, null,
- mutableRequiredNodes, 0, dataSizeRequired);
+ mutableRequiredNodes, 0, dataSizeRequired, storageType);
}
} catch (IOException e) {
LOG.debug("Placement policy was not able to return {} nodes for " +
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/ECPipelineProvider.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/ECPipelineProvider.java
index 1560c9890c5..dc9618a1cb4 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/ECPipelineProvider.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/ECPipelineProvider.java
@@ -25,7 +25,10 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
+import org.apache.hadoop.hdds.client.StorageTier;
+import org.apache.hadoop.hdds.client.StorageTierUtil;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.StorageUnit;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -76,19 +79,21 @@ public ECPipelineProvider(NodeManager nodeManager,
}
@Override
- public synchronized Pipeline create(ECReplicationConfig replicationConfig)
+ public synchronized Pipeline create(ECReplicationConfig replicationConfig, StorageTier storageTier)
throws IOException {
return create(replicationConfig, Collections.emptyList(),
- Collections.emptyList());
+ Collections.emptyList(), storageTier);
}
@Override
protected Pipeline create(ECReplicationConfig replicationConfig,
- List<DatanodeDetails> excludedNodes, List<DatanodeDetails> favoredNodes)
+ List<DatanodeDetails> excludedNodes, List<DatanodeDetails> favoredNodes,
+ StorageTier storageTier)
throws IOException {
+ StorageType storageType = StorageTierUtil.getStorageTypeForUniformStorageTier(storageTier, replicationConfig);
List<DatanodeDetails> dns = placementPolicy
.chooseDatanodes(excludedNodes, favoredNodes,
- replicationConfig.getRequiredNodes(), 0, this.containerSizeBytes);
+ replicationConfig.getRequiredNodes(), 0, this.containerSizeBytes, storageType);
return create(replicationConfig, dns);
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineFactory.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineFactory.java
index e55947d0d4d..9da5bac2f28 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineFactory.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineFactory.java
@@ -24,6 +24,7 @@
import java.util.Map;
import java.util.Set;
import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.client.StorageTier;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
@@ -82,8 +83,9 @@ public Pipeline create(
ReplicationConfig replicationConfig, List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> favoredNodes)
throws IOException {
+ // TODO StoragePolicy replace this StorageTier with the passed StorageTier
Pipeline pipeline = providers.get(replicationConfig.getReplicationType())
- .create(replicationConfig, excludedNodes, favoredNodes);
+ .create(replicationConfig, excludedNodes, favoredNodes, StorageTier.getDefaultTier());
checkPipeline(pipeline);
return pipeline;
}
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 615d466f6e4..e2f880121fc 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
@@ -24,6 +24,7 @@
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.RatisReplicationConfig;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -142,7 +143,7 @@ protected int getMaxReplicasPerRack(int numReplicas, int numberOfRacks) {
List<DatanodeDetails> filterViableNodes(
List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> usedNodes, int nodesRequired,
- long metadataSizeRequired, long dataSizeRequired)
+ long metadataSizeRequired, long dataSizeRequired, StorageType storageType)
throws SCMException {
// get nodes in HEALTHY state
List<DatanodeDetails> healthyNodes =
@@ -155,8 +156,8 @@ List<DatanodeDetails> filterViableNodes(
.FAILED_TO_FIND_HEALTHY_NODES);
}
- healthyNodes = filterNodesWithSpace(healthyNodes, nodesRequired,
- metadataSizeRequired);
+ healthyNodes = filterNodesWithSpaceAndStorageType(healthyNodes, nodesRequired,
+ metadataSizeRequired, dataSizeRequired, storageType);
boolean multipleRacks = multipleRacksAvailable(healthyNodes);
int excludedNodesSize = 0;
if (excludedNodes != null) {
@@ -171,8 +172,8 @@ List<DatanodeDetails> filterViableNodes(
if (initialHealthyNodesCount < nodesRequired) {
msg = String.format("Pipeline creation failed due to no sufficient" +
- " healthy datanodes. Required %d. Found %d. Excluded %d.",
- nodesRequired, initialHealthyNodesCount, excludedNodesSize);
+ " healthy datanodes. Required %d StorageType Required %s. Found %d. Excluded %d.",
+ nodesRequired, storageType, initialHealthyNodesCount, excludedNodesSize);
LOG.debug(msg);
throw new SCMException(msg,
SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
@@ -187,10 +188,10 @@ List<DatanodeDetails> filterViableNodes(
if (LOG.isDebugEnabled()) {
LOG.debug("Unable to find enough nodes that meet the criteria that" +
" cannot engage in more than" + datanodePipelineLimit +
- " pipelines. Nodes required: " + nodesRequired + " Excluded: " +
- excludedNodesSize + " Found:" +
- healthyList.size() + " healthy nodes count in NodeManager: " +
- initialHealthyNodesCount);
+ " pipelines. StorageType required: " + storageType + " Nodes required: " +
+ nodesRequired + " Excluded: " +
+ excludedNodesSize + " Found:" + healthyList.size() +
+ " healthy nodes count in NodeManager: " + initialHealthyNodesCount);
}
msg = String.format("Pipeline creation failed because nodes are engaged" +
" in other pipelines and every node can only be engaged in" +
@@ -242,6 +243,7 @@ private boolean multipleRacksAvailable(List<DatanodeDetails> dns) {
* @param nodesRequired - number of datanodes required.
* @param dataSizeRequired - size required for the container.
* @param metadataSizeRequired - size required for Ratis metadata.
+ * @param storageType - StorageType required for the container.
* @return a list of chosen datanodeDetails
* @throws SCMException when chosen nodes are not enough in numbers
*/
@@ -249,12 +251,13 @@ private boolean multipleRacksAvailable(List<DatanodeDetails> dns) {
protected List<DatanodeDetails> chooseDatanodesInternal(
List<DatanodeDetails> usedNodes, List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> favoredNodes,
- int nodesRequired, long metadataSizeRequired, long dataSizeRequired)
+ int nodesRequired, long metadataSizeRequired, long dataSizeRequired,
+ StorageType storageType)
throws SCMException {
// Get a list of viable nodes based on criteria
// and make sure excludedNodes are excluded from list.
List<DatanodeDetails> healthyNodes = filterViableNodes(excludedNodes,
- usedNodes, nodesRequired, metadataSizeRequired, dataSizeRequired);
+ usedNodes, nodesRequired, metadataSizeRequired, dataSizeRequired, storageType);
// Randomly picks nodes when all nodes are equal or factor is ONE.
// This happens when network topology is absent or
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 978ebbb0406..983da957ded 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
@@ -22,7 +22,10 @@
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.client.StorageTier;
+import org.apache.hadoop.hdds.client.StorageTierUtil;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.scm.SCMCommonPlacementPolicy;
import org.apache.hadoop.hdds.scm.container.ContainerReplica;
@@ -61,11 +64,12 @@ public PipelineStateManager getPipelineStateManager() {
return stateManager;
}
- protected abstract Pipeline create(REPLICATION_CONFIG replicationConfig)
- throws IOException;
+ protected abstract Pipeline create(REPLICATION_CONFIG replicationConfig,
+ StorageTier storageTier) throws IOException;
protected abstract Pipeline create(REPLICATION_CONFIG replicationConfig,
- List<DatanodeDetails> excludedNodes, List<DatanodeDetails> favoredNodes)
+ List<DatanodeDetails> excludedNodes, List<DatanodeDetails> favoredNodes,
+ StorageTier storageTier)
throws IOException;
protected abstract Pipeline create(
@@ -82,13 +86,15 @@ protected abstract Pipeline createForRead(
List<DatanodeDetails> pickNodesNotUsed(REPLICATION_CONFIG replicationConfig,
long metadataSizeRequired,
- long dataSizeRequired)
+ long dataSizeRequired, StorageTier storageTier)
throws SCMException {
+ StorageTierUtil.validateNotEmpty(storageTier);
+ StorageType storageType = StorageTierUtil.getStorageTypeForUniformStorageTier(storageTier, replicationConfig);
int nodesRequired = replicationConfig.getRequiredNodes();
List<DatanodeDetails> healthyDNs = pickAllNodesNotUsed(replicationConfig);
List<DatanodeDetails> healthyDNsWithSpace = healthyDNs.stream()
.filter(dn -> SCMCommonPlacementPolicy.hasEnoughSpace(
- dn, metadataSizeRequired, nodeManager))
+ dn, metadataSizeRequired, dataSizeRequired, storageType, nodeManager))
.limit(nodesRequired)
.collect(Collectors.toList());
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java
index 35512b2e4f3..e9cccd72b5d 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java
@@ -25,7 +25,10 @@
import java.util.Collections;
import java.util.List;
import java.util.Set;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.client.StorageTier;
+import org.apache.hadoop.hdds.client.StorageTierUtil;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.StorageUnit;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -140,15 +143,16 @@ public LeaderChoosePolicy getLeaderChoosePolicy() {
}
@Override
- public synchronized Pipeline create(RatisReplicationConfig replicationConfig)
+ public synchronized Pipeline create(RatisReplicationConfig replicationConfig,
+ StorageTier storageTier)
throws IOException {
return create(replicationConfig, Collections.emptyList(),
- Collections.emptyList());
+ Collections.emptyList(), storageTier);
}
@Override
public synchronized Pipeline create(RatisReplicationConfig replicationConfig,
- List<DatanodeDetails> excludedNodes, List<DatanodeDetails> favoredNodes)
+ List<DatanodeDetails> excludedNodes, List<DatanodeDetails> favoredNodes, StorageTier storageTier)
throws IOException {
if (exceedPipelineNumberLimit(replicationConfig)) {
String limitInfo = (datanodePipelineLimit > 0)
@@ -167,10 +171,12 @@ public synchronized Pipeline create(RatisReplicationConfig replicationConfig,
replicationConfig.getReplicationFactor();
switch (factor) {
case ONE:
- dns = pickNodesNotUsed(replicationConfig, minRatisVolumeSizeBytes, containerSizeBytes);
+ dns = pickNodesNotUsed(replicationConfig, minRatisVolumeSizeBytes, containerSizeBytes, storageTier);
break;
case THREE:
- dns = chooseThreeFactorDatanodes(excludedNodes, favoredNodes, factor.getNumber());
+ StorageTierUtil.validateNotEmpty(storageTier);
+ StorageType storageType = StorageTierUtil.getStorageTypeForUniformStorageTier(storageTier, replicationConfig);
+ dns = chooseThreeFactorDatanodes(excludedNodes, favoredNodes, factor.getNumber(), storageType);
break;
default:
@@ -223,7 +229,8 @@ public Pipeline createForRead(
}
private List<DatanodeDetails> chooseThreeFactorDatanodes(
- List<DatanodeDetails> excludedNodes, List<DatanodeDetails> favoredNodes, int requiredNode)
+ List<DatanodeDetails> excludedNodes, List<DatanodeDetails> favoredNodes, int requiredNode,
+ StorageType storageType)
throws IOException {
final NodeManager nodeManager = getNodeManager();
final PipelineStateManager stateManager = getPipelineStateManager();
@@ -252,7 +259,7 @@ private List<DatanodeDetails> chooseThreeFactorDatanodes(
try {
return placementPolicy.chooseDatanodes(additionalExcludedNodes,
favoredNodes, requiredNode, minRatisVolumeSizeBytes,
- containerSizeBytes);
+ containerSizeBytes, storageType);
} catch (SCMException scmException) {
LOG.debug("Failed to allocate datanodes with strict exclusion (non-streaming nodes excluded)." +
" Falling back.", scmException);
@@ -260,7 +267,7 @@ private List<DatanodeDetails> chooseThreeFactorDatanodes(
}
return placementPolicy.chooseDatanodes(excludedNodes,
favoredNodes, requiredNode, minRatisVolumeSizeBytes,
- containerSizeBytes);
+ containerSizeBytes, storageType);
}
/**
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SimplePipelineProvider.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SimplePipelineProvider.java
index 50ed2b015b9..1a41b183ed3 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SimplePipelineProvider.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SimplePipelineProvider.java
@@ -21,10 +21,16 @@
import java.util.Collections;
import java.util.List;
import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.ReplicationConfig;
import org.apache.hadoop.hdds.client.StandaloneReplicationConfig;
+import org.apache.hadoop.hdds.client.StorageTier;
+import org.apache.hadoop.hdds.client.StorageTierUtil;
+import org.apache.hadoop.hdds.client.StorageTypeUtils;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.scm.container.ContainerReplica;
+import org.apache.hadoop.hdds.scm.node.DatanodeInfo;
import org.apache.hadoop.hdds.scm.node.NodeManager;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline.PipelineState;
@@ -40,23 +46,30 @@ public SimplePipelineProvider(NodeManager nodeManager,
}
@Override
- public Pipeline create(StandaloneReplicationConfig replicationConfig)
+ public Pipeline create(StandaloneReplicationConfig replicationConfig, StorageTier storageTier)
throws IOException {
+ StorageTierUtil.validateNotEmpty(storageTier);
return create(replicationConfig, Collections.emptyList(),
- Collections.emptyList());
+ Collections.emptyList(), storageTier);
}
@Override
public Pipeline create(StandaloneReplicationConfig replicationConfig,
- List<DatanodeDetails> excludedNodes, List<DatanodeDetails> favoredNodes)
+ List<DatanodeDetails> excludedNodes, List<DatanodeDetails> favoredNodes, StorageTier storageTier)
throws IOException {
+ StorageType storageType = StorageTierUtil.getStorageTypeForUniformStorageTier(storageTier, replicationConfig);
List<DatanodeDetails> dns = pickNodesNotUsed(replicationConfig);
+ dns = dns.stream().filter(dn ->
+ ((DatanodeInfo) dn).getStorageReports().stream()
+ .anyMatch(reportProto ->
+ StorageTypeUtils.getFromProtobuf(reportProto.getStorageType()).equals(storageType)))
+ .collect(Collectors.toList());
int available = dns.size();
int required = replicationConfig.getRequiredNodes();
if (available < required) {
String msg = String.format(
- "Cannot create pipeline of factor %d using %d nodes.",
- required, available);
+ "Cannot create pipeline of factor %d using %d nodes storageTier %s",
+ required, available, storageTier);
throw new InsufficientDatanodesException(required, available, msg);
}
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 9df7c4f69d6..4c5d1539899 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
@@ -47,6 +47,7 @@
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -466,14 +467,14 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> favoredNodes,
int nodesRequired, long metadataSizeRequired,
- long dataSizeRequired) {
+ long dataSizeRequired, StorageType storageType) {
usedNodesIdentity.set(usedNodesPassed(usedNodes));
return null;
}
};
- dummyPlacementPolicy.chooseDatanodes(null, null, 1, 1, 1);
+ dummyPlacementPolicy.chooseDatanodes(null, null, 1, 1, 1, StorageType.DEFAULT);
assertFalse(usedNodesIdentity.get());
- dummyPlacementPolicy.chooseDatanodes(null, null, null, 1, 1, 1);
+ dummyPlacementPolicy.chooseDatanodes(null, null, null, 1, 1, 1, StorageType.DEFAULT);
assertTrue(usedNodesIdentity.get());
}
@@ -499,13 +500,14 @@ public void testDatanodeIsInvalidWhenNoSlotsAvailable() {
.thenReturn(Collections.singletonList(metaReport));
// Space check now uses PendingContainerTracker.hasAvailableSpace:
- // slot available → isValidNode returns true
+ // slot available → isValidNode returns true.
+ // storageType == null skips the tier-aware check and relies on the slot check.
when(nodeMngr.hasAvailableSpace(datanodeInfo)).thenReturn(true);
- assertTrue(placementPolicy.isValidNode(datanodeDetails, 100, 4000));
+ assertTrue(placementPolicy.isValidNode(datanodeDetails, 100, 4000, null));
// No slot available (all pending) → isValidNode returns false
when(nodeMngr.hasAvailableSpace(datanodeInfo)).thenReturn(false);
- assertFalse(placementPolicy.isValidNode(datanodeDetails, 100, 4000));
+ assertFalse(placementPolicy.isValidNode(datanodeDetails, 100, 4000, null));
}
/**
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 46bccad1119..ff0fb103f26 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
@@ -17,6 +17,7 @@
package org.apache.hadoop.hdds.scm.container;
+import static org.apache.hadoop.hdds.client.StorageTypeUtils.getStorageTypeProto;
import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState.DEAD;
import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState.HEALTHY;
import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState.STALE;
@@ -37,6 +38,7 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.DatanodeID;
@@ -117,6 +119,7 @@ public class MockNodeManager implements NodeManager {
private int numPipelinePerDatanode;
private PendingContainerTracker pendingContainerTracker;
private final OzoneConfiguration conf = new OzoneConfiguration();
+ private StorageType storageType = null;
{
this.healthyNodes = new LinkedList<>();
@@ -131,6 +134,13 @@ public class MockNodeManager implements NodeManager {
HddsTestUtils.ROLL_INTERVAL_MS_DEFAULT, null);
}
+ public MockNodeManager(NetworkTopologyImpl clusterMap,
+ List<DatanodeDetails> nodes,
+ boolean initializeFakeNodes, int nodeCount, StorageType storageType) {
+ this(clusterMap, nodes, initializeFakeNodes, nodeCount);
+ this.storageType = storageType;
+ }
+
public MockNodeManager(NetworkTopologyImpl clusterMap,
List<DatanodeDetails> nodes,
boolean initializeFakeNodes, int nodeCount) {
@@ -155,6 +165,12 @@ public MockNodeManager(NetworkTopologyImpl clusterMap,
NUM_PIPELINE_PER_METADATA_DISK;
}
+ public MockNodeManager(boolean initializeFakeNodes, int nodeCount, StorageType storageType) {
+ this(new NetworkTopologyImpl(new OzoneConfiguration()), new ArrayList<>(),
+ initializeFakeNodes, nodeCount);
+ this.storageType = storageType;
+ }
+
public MockNodeManager(boolean initializeFakeNodes, int nodeCount) {
this(new NetworkTopologyImpl(new OzoneConfiguration()), new ArrayList<>(),
initializeFakeNodes, nodeCount);
@@ -284,7 +300,8 @@ private List<DatanodeDetails> getDatanodeDetails(
long remaining = nodeMetricMap.get(dd).getRemaining().get();
StorageReportProto storage1 = HddsTestUtils.createStorageReport(
di.getID(), "/data1-" + di.getID(),
- capacity, used, remaining, null);
+ capacity, used, remaining,
+ storageType == null ? null : getStorageTypeProto(storageType));
MetadataStorageReportProto metaStorage1 =
HddsTestUtils.createMetadataStorageReport(
"/metadata1-" + di.getID(), capacity, used, remaining, null);
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 5e6f0ef3959..869877dc04c 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
@@ -37,6 +37,7 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.conf.StorageUnit;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -155,7 +156,7 @@ public void testRackAwarePolicy() throws IOException {
int nodeNum = 3;
List<DatanodeDetails> datanodeDetails =
- policy.chooseDatanodes(null, null, nodeNum, 15, 15);
+ policy.chooseDatanodes(null, null, nodeNum, 15, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertTrue(cluster.isSameParent(datanodeDetails.get(0),
datanodeDetails.get(1)));
@@ -193,7 +194,7 @@ public List<DatanodeDetails> chooseDatanodes(
List<DatanodeDetails> usedNodes,
List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> favoredNodes,
- int nodesRequired, long metadataSizeRequired, long dataSizeRequired) {
+ int nodesRequired, long metadataSizeRequired, long dataSizeRequired, StorageType storageType) {
return null;
}
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 b86bf3e58d0..8330e14790a 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
@@ -30,6 +30,7 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.conf.StorageUnit;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -140,7 +141,7 @@ public void chooseDatanodes() throws SCMException {
//when
List<DatanodeDetails> datanodeDetails = scmContainerPlacementRandom
- .chooseDatanodes(existingNodes, null, 1, 15, 15);
+ .chooseDatanodes(existingNodes, null, 1, 15, 15, StorageType.DEFAULT);
//then
assertEquals(1, datanodeDetails.size());
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 5b1961df107..c0c3d5d07aa 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
@@ -17,6 +17,7 @@
package org.apache.hadoop.hdds.scm.container.placement.algorithms;
+import static org.apache.hadoop.hdds.client.StorageTypeUtils.getStorageTypeProto;
import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState.DECOMMISSIONED;
import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState.IN_SERVICE;
import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState.HEALTHY;
@@ -42,6 +43,7 @@
import java.util.stream.IntStream;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.conf.StorageUnit;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -90,7 +92,7 @@ private static IntStream numDatanodes() {
return IntStream.rangeClosed(3, 15);
}
- private void setup(int datanodeCount) {
+ private void setup(int datanodeCount, StorageType storageType) {
//initialize network topology instance
conf = new OzoneConfiguration();
// We are using small units here
@@ -118,7 +120,7 @@ private void setup(int datanodeCount) {
StorageReportProto storage1 = HddsTestUtils.createStorageReport(
datanodeInfo.getID(), "/data1-" + datanodeInfo.getID(),
- STORAGE_CAPACITY, 0, 100L, null);
+ STORAGE_CAPACITY, 0, 100L, getStorageTypeProto(storageType));
MetadataStorageReportProto metaStorage1 =
HddsTestUtils.createMetadataStorageReport(
"/metadata1-" + datanodeInfo.getID(),
@@ -134,39 +136,39 @@ private void setup(int datanodeCount) {
StorageReportProto storage2 = HddsTestUtils.createStorageReport(
dnInfos.get(2).getID(),
"/data1-" + datanodes.get(2).getID(),
- STORAGE_CAPACITY, 90L, 10L, null);
+ STORAGE_CAPACITY, 90L, 10L, getStorageTypeProto(storageType));
dnInfos.get(2).updateStorageReports(
new ArrayList<>(Arrays.asList(storage2)));
StorageReportProto storage3 = HddsTestUtils.createStorageReport(
dnInfos.get(3).getID(),
"/data1-" + dnInfos.get(3).getID(),
- STORAGE_CAPACITY, 80L, 20L, null);
+ STORAGE_CAPACITY, 80L, 20L, getStorageTypeProto(storageType));
dnInfos.get(3).updateStorageReports(
new ArrayList<>(Arrays.asList(storage3)));
StorageReportProto storage4 = HddsTestUtils.createStorageReport(
dnInfos.get(4).getID(),
"/data1-" + dnInfos.get(4).getID(),
- STORAGE_CAPACITY, 70L, 30L, null);
+ STORAGE_CAPACITY, 70L, 30L, getStorageTypeProto(storageType));
dnInfos.get(4).updateStorageReports(
new ArrayList<>(Arrays.asList(storage4)));
} else if (datanodeCount > 3) {
StorageReportProto storage2 = HddsTestUtils.createStorageReport(
dnInfos.get(2).getID(),
"/data1-" + dnInfos.get(2).getID(),
- STORAGE_CAPACITY, 90L, 10L, null);
+ STORAGE_CAPACITY, 90L, 10L, getStorageTypeProto(storageType));
dnInfos.get(2).updateStorageReports(
new ArrayList<>(Arrays.asList(storage2)));
StorageReportProto storage3 = HddsTestUtils.createStorageReport(
dnInfos.get(3).getID(),
"/data1-" + dnInfos.get(3).getID(),
- STORAGE_CAPACITY, 80L, 20L, null);
+ STORAGE_CAPACITY, 80L, 20L, getStorageTypeProto(storageType));
dnInfos.get(3).updateStorageReports(
new ArrayList<>(Arrays.asList(storage3)));
} else if (datanodeCount > 2) {
StorageReportProto storage2 = HddsTestUtils.createStorageReport(
dnInfos.get(2).getID(),
"/data1-" + dnInfos.get(2).getID(),
- STORAGE_CAPACITY, 84L, 16L, null);
+ STORAGE_CAPACITY, 84L, 16L, getStorageTypeProto(storageType));
dnInfos.get(2).updateStorageReports(
new ArrayList<>(Arrays.asList(storage2)));
}
@@ -204,24 +206,25 @@ public void teardown() {
@MethodSource("numDatanodes")
public void chooseNodeWithNoExcludedNodes(int datanodeCount)
throws SCMException {
- setup(datanodeCount);
+ StorageType storageType = StorageType.DEFAULT;
+ setup(datanodeCount, storageType);
// test choose new datanodes for new pipeline cases
// 1 replica
int nodeNum = 1;
List<DatanodeDetails> datanodeDetails =
- policy.chooseDatanodes(null, null, nodeNum, 0, 15);
+ policy.chooseDatanodes(null, null, nodeNum, 0, 15, storageType);
assertEquals(nodeNum, datanodeDetails.size());
// 2 replicas
nodeNum = 2;
- datanodeDetails = policy.chooseDatanodes(null, null, nodeNum, 0, 15);
+ datanodeDetails = policy.chooseDatanodes(null, null, nodeNum, 0, 15, storageType);
assertEquals(nodeNum, datanodeDetails.size());
assertTrue(cluster.isSameParent(datanodeDetails.get(0),
datanodeDetails.get(1)) || (datanodeCount % NODE_PER_RACK == 1));
// 3 replicas
nodeNum = 3;
- datanodeDetails = policy.chooseDatanodes(null, null, nodeNum, 0, 15);
+ datanodeDetails = policy.chooseDatanodes(null, null, nodeNum, 0, 15, storageType);
assertEquals(nodeNum, datanodeDetails.size());
// requires at least 2 racks for following statement
assumeTrue(datanodeCount > NODE_PER_RACK &&
@@ -235,7 +238,7 @@ public void chooseNodeWithNoExcludedNodes(int datanodeCount)
// 4 replicas
nodeNum = 4;
- datanodeDetails = policy.chooseDatanodes(null, null, nodeNum, 0, 15);
+ datanodeDetails = policy.chooseDatanodes(null, null, nodeNum, 0, 15, storageType);
assertEquals(nodeNum, datanodeDetails.size());
// requires at least 2 racks and enough datanodes for following statement
assumeTrue(datanodeCount > NODE_PER_RACK + 1);
@@ -251,17 +254,18 @@ public void chooseNodeWithNoExcludedNodes(int datanodeCount)
@MethodSource("numDatanodes")
public void chooseNodeWithExcludedNodes(int datanodeCount)
throws SCMException {
+ StorageType storageType = StorageType.DEFAULT;
// test choose new datanodes for under replicated pipeline
// 3 replicas, two existing datanodes on same rack
assumeTrue(datanodeCount > NODE_PER_RACK);
- setup(datanodeCount);
+ setup(datanodeCount, storageType);
int nodeNum = 1;
List<DatanodeDetails> excludedNodes = new ArrayList<>();
excludedNodes.add(datanodes.get(0));
excludedNodes.add(datanodes.get(1));
List<DatanodeDetails> datanodeDetails = policy.chooseDatanodes(
- excludedNodes, null, nodeNum, 0, 15);
+ excludedNodes, null, nodeNum, 0, 15, storageType);
assertEquals(nodeNum, datanodeDetails.size());
assertFalse(cluster.isSameParent(datanodeDetails.get(0),
excludedNodes.get(0)));
@@ -273,7 +277,7 @@ public void chooseNodeWithExcludedNodes(int datanodeCount)
excludedNodes.clear();
excludedNodes.add(datanodes.get(0));
datanodeDetails = policy.chooseDatanodes(
- excludedNodes, null, nodeNum, 0, 15);
+ excludedNodes, null, nodeNum, 0, 15, storageType);
assertEquals(nodeNum, datanodeDetails.size());
assertTrue(cluster.isSameParent(
datanodeDetails.get(0), excludedNodes.get(0)) ||
@@ -285,7 +289,7 @@ public void chooseNodeWithExcludedNodes(int datanodeCount)
excludedNodes.add(datanodes.get(0));
excludedNodes.add(datanodes.get(5));
datanodeDetails = policy.chooseDatanodes(
- excludedNodes, null, nodeNum, 0, 15);
+ excludedNodes, null, nodeNum, 0, 15, storageType);
assertEquals(nodeNum, datanodeDetails.size());
assertTrue(cluster.isSameParent(
datanodeDetails.get(0), excludedNodes.get(0)) ||
@@ -295,14 +299,15 @@ public void chooseNodeWithExcludedNodes(int datanodeCount)
@ParameterizedTest
@ValueSource(ints = {NODE_PER_RACK + 1, 2 * NODE_PER_RACK + 1})
public void testSingleNodeRack(int datanodeCount) throws SCMException {
+ StorageType storageType = StorageType.DEFAULT;
// make sure there is a single node rack
assumeTrue(datanodeCount % NODE_PER_RACK == 1);
- setup(datanodeCount);
+ setup(datanodeCount, storageType);
List<DatanodeDetails> excludeNodes = new ArrayList<>();
excludeNodes.add(datanodes.get(datanodeCount - 1));
excludeNodes.add(datanodes.get(0));
List<DatanodeDetails> chooseDatanodes =
- policy.chooseDatanodes(excludeNodes, null, 1, 0, 0);
+ policy.chooseDatanodes(excludeNodes, null, 1, 0, 0, storageType);
assertEquals(1, chooseDatanodes.size());
// the selected node should be on the same rack as the second exclude node
assertTrue(
@@ -316,12 +321,13 @@ public void testFallback(int datanodeCount) throws SCMException {
// 5 replicas. there are only 3 racks. policy with fallback should
// allocate the 5th datanode though it will break the rack rule(first
// 2 replicas on same rack, others on different racks).
+ StorageType storageType = StorageType.DEFAULT;
assumeTrue(datanodeCount > NODE_PER_RACK * 2 &&
(datanodeCount % NODE_PER_RACK > 1));
- setup(datanodeCount);
+ setup(datanodeCount, storageType);
int nodeNum = 5;
List<DatanodeDetails> datanodeDetails =
- policy.chooseDatanodes(null, null, nodeNum, 0, 15);
+ policy.chooseDatanodes(null, null, nodeNum, 0, 15, storageType);
assertEquals(nodeNum, datanodeDetails.size());
assertTrue(cluster.isSameParent(datanodeDetails.get(0),
datanodeDetails.get(1)));
@@ -350,14 +356,15 @@ public void testFallback(int datanodeCount) throws SCMException {
@ParameterizedTest
@ValueSource(ints = {11, 12, 13, 14, 15})
public void testNoFallback(int datanodeCount) {
+ StorageType storageType = StorageType.DEFAULT;
assumeTrue(datanodeCount > (NODE_PER_RACK * 2) &&
(datanodeCount <= NODE_PER_RACK * 3));
- setup(datanodeCount);
+ setup(datanodeCount, storageType);
// 5 replicas. there are only 3 racks. policy prohibit fallback should fail.
int nodeNum = 5;
Exception e =
assertThrows(Exception.class,
- () -> policyNoFallback.chooseDatanodes(null, null, nodeNum, 0, 15),
+ () -> policyNoFallback.chooseDatanodes(null, null, nodeNum, 0, 15, storageType),
"Fallback prohibited, this call should fail");
assertEquals("SCMException", e.getClass().getSimpleName());
@@ -379,7 +386,8 @@ public void testNoFallback(int datanodeCount) {
@MethodSource("numDatanodes")
public void chooseNodeWithFavoredNodes(int datanodeCount)
throws SCMException {
- setup(datanodeCount);
+ StorageType storageType = StorageType.DEFAULT;
+ setup(datanodeCount, storageType);
int nodeNum = 1;
List<DatanodeDetails> excludedNodes = new ArrayList<>();
List<DatanodeDetails> favoredNodes = new ArrayList<>();
@@ -387,7 +395,7 @@ public void chooseNodeWithFavoredNodes(int datanodeCount)
// no excludedNodes, only favoredNodes
favoredNodes.add(datanodes.get(0));
List<DatanodeDetails> datanodeDetails = policy.chooseDatanodes(
- excludedNodes, favoredNodes, nodeNum, 0, 15);
+ excludedNodes, favoredNodes, nodeNum, 0, 15, storageType);
assertEquals(nodeNum, datanodeDetails.size());
assertEquals(datanodeDetails.get(0).getNetworkFullPath(),
favoredNodes.get(0).getNetworkFullPath());
@@ -399,7 +407,7 @@ public void chooseNodeWithFavoredNodes(int datanodeCount)
excludedNodes.add(datanodes.get(0));
favoredNodes.add(datanodes.get(2));
datanodeDetails = policy.chooseDatanodes(
- excludedNodes, favoredNodes, nodeNum, 0, 15);
+ excludedNodes, favoredNodes, nodeNum, 0, 15, storageType);
assertEquals(nodeNum, datanodeDetails.size());
assertEquals(datanodeDetails.get(0).getNetworkFullPath(),
favoredNodes.get(0).getNetworkFullPath());
@@ -411,7 +419,7 @@ public void chooseNodeWithFavoredNodes(int datanodeCount)
excludedNodes.add(datanodes.get(0));
favoredNodes.add(datanodes.get(0));
datanodeDetails = policy.chooseDatanodes(
- excludedNodes, favoredNodes, nodeNum, 0, 15);
+ excludedNodes, favoredNodes, nodeNum, 0, 15, storageType);
assertEquals(nodeNum, datanodeDetails.size());
assertNotEquals(datanodeDetails.get(0).getNetworkFullPath(),
favoredNodes.get(0).getNetworkFullPath());
@@ -420,13 +428,14 @@ public void chooseNodeWithFavoredNodes(int datanodeCount)
@ParameterizedTest
@MethodSource("numDatanodes")
public void testNoInfiniteLoop(int datanodeCount) {
- setup(datanodeCount);
+ StorageType storageType = StorageType.DEFAULT;
+ setup(datanodeCount, storageType);
int nodeNum = 1;
// request storage space larger than node capability
Exception e =
assertThrows(Exception.class,
- () -> policy.chooseDatanodes(null, null, nodeNum, STORAGE_CAPACITY + 0, 15),
+ () -> policy.chooseDatanodes(null, null, nodeNum, STORAGE_CAPACITY + 0, 15, storageType),
"Storage requested exceeds capacity, this call should fail");
assertEquals("SCMException", e.getClass().getSimpleName());
@@ -446,7 +455,8 @@ public void testNoInfiniteLoop(int datanodeCount) {
@MethodSource("numDatanodes")
public void testDatanodeWithDefaultNetworkLocation(int datanodeCount)
throws SCMException {
- setup(datanodeCount);
+ StorageType storageType = StorageType.DEFAULT;
+ setup(datanodeCount, storageType);
String hostname = "node";
List<DatanodeInfo> dnInfoList = new ArrayList<>();
List<DatanodeDetails> dataList = new ArrayList<>();
@@ -463,7 +473,7 @@ public void testDatanodeWithDefaultNetworkLocation(int datanodeCount)
StorageReportProto storage1 = HddsTestUtils.createStorageReport(
dnInfo.getID(), "/data1-" + dnInfo.getID(),
- STORAGE_CAPACITY, 0, 100L, null);
+ STORAGE_CAPACITY, 0, 100L, getStorageTypeProto(storageType));
MetadataStorageReportProto metaStorage1 =
HddsTestUtils.createMetadataStorageReport(
"/metadata1-" + dnInfo.getID(),
@@ -490,7 +500,7 @@ public void testDatanodeWithDefaultNetworkLocation(int datanodeCount)
new SCMContainerPlacementRackAware(nodeManager, conf, clusterMap, true,
metrics);
List<DatanodeDetails> datanodeDetails =
- newPolicy.chooseDatanodes(null, null, nodeNum, 0, 15);
+ newPolicy.chooseDatanodes(null, null, nodeNum, 0, 15, storageType);
assertEquals(nodeNum, datanodeDetails.size());
assertTrue(cluster.isSameParent(datanodeDetails.get(0),
datanodeDetails.get(1)));
@@ -504,7 +514,8 @@ public void testDatanodeWithDefaultNetworkLocation(int datanodeCount)
public void testvalidateContainerPlacement() {
// Only run this test for the full set of DNs. 5 DNs per rack on 3 racks.
final int datanodeCount = 15;
- setup(datanodeCount);
+ StorageType storageType = StorageType.DEFAULT;
+ setup(datanodeCount, storageType);
List<DatanodeDetails> dns = new ArrayList<>();
// First 5 node are on the same rack
dns.add(datanodes.get(0));
@@ -541,7 +552,8 @@ public void testvalidateContainerPlacement() {
@Test
public void testvalidateContainerPlacementSingleRackCluster() {
final int datanodeCount = 5;
- setup(datanodeCount);
+ StorageType storageType = StorageType.DEFAULT;
+ setup(datanodeCount, storageType);
// All nodes are on the same rack in this test, and the cluster only has
// one rack.
@@ -571,7 +583,8 @@ public void testvalidateContainerPlacementSingleRackCluster() {
@ParameterizedTest
@MethodSource("org.apache.hadoop.hdds.scm.node.NodeStatus#outOfServiceStates")
public void testOverReplicationAndOutOfServiceNodes(HddsProtos.NodeOperationalState state) {
- setup(7);
+ StorageType storageType = StorageType.DEFAULT;
+ setup(7, storageType);
// 7 datanodes, all nodes are used.
// /rack0/node0 -> IN_SERVICE
// /rack0/node1 -> IN_SERVICE
@@ -615,7 +628,8 @@ public void testOverReplicationAndOutOfServiceNodes(HddsProtos.NodeOperationalSt
@ParameterizedTest
@MethodSource("numDatanodes")
public void testOutOfServiceNodesNotSelected(int datanodeCount) {
- setup(datanodeCount);
+ StorageType storageType = StorageType.DEFAULT;
+ setup(datanodeCount, storageType);
// Set all the nodes to out of service
for (DatanodeInfo dn : dnInfos) {
dn.setNodeStatus(NodeStatus.valueOf(DECOMMISSIONED, HEALTHY));
@@ -627,7 +641,7 @@ public void testOutOfServiceNodesNotSelected(int datanodeCount) {
dnInfos.get(index).setNodeStatus(NodeStatus.inServiceHealthy());
try {
List<DatanodeDetails> datanodeDetails =
- policy.chooseDatanodes(null, null, 1, 0, 0);
+ policy.chooseDatanodes(null, null, 1, 0, 0, storageType);
assertEquals(dnInfos.get(index), datanodeDetails.get(0));
} catch (SCMException e) {
// If we get SCMException: No satisfied datanode to meet the ... this is
@@ -642,8 +656,9 @@ public void testOutOfServiceNodesNotSelected(int datanodeCount) {
@ValueSource(ints = {NODE_PER_RACK + 1, 2 * NODE_PER_RACK + 1})
public void chooseNodeWithUsedNodesMultipleRack(int datanodeCount)
throws SCMException {
+ StorageType storageType = StorageType.DEFAULT;
assumeTrue(datanodeCount > NODE_PER_RACK);
- setup(datanodeCount);
+ setup(datanodeCount, storageType);
int nodeNum = 1;
List<DatanodeDetails> excludedNodes = new ArrayList<>();
List<DatanodeDetails> usedNodes = new ArrayList<>();
@@ -653,7 +668,7 @@ public void chooseNodeWithUsedNodesMultipleRack(int datanodeCount)
usedNodes.add(datanodes.get(1));
List<DatanodeDetails> datanodeDetails = policy.chooseDatanodes(usedNodes,
- excludedNodes, null, nodeNum, 0, 5);
+ excludedNodes, null, nodeNum, 0, 5, storageType);
assertEquals(nodeNum, datanodeDetails.size());
// New DN should be on different rack than DN0 & DN1
@@ -669,7 +684,7 @@ public void chooseNodeWithUsedNodesMultipleRack(int datanodeCount)
usedNodes.add(datanodes.get(5));
datanodeDetails = policy.chooseDatanodes(usedNodes,
- excludedNodes, null, nodeNum, 0, 5);
+ excludedNodes, null, nodeNum, 0, 5, storageType);
assertEquals(nodeNum, datanodeDetails.size());
// New replica should be either on rack0 or rack1
@@ -681,8 +696,9 @@ public void chooseNodeWithUsedNodesMultipleRack(int datanodeCount)
@Test
public void chooseSingleNodeRackWithUsedAndExcludeNodes()
throws SCMException {
+ StorageType storageType = StorageType.DEFAULT;
int datanodeCount = 5;
- setup(datanodeCount);
+ setup(datanodeCount, storageType);
int nodeNum = 1;
List<DatanodeDetails> excludedNodes = new ArrayList<>();
List<DatanodeDetails> usedNodes = new ArrayList<>();
@@ -693,7 +709,7 @@ public void chooseSingleNodeRackWithUsedAndExcludeNodes()
excludedNodes.add(datanodes.get(2));
List<DatanodeDetails> datanodeDetails = policy.chooseDatanodes(usedNodes,
- excludedNodes, null, nodeNum, 0, 5);
+ excludedNodes, null, nodeNum, 0, 5, storageType);
assertEquals(nodeNum, datanodeDetails.size());
assertTrue(cluster.isSameParent(datanodes.get(0), datanodeDetails.get(0)));
@@ -708,7 +724,7 @@ public void chooseSingleNodeRackWithUsedAndExcludeNodes()
usedNodes.add(datanodes.get(0));
datanodeDetails = policy.chooseDatanodes(usedNodes,
- excludedNodes, null, nodeNum, 0, 5);
+ excludedNodes, null, nodeNum, 0, 5, storageType);
assertEquals(nodeNum, datanodeDetails.size());
assertNotEquals(excludedNodes.get(0), datanodeDetails.get(0));
@@ -719,7 +735,7 @@ public void chooseSingleNodeRackWithUsedAndExcludeNodes()
usedNodes.clear();
datanodeDetails = policy.chooseDatanodes(usedNodes,
- excludedNodes, null, nodeNum, 0, 5);
+ excludedNodes, null, nodeNum, 0, 5, storageType);
assertEquals(nodeNum, datanodeDetails.size());
assertNotEquals(excludedNodes.get(0), datanodeDetails.get(0));
@@ -730,8 +746,9 @@ public void chooseSingleNodeRackWithUsedAndExcludeNodes()
@MethodSource("numDatanodes")
public void chooseNodeWithUsedAndExcludeNodesMultipleRack(int datanodeCount)
throws SCMException {
+ StorageType storageType = StorageType.DEFAULT;
assumeTrue(datanodeCount > NODE_PER_RACK);
- setup(datanodeCount);
+ setup(datanodeCount, storageType);
int nodeNum = 2;
List<DatanodeDetails> excludedNodes = new ArrayList<>();
List<DatanodeDetails> usedNodes = new ArrayList<>();
@@ -742,7 +759,7 @@ public void chooseNodeWithUsedAndExcludeNodesMultipleRack(int datanodeCount)
excludedNodes.add(datanodes.get(1));
List<DatanodeDetails> datanodeDetails = policy.chooseDatanodes(usedNodes,
- excludedNodes, null, nodeNum, 0, 5);
+ excludedNodes, null, nodeNum, 0, 5, storageType);
assertEquals(nodeNum, datanodeDetails.size());
@@ -759,7 +776,7 @@ public void chooseNodeWithUsedAndExcludeNodesMultipleRack(int datanodeCount)
excludedNodes.add(datanodes.get(2));
datanodeDetails = policy.chooseDatanodes(usedNodes,
- excludedNodes, null, nodeNum, 0, 5);
+ excludedNodes, null, nodeNum, 0, 5, storageType);
assertEquals(nodeNum, datanodeDetails.size());
@@ -774,15 +791,16 @@ public void chooseNodeWithUsedAndExcludeNodesMultipleRack(int datanodeCount)
@MethodSource("numDatanodes")
public void chooseNodeWithOnlyExcludeAndNoUsedNodes(int datanodeCount)
throws SCMException {
+ StorageType storageType = StorageType.DEFAULT;
assumeTrue(datanodeCount > NODE_PER_RACK);
- setup(datanodeCount);
+ setup(datanodeCount, storageType);
int nodeNum = 3;
List<DatanodeDetails> excludedNodes = new ArrayList<>();
// 1 exclude node
excludedNodes.add(datanodes.get(1));
List<DatanodeDetails> datanodeDetails = policy.chooseDatanodes(null,
- excludedNodes, null, nodeNum, 0, 5);
+ excludedNodes, null, nodeNum, 0, 5, storageType);
assertEquals(nodeNum, datanodeDetails.size());
@@ -796,7 +814,7 @@ public void chooseNodeWithOnlyExcludeAndNoUsedNodes(int datanodeCount)
excludedNodes.add(datanodes.get(2));
datanodeDetails = policy.chooseDatanodes(null,
- excludedNodes, null, nodeNum, 0, 5);
+ excludedNodes, null, nodeNum, 0, 5, storageType);
assertEquals(nodeNum, datanodeDetails.size());
@@ -811,9 +829,10 @@ public void chooseNodeWithOnlyExcludeAndNoUsedNodes(int datanodeCount)
@ParameterizedTest
@ValueSource(ints = {11, 12, 13, 14, 15})
public void testNoFallbackWithUsedNodes(int datanodeCount) {
+ StorageType storageType = StorageType.DEFAULT;
assumeTrue(datanodeCount > (NODE_PER_RACK * 2) &&
(datanodeCount <= NODE_PER_RACK * 3));
- setup(datanodeCount);
+ setup(datanodeCount, storageType);
List<DatanodeDetails> usedNodes = new ArrayList<>();
usedNodes.add(datanodes.get(0));
@@ -822,7 +841,7 @@ public void testNoFallbackWithUsedNodes(int datanodeCount) {
int nodeNum = 5;
Exception e =
assertThrows(Exception.class,
- () -> policyNoFallback.chooseDatanodes(usedNodes, null, null, nodeNum, 0, 15),
+ () -> policyNoFallback.chooseDatanodes(usedNodes, null, null, nodeNum, 0, 15, storageType),
"Fallback prohibited, this call should fail");
assertEquals("SCMException", e.getClass().getSimpleName());
@@ -843,8 +862,9 @@ public void testNoFallbackWithUsedNodes(int datanodeCount) {
@Test
public void chooseNodeWithUsedAndFavouredNodesMultipleRack()
throws SCMException {
+ StorageType storageType = StorageType.DEFAULT;
int datanodeCount = 12;
- setup(datanodeCount);
+ setup(datanodeCount, storageType);
int nodeNum = 1;
List<DatanodeDetails> usedNodes = new ArrayList<>();
List<DatanodeDetails> favouredNodes = new ArrayList<>();
@@ -856,7 +876,7 @@ public void chooseNodeWithUsedAndFavouredNodesMultipleRack()
favouredNodes.add(datanodes.get(2));
List<DatanodeDetails> datanodeDetails = policy.chooseDatanodes(usedNodes,
- null, favouredNodes, nodeNum, 0, 5);
+ null, favouredNodes, nodeNum, 0, 5, storageType);
assertEquals(nodeNum, datanodeDetails.size());
// Favoured node should not be returned,
@@ -869,7 +889,7 @@ public void chooseNodeWithUsedAndFavouredNodesMultipleRack()
favouredNodes.add(datanodes.get(6));
datanodeDetails = policy.chooseDatanodes(usedNodes,
- null, favouredNodes, nodeNum, 0, 5);
+ null, favouredNodes, nodeNum, 0, 5, storageType);
assertEquals(nodeNum, datanodeDetails.size());
@@ -881,13 +901,13 @@ public void chooseNodeWithUsedAndFavouredNodesMultipleRack()
@Test
public void testSourceDatanodeIsNotChosenAsTarget() {
- setup(2);
+ setup(2, StorageType.DEFAULT);
List<DatanodeDetails> usedNodes = new ArrayList<>();
usedNodes.add(datanodes.get(0));
dnInfos.get(1).setNodeStatus(NodeStatus.inServiceHealthyReadOnly());
assertThrows(SCMException.class,
- () -> policy.chooseDatanodes(usedNodes, null, null, 1, 0, 0),
+ () -> policy.chooseDatanodes(usedNodes, null, null, 1, 0, 0, StorageType.DEFAULT),
"No target datanode, this call should fail");
}
}
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 eb3bb5670c3..ab0daadfee2 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
@@ -17,6 +17,7 @@
package org.apache.hadoop.hdds.scm.container.placement.algorithms;
+import static org.apache.hadoop.hdds.client.StorageTypeUtils.getStorageTypeProto;
import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState.DECOMMISSIONED;
import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState.HEALTHY;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_RATIS_VOLUME_FREE_SPACE_MIN;
@@ -48,6 +49,7 @@
import java.util.stream.IntStream;
import java.util.stream.Stream;
import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.conf.StorageUnit;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -108,7 +110,11 @@ private void updateStorageInDatanode(int dnIndex, long used, long remaining) {
}
private void setup(int datanodeCount) {
- setup(datanodeCount, NODE_PER_RACK);
+ setup(datanodeCount, NODE_PER_RACK, StorageType.DEFAULT);
+ }
+
+ private void setup(int datanodeCount, StorageType storageType) {
+ setup(datanodeCount, NODE_PER_RACK, storageType);
}
/**
@@ -142,7 +148,7 @@ private void setupOneDatanodePerRackWithExtraInLastRack(int rackCount,
}
createMocksAndUpdateStorageReports(
- rackCount - 1 + datanodesInLastRackCount);
+ rackCount - 1 + datanodesInLastRackCount, StorageType.DEFAULT);
}
private void setupConfiguration() {
@@ -158,6 +164,10 @@ private void setupConfiguration() {
}
private void setup(int datanodeCount, int nodesPerRack) {
+ setup(datanodeCount, nodesPerRack, StorageType.DEFAULT);
+ }
+
+ private void setup(int datanodeCount, int nodesPerRack, StorageType storageType) {
setupConfiguration();
// build datanodes, and network topology
String rack = "/rack";
@@ -170,7 +180,7 @@ private void setup(int datanodeCount, int nodesPerRack) {
setupDatanode(datanodeDetails);
}
- createMocksAndUpdateStorageReports(datanodeCount);
+ createMocksAndUpdateStorageReports(datanodeCount, storageType);
}
/**
@@ -200,44 +210,44 @@ private void setupDatanode(DatanodeDetails datanodeDetails) {
dnInfos.add(datanodeInfo);
}
- private void createMocksAndUpdateStorageReports(int datanodeCount) {
+ private void createMocksAndUpdateStorageReports(int datanodeCount, StorageType storageType) {
if (datanodeCount > 4) {
StorageReportProto storage2 = HddsTestUtils.createStorageReport(
dnInfos.get(2).getID(),
"/data1-" + datanodes.get(2).getID(),
- STORAGE_CAPACITY, 90L, 10L, null);
+ STORAGE_CAPACITY, 90L, 10L, getStorageTypeProto(storageType));
dnInfos.get(2).updateStorageReports(
new ArrayList<>(Arrays.asList(storage2)));
StorageReportProto storage3 = HddsTestUtils.createStorageReport(
dnInfos.get(3).getID(),
"/data1-" + dnInfos.get(3).getID(),
- STORAGE_CAPACITY, 80L, 20L, null);
+ STORAGE_CAPACITY, 80L, 20L, getStorageTypeProto(storageType));
dnInfos.get(3).updateStorageReports(
new ArrayList<>(Arrays.asList(storage3)));
StorageReportProto storage4 = HddsTestUtils.createStorageReport(
dnInfos.get(4).getID(),
"/data1-" + dnInfos.get(4).getID(),
- STORAGE_CAPACITY, 70L, 30L, null);
+ STORAGE_CAPACITY, 70L, 30L, getStorageTypeProto(storageType));
dnInfos.get(4).updateStorageReports(
new ArrayList<>(Arrays.asList(storage4)));
} else if (datanodeCount > 3) {
StorageReportProto storage2 = HddsTestUtils.createStorageReport(
dnInfos.get(2).getID(),
"/data1-" + dnInfos.get(2).getID(),
- STORAGE_CAPACITY, 90L, 10L, null);
+ STORAGE_CAPACITY, 90L, 10L, getStorageTypeProto(storageType));
dnInfos.get(2).updateStorageReports(
new ArrayList<>(Arrays.asList(storage2)));
StorageReportProto storage3 = HddsTestUtils.createStorageReport(
dnInfos.get(3).getID(),
"/data1-" + dnInfos.get(3).getID(),
- STORAGE_CAPACITY, 80L, 20L, null);
+ STORAGE_CAPACITY, 80L, 20L, getStorageTypeProto(storageType));
dnInfos.get(3).updateStorageReports(
new ArrayList<>(Arrays.asList(storage3)));
} else if (datanodeCount > 2) {
StorageReportProto storage2 = HddsTestUtils.createStorageReport(
dnInfos.get(2).getID(),
"/data1-" + dnInfos.get(2).getID(),
- STORAGE_CAPACITY, 84L, 16L, null);
+ STORAGE_CAPACITY, 84L, 16L, getStorageTypeProto(storageType));
dnInfos.get(2).updateStorageReports(
new ArrayList<>(Arrays.asList(storage2)));
}
@@ -300,12 +310,12 @@ public void chooseNodeWithNoExcludedNodes(int datanodeCount)
// 1 replica
int nodeNum = 1;
List<DatanodeDetails> datanodeDetails =
- policy.chooseDatanodes(null, null, nodeNum, 0, 15);
+ policy.chooseDatanodes(null, null, nodeNum, 0, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
// 2 replicas
nodeNum = 2;
- datanodeDetails = policy.chooseDatanodes(null, null, nodeNum, 0, 15);
+ datanodeDetails = policy.chooseDatanodes(null, null, nodeNum, 0, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertTrue(!cluster.isSameParent(datanodeDetails.get(0),
datanodeDetails.get(1)) || (datanodeCount <= NODE_PER_RACK));
@@ -313,7 +323,7 @@ public void chooseNodeWithNoExcludedNodes(int datanodeCount)
// 3 replicas
nodeNum = 3;
if (datanodeCount > nodeNum) {
- datanodeDetails = policy.chooseDatanodes(null, null, nodeNum, 0, 15);
+ datanodeDetails = policy.chooseDatanodes(null, null, nodeNum, 0, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertEquals(getRackSize(datanodeDetails),
Math.min(nodeNum, rackNum));
@@ -326,10 +336,10 @@ public void chooseNodeWithNoExcludedNodes(int datanodeCount)
if (datanodeCount == 6) {
int finalNodeNum = nodeNum;
SCMException e = assertThrows(SCMException.class,
- () -> policy.chooseDatanodes(null, null, finalNodeNum, 0, 15));
+ () -> policy.chooseDatanodes(null, null, finalNodeNum, 0, 15, StorageType.DEFAULT));
assertEquals(FAILED_TO_FIND_HEALTHY_NODES, e.getResult());
} else {
- datanodeDetails = policy.chooseDatanodes(null, null, nodeNum, 0, 15);
+ datanodeDetails = policy.chooseDatanodes(null, null, nodeNum, 0, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertEquals(getRackSize(datanodeDetails), Math.min(nodeNum, rackNum));
}
@@ -342,16 +352,37 @@ public void chooseNodeWithNoExcludedNodes(int datanodeCount)
if (datanodeCount == 11) {
int finalNodeNum = nodeNum;
SCMException e = assertThrows(SCMException.class,
- () -> policy.chooseDatanodes(null, null, finalNodeNum, 0, 15));
+ () -> policy.chooseDatanodes(null, null, finalNodeNum, 0, 15, StorageType.DEFAULT));
assertEquals(FAILED_TO_FIND_HEALTHY_NODES, e.getResult());
} else {
- datanodeDetails = policy.chooseDatanodes(null, null, nodeNum, 0, 15);
+ datanodeDetails = policy.chooseDatanodes(null, null, nodeNum, 0, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertEquals(getRackSize(datanodeDetails), Math.min(nodeNum, rackNum));
}
}
}
+ @Test
+ public void chooseNodeWithStorageType() throws SCMException {
+ StorageType[] storageTypes = StorageType.values();
+ int datanodeCount = 3;
+ for (StorageType storageType : storageTypes) {
+ setup(datanodeCount, storageType);
+ int rackLevel = cluster.getMaxLevel() - 1;
+ int rackNum = cluster.getNumOfNodes(rackLevel);
+ StorageType anotherStorageType = storageTypes[((storageTypes.length + 1) % storageTypes.length)];
+
+ // 1 replica
+ final int nodeNum = 1;
+ List<DatanodeDetails> datanodeDetails =
+ policy.chooseDatanodes(null, null, nodeNum, 0, 15, StorageType.DEFAULT);
+ assertEquals(nodeNum, datanodeDetails.size());
+ // We only have storageType Datanode, so we cannot choose another storageType Datanode
+ assertThrows(SCMException.class,
+ () -> policy.chooseDatanodes(null, null, nodeNum, 0, 15, anotherStorageType));
+ }
+ }
+
@ParameterizedTest
@MethodSource("numDatanodes")
public void chooseNodeWithExcludedNodes(int datanodeCount)
@@ -369,7 +400,7 @@ public void chooseNodeWithExcludedNodes(int datanodeCount)
excludedNodes.add(datanodes.get(0));
excludedNodes.add(datanodes.get(1));
List<DatanodeDetails> datanodeDetails = policy.chooseDatanodes(
- excludedNodes, null, nodeNum, 0, 15);
+ excludedNodes, null, nodeNum, 0, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertFalse(cluster.isSameParent(datanodeDetails.get(0),
excludedNodes.get(0)));
@@ -382,7 +413,7 @@ public void chooseNodeWithExcludedNodes(int datanodeCount)
excludedNodes.clear();
excludedNodes.add(datanodes.get(0));
datanodeDetails = policy.chooseDatanodes(
- excludedNodes, null, nodeNum, 0, 15);
+ excludedNodes, null, nodeNum, 0, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertEquals(getRackSize(datanodeDetails, excludedNodes),
Math.min(totalNum, rackNum));
@@ -394,7 +425,7 @@ public void chooseNodeWithExcludedNodes(int datanodeCount)
excludedNodes.add(datanodes.get(0));
excludedNodes.add(datanodes.get(5));
datanodeDetails = policy.chooseDatanodes(
- excludedNodes, null, nodeNum, 0, 15);
+ excludedNodes, null, nodeNum, 0, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertEquals(getRackSize(datanodeDetails, excludedNodes),
Math.min(totalNum, rackNum));
@@ -408,11 +439,11 @@ public void chooseNodeWithExcludedNodes(int datanodeCount)
int finalNodeNum = nodeNum;
SCMException e = assertThrows(SCMException.class,
() -> policy.chooseDatanodes(excludedNodes, null,
- finalNodeNum, 0, 15));
+ finalNodeNum, 0, 15, StorageType.DEFAULT));
assertEquals(FAILED_TO_FIND_HEALTHY_NODES, e.getResult());
} else {
datanodeDetails = policy.chooseDatanodes(
- excludedNodes, null, nodeNum, 0, 15);
+ excludedNodes, null, nodeNum, 0, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertEquals(getRackSize(datanodeDetails, excludedNodes),
Math.min(totalNum, rackNum));
@@ -426,7 +457,7 @@ public void chooseNodeWithExcludedNodes(int datanodeCount)
excludedNodes.add(datanodes.get(0));
excludedNodes.add(datanodes.get(5));
datanodeDetails = policy.chooseDatanodes(
- excludedNodes, null, nodeNum, 0, 15);
+ excludedNodes, null, nodeNum, 0, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertEquals(getRackSize(datanodeDetails, excludedNodes),
Math.min(totalNum, rackNum));
@@ -444,7 +475,7 @@ public void chooseNodeWithFavoredNodes(int datanodeCount)
// no excludedNodes, only favoredNodes
favoredNodes.add(datanodes.get(0));
List<DatanodeDetails> datanodeDetails = policy.chooseDatanodes(
- excludedNodes, favoredNodes, nodeNum, 0, 15);
+ excludedNodes, favoredNodes, nodeNum, 0, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertEquals(datanodeDetails.get(0).getNetworkFullPath(),
favoredNodes.get(0).getNetworkFullPath());
@@ -456,7 +487,7 @@ public void chooseNodeWithFavoredNodes(int datanodeCount)
excludedNodes.add(datanodes.get(0));
favoredNodes.add(datanodes.get(1));
datanodeDetails = policy.chooseDatanodes(
- excludedNodes, favoredNodes, nodeNum, 0, 15);
+ excludedNodes, favoredNodes, nodeNum, 0, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertEquals(datanodeDetails.get(0).getNetworkFullPath(),
favoredNodes.get(0).getNetworkFullPath());
@@ -468,7 +499,7 @@ public void chooseNodeWithFavoredNodes(int datanodeCount)
excludedNodes.add(datanodes.get(0));
favoredNodes.add(datanodes.get(0));
datanodeDetails = policy.chooseDatanodes(
- excludedNodes, favoredNodes, nodeNum, 0, 15);
+ excludedNodes, favoredNodes, nodeNum, 0, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertNotEquals(datanodeDetails.get(0).getNetworkFullPath(),
favoredNodes.get(0).getNetworkFullPath());
@@ -482,7 +513,7 @@ public void testNoInfiniteLoop(int datanodeCount) {
// request storage space larger than node capability
Exception e =
assertThrows(Exception.class,
- () -> policy.chooseDatanodes(null, null, nodeNum, STORAGE_CAPACITY + 0, 15),
+ () -> policy.chooseDatanodes(null, null, nodeNum, STORAGE_CAPACITY + 0, 15, StorageType.DEFAULT),
"Storage requested exceeds capacity, this call should fail");
assertEquals("SCMException", e.getClass().getSimpleName());
@@ -546,7 +577,7 @@ public void testDatanodeWithDefaultNetworkLocation(int datanodeCount)
new SCMContainerPlacementRackScatter(nodeManager, conf, clusterMap,
true, metrics);
List<DatanodeDetails> datanodeDetails =
- newPolicy.chooseDatanodes(null, null, nodeNum, 0, 15);
+ newPolicy.chooseDatanodes(null, null, nodeNum, 0, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertEquals(1, getRackSize(datanodeDetails));
}
@@ -681,7 +712,7 @@ public void testPipelineProviderRackScatter() throws SCMException {
List<DatanodeDetails> usedDns = new ArrayList<>();
List<DatanodeDetails> excludedDns = new ArrayList<>();
List<DatanodeDetails> additionalNodes = policy.chooseDatanodes(usedDns,
- excludedDns, null, 3, 0, 5);
+ excludedDns, null, 3, 0, 5, StorageType.DEFAULT);
assertPlacementPolicySatisfied(usedDns, additionalNodes, excludedDns, 3,
true, 0);
}
@@ -699,13 +730,13 @@ public void testPipelineProviderRackScatterFallback() throws SCMException {
List<DatanodeDetails> usedDns = new ArrayList<>();
List<DatanodeDetails> excludedDns = new ArrayList<>();
List<DatanodeDetails> additionalNodes = policy.chooseDatanodes(usedDns,
- excludedDns, null, 3, 0, 5);
+ excludedDns, null, 3, 0, 5, StorageType.DEFAULT);
assertPlacementPolicySatisfied(usedDns, additionalNodes, excludedDns, 3,
true, 0);
setup(3, 3);
additionalNodes = policy.chooseDatanodes(usedDns,
- excludedDns, null, 3, 0, 5);
+ excludedDns, null, 3, 0, 5, StorageType.DEFAULT);
assertPlacementPolicySatisfied(usedDns, additionalNodes, excludedDns, 3,
true, 0);
}
@@ -718,7 +749,7 @@ public void testValidChooseNodesWithUsedNodes() throws SCMException {
List<DatanodeDetails> usedDns = getDatanodes(Lists.newArrayList(0, 1));
List<DatanodeDetails> excludedDns = getDatanodes(Lists.newArrayList(2));
List<DatanodeDetails> additionalNodes = policy.chooseDatanodes(usedDns,
- excludedDns, null, 2, 0, 5);
+ excludedDns, null, 2, 0, 5, StorageType.DEFAULT);
assertPlacementPolicySatisfied(usedDns, additionalNodes, excludedDns, 4,
true, 0);
}
@@ -737,7 +768,7 @@ public void shouldChooseNodeIfNodesRequiredLessThanAdditionalRacksRequired()
List<DatanodeDetails> chosenNodes =
policy.chooseDatanodes(usedDns, excludedDns,
- null, 1, 0, 5);
+ null, 1, 0, 5, StorageType.DEFAULT);
assertEquals(1, chosenNodes.size());
/*
The chosen node should be node4 from the third rack because we prefer to
@@ -771,7 +802,7 @@ public void shouldChooseNodeWhenOneNodeRequiredAndTwoRacksRequired()
List<DatanodeDetails> chosenNode =
policy.chooseDatanodes(usedDns, excludedDns,
- null, 1, 0, 5);
+ null, 1, 0, 5, StorageType.DEFAULT);
assertEquals(1, chosenNode.size());
assertTrue(chosenNode.get(0).equals(datanodes.get(3)) ||
chosenNode.get(0).equals(datanodes.get(4)));
@@ -784,7 +815,7 @@ public void testChooseNodesWithInsufficientNodesAvailable() {
List<DatanodeDetails> excludedDns = getDatanodes(Lists.newArrayList(2));
SCMException exception = assertThrows(SCMException.class, () ->
policy.chooseDatanodes(usedDns, excludedDns,
- null, 3, 0, 5));
+ null, 3, 0, 5, StorageType.DEFAULT));
assertThat(exception.getMessage())
.matches("^No enough datanodes to choose.*");
assertEquals(SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE,
@@ -806,7 +837,7 @@ public void chooseNodesOnTheSameRackWhenInSufficientRacks()
List<DatanodeDetails> excludedDns = getDatanodes(Lists.newArrayList(5));
List<DatanodeDetails> chosenDatanodes =
- policy.chooseDatanodes(usedDns, excludedDns, null, 2, 0, 5);
+ policy.chooseDatanodes(usedDns, excludedDns, null, 2, 0, 5, StorageType.DEFAULT);
assertEquals(2, chosenDatanodes.size());
for (DatanodeDetails dn : chosenDatanodes) {
@@ -865,7 +896,7 @@ public void testExcludedNodesOverlapsOutOfServiceNodes() throws SCMException {
excludedNodes.add(datanodes.get(5));
List<DatanodeDetails> datanodeDetails = policy.chooseDatanodes(
- excludedNodes, null, nodeNum, 0, 5);
+ excludedNodes, null, nodeNum, 0, 5, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
}
@@ -893,7 +924,7 @@ public void testAllNodesOnRackExcludedReducesRackCount()
List<DatanodeDetails> chosenNodes =
policy.chooseDatanodes(usedDns, excludedDns,
- null, 1, 0, 5);
+ null, 1, 0, 5, StorageType.DEFAULT);
assertEquals(1, chosenNodes.size());
}
@@ -914,7 +945,7 @@ public void testAllNodesOnRackExcludedReducesRackCount2()
List<DatanodeDetails> chosenNodes =
policy.chooseDatanodes(usedDns, excludedDns,
- null, 1, 0, 5);
+ null, 1, 0, 5, StorageType.DEFAULT);
assertEquals(1, chosenNodes.size());
}
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 b8c1a0f4c4f..320490bf15f 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
@@ -29,6 +29,7 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.conf.StorageUnit;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -106,7 +107,7 @@ public void chooseDatanodes() throws SCMException {
for (int i = 0; i < 100; i++) {
//when
List<DatanodeDetails> datanodeDetails = scmContainerPlacementRandom
- .chooseDatanodes(existingNodes, null, 1, 15, 15);
+ .chooseDatanodes(existingNodes, null, 1, 15, 15, StorageType.DEFAULT);
//then
assertEquals(1, datanodeDetails.size());
@@ -224,11 +225,11 @@ public void testIsValidNode() throws SCMException {
mock(SCMContainerPlacementMetrics.class));
assertTrue(
- scmContainerPlacementRandom.isValidNode(datanodes.get(0), 15L, 15L));
+ scmContainerPlacementRandom.isValidNode(datanodes.get(0), 15L, 15L, StorageType.DEFAULT));
assertFalse(
- scmContainerPlacementRandom.isValidNode(datanodes.get(1), 15L, 15L));
+ scmContainerPlacementRandom.isValidNode(datanodes.get(1), 15L, 15L, StorageType.DEFAULT));
assertFalse(
- scmContainerPlacementRandom.isValidNode(datanodes.get(2), 15L, 15L));
+ scmContainerPlacementRandom.isValidNode(datanodes.get(2), 15L, 15L, StorageType.DEFAULT));
}
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandlerTests.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandlerTests.java
index 4d3c42d84f4..d0741e3caac 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandlerTests.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandlerTests.java
@@ -45,6 +45,7 @@
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.ReplicationConfig;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -204,7 +205,7 @@ protected void testMisReplication(Set<ContainerReplica> availableReplicas,
if (expectedNumberOfNodes > 0) {
when(mockedPlacementPolicy.chooseDatanodes(
any(), any(), any(),
- eq(copy.size()), anyLong(), anyLong()))
+ eq(copy.size()), anyLong(), anyLong(), any(StorageType.class)))
.thenAnswer(invocation -> {
List<DatanodeDetails> datanodeDetails =
invocation.getArgument(0);
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationTestUtil.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationTestUtil.java
index 63f4c6c9f33..7642e797a95 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationTestUtil.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationTestUtil.java
@@ -34,6 +34,7 @@
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.ReplicationConfig;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.conf.StorageUnit;
@@ -319,7 +320,7 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
List<DatanodeDetails> usedNodes,
List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> favoredNodes, int nodesRequiredToChoose,
- long metadataSizeRequired, long dataSizeRequired) {
+ long metadataSizeRequired, long dataSizeRequired, StorageType storageType) {
List<DatanodeDetails> dns = new ArrayList<>();
for (int i = 0; i < nodesRequiredToChoose; i++) {
dns.add(MockDatanodeDetails.randomDatanodeDetails());
@@ -355,7 +356,7 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
List<DatanodeDetails> usedNodes,
List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> favoredNodes, int nodesRequiredToChoose,
- long metadataSizeRequired, long dataSizeRequired)
+ long metadataSizeRequired, long dataSizeRequired, StorageType storageType)
throws SCMException {
long containerSize = (long) conf.getStorageSize(ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE,
ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE_DEFAULT, StorageUnit.BYTES);
@@ -386,7 +387,7 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
List<DatanodeDetails> usedNodes,
List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> favoredNodes, int nodesRequiredToChoose,
- long metadataSizeRequired, long dataSizeRequired)
+ long metadataSizeRequired, long dataSizeRequired, StorageType storageType)
throws SCMException {
long containerSize = (long) conf.getStorageSize(ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE,
ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE_DEFAULT, StorageUnit.BYTES);
@@ -416,7 +417,7 @@ protected List<DatanodeDetails> chooseDatanodesInternal(
List<DatanodeDetails> usedNodes,
List<DatanodeDetails> excludedNodes,
List<DatanodeDetails> favoredNodes, int nodesRequiredToChoose,
- long metadataSizeRequired, long dataSizeRequired)
+ long metadataSizeRequired, long dataSizeRequired, StorageType storageType)
throws SCMException {
long containerSize = (long) conf.getStorageSize(ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE,
ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE_DEFAULT, StorageUnit.BYTES);
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestECMisReplicationHandler.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestECMisReplicationHandler.java
index 5bcba983a2c..3b298ad9a29 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestECMisReplicationHandler.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestECMisReplicationHandler.java
@@ -37,6 +37,7 @@
import java.util.Map;
import java.util.Set;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -93,8 +94,8 @@ public void testMisReplicationWithNoNodesReturned() throws IOException {
when(placementPolicy.validateContainerPlacement(anyList(),
anyInt())).thenReturn(mockedContainerPlacementStatus);
when(placementPolicy.chooseDatanodes(
- any(), any(), any(), anyInt(), anyLong(), anyLong()))
- .thenThrow(new IOException("No nodes found"));
+ any(), any(), any(), anyInt(), anyLong(), anyLong(), any(StorageType.class)))
+ .thenThrow(new SCMException("No nodes found", SCMException.ResultCodes.NO_SUCH_DATANODE));
assertThrows(SCMException.class, () -> testMisReplication(
availableReplicas, placementPolicy, Collections.emptyList(),
0, 2, 0));
@@ -208,7 +209,7 @@ public void commandsForFewerThanRequiredNodes() throws IOException {
PlacementPolicy placementPolicy = mock(PlacementPolicy.class);
List<DatanodeDetails> targetDatanodes = singletonList(
availableReplicas.iterator().next().getDatanodeDetails());
- when(placementPolicy.chooseDatanodes(any(), any(), any(), anyInt(), anyLong(), anyLong()))
+ when(placementPolicy.chooseDatanodes(any(), any(), any(), anyInt(), anyLong(), anyLong(), any(StorageType.class)))
.thenReturn(targetDatanodes);
assertThrows(InsufficientDatanodesException.class,
() -> testMisReplication(availableReplicas, Collections.emptyList(),
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestECUnderReplicationHandler.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestECUnderReplicationHandler.java
index 5d2af561196..10ad9245d4f 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestECUnderReplicationHandler.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestECUnderReplicationHandler.java
@@ -69,6 +69,7 @@
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.IntStream;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -1036,9 +1037,8 @@ public void testMaintenanceDoesNotRequestZeroNodes() throws IOException {
.createReplicas(Pair.of(DECOMMISSIONING, 1), Pair.of(IN_SERVICE, 2),
Pair.of(IN_MAINTENANCE, 3), Pair.of(IN_SERVICE, 4),
Pair.of(IN_SERVICE, 5));
-
when(ecPlacementPolicy.chooseDatanodes(anyList(), anyList(),
- isNull(), anyInt(), anyLong(), anyLong()))
+ isNull(), anyInt(), anyLong(), anyLong(), any(StorageType.class)))
.thenAnswer(invocationOnMock -> {
int numNodes = invocationOnMock.getArgument(3);
List<DatanodeDetails> targets = new ArrayList<>();
@@ -1059,7 +1059,7 @@ public void testMaintenanceDoesNotRequestZeroNodes() throws IOException {
assertEquals(1, commandsSent.size());
verify(ecPlacementPolicy, times(0))
.chooseDatanodes(anyList(), isNull(), eq(0), anyLong(),
- anyLong());
+ anyLong(), any(StorageType.class));
}
/**
@@ -1082,7 +1082,7 @@ public void testDatanodesPendingAddAreNotSelectedAsTargets()
contain the DN pending ADD.
*/
when(ecPlacementPolicy.chooseDatanodes(anyList(), anyList(),
- isNull(), anyInt(), anyLong(), anyLong()))
+ isNull(), anyInt(), anyLong(), anyLong(), any(StorageType.class)))
.thenAnswer(invocationOnMock -> {
List<DatanodeDetails> usedList = invocationOnMock.getArgument(0);
List<DatanodeDetails> excludeList = invocationOnMock.getArgument(1);
@@ -1203,7 +1203,7 @@ private DeleteContainerCommand createDeleteContainerCommand(
/**
* Helper to mock and verify calls to
- * {@link PlacementPolicy#chooseDatanodes(List, List, int, long, long)}.
+ * {@link PlacementPolicy#chooseDatanodes(List, List, int, long, long, StorageType)}.
*/
private static class PlacementPolicySpy {
@@ -1215,7 +1215,7 @@ private static class PlacementPolicySpy {
PlacementPolicySpy(PlacementPolicy placementPolicy, int totalNodes)
throws IOException {
when(placementPolicy.chooseDatanodes(any(), any(),
- any(), anyInt(), anyLong(), anyLong())
+ any(), anyInt(), anyLong(), anyLong(), any(StorageType.class))
).thenAnswer(invocation -> {
final Collection<DatanodeDetails> used = invocation.getArgument(0);
final Collection<DatanodeDetails> excluded = invocation.getArgument(1);
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestRatisMisReplicationHandler.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestRatisMisReplicationHandler.java
index e5b25f8be4d..48dec2e5c6e 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestRatisMisReplicationHandler.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestRatisMisReplicationHandler.java
@@ -36,6 +36,7 @@
import java.util.Map;
import java.util.Set;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.RatisReplicationConfig;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -101,8 +102,8 @@ public void testMisReplicationWithNoNodesReturned() throws IOException {
anyInt())).thenReturn(mockedContainerPlacementStatus);
when(placementPolicy.chooseDatanodes(
any(), any(), any(),
- anyInt(), anyLong(), anyLong()))
- .thenThrow(new IOException("No nodes found"));
+ anyInt(), anyLong(), anyLong(), any(StorageType.class)))
+ .thenThrow(new SCMException("No nodes found", SCMException.ResultCodes.NO_SUCH_DATANODE));
assertThrows(SCMException.class, () -> testMisReplication(
availableReplicas, placementPolicy, Collections.emptyList(),
0, 2, 0));
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestRatisUnderReplicationHandler.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestRatisUnderReplicationHandler.java
index b368e56150e..f7c94a5210d 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestRatisUnderReplicationHandler.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestRatisUnderReplicationHandler.java
@@ -50,6 +50,7 @@
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.RatisReplicationConfig;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -467,7 +468,7 @@ public void testOnlyHighestBcsidShouldBeASource() throws IOException {
public void testCorrectUsedAndExcludedNodesPassed() throws IOException {
PlacementPolicy mockPolicy = mock(PlacementPolicy.class);
when(mockPolicy.chooseDatanodes(any(), any(), any(),
- anyInt(), anyLong(), anyLong()))
+ anyInt(), anyLong(), anyLong(), any(StorageType.class)))
.thenReturn(Collections.singletonList(
MockDatanodeDetails.randomDatanodeDetails()));
@@ -510,10 +511,9 @@ public void testCorrectUsedAndExcludedNodesPassed() throws IOException {
handler.processAndSendCommands(replicas, pendingOps,
getUnderReplicatedHealthResult(), 2);
-
verify(mockPolicy, times(1)).chooseDatanodes(
usedNodesCaptor.capture(), excludedNodesCaptor.capture(), any(),
- anyInt(), anyLong(), anyLong());
+ anyInt(), anyLong(), anyLong(), any(StorageType.class));
List<DatanodeDetails> usedNodes = usedNodesCaptor.getValue();
List<DatanodeDetails> excludedNodes = excludedNodesCaptor.getValue();
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java
index bcb3dea9768..70c203357f3 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java
@@ -1587,7 +1587,7 @@ public void testPendingOpExpiry() throws ContainerNotFoundException {
@ParameterizedTest
@ValueSource(booleans = {true, false})
- public void testNotifyNodeStateChangeWakesUpThread(boolean queueIsEmpty)
+ public void testNotifyNodeStateChangeWakesUpThread(boolean queueIsEmpty)
throws IOException, InterruptedException, ReflectiveOperationException, TimeoutException {
AtomicBoolean processAllCalled = new AtomicBoolean(false);
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java
index 933465c0245..4ad8ee4a302 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java
@@ -21,6 +21,7 @@
import java.util.List;
import org.apache.hadoop.hdds.client.RatisReplicationConfig;
import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.client.StorageTier;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.scm.ha.SCMContext;
@@ -64,12 +65,12 @@ protected void initializePipeline(Pipeline pipeline) throws IOException {
}
@Override
- public Pipeline create(RatisReplicationConfig replicationConfig)
+ public Pipeline create(RatisReplicationConfig replicationConfig, StorageTier storageTier)
throws IOException {
if (autoOpenPipeline) {
- return super.create(replicationConfig);
+ return super.create(replicationConfig, storageTier);
} else {
- Pipeline initialPipeline = super.create(replicationConfig);
+ Pipeline initialPipeline = super.create(replicationConfig, storageTier);
Pipeline pipeline = Pipeline.newBuilder()
.setId(initialPipeline.getId())
// overwrite pipeline state to main ALLOCATED
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestECPipelineProvider.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestECPipelineProvider.java
index f2511e624f7..92a28c99d46 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestECPipelineProvider.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestECPipelineProvider.java
@@ -41,7 +41,9 @@
import java.util.Iterator;
import java.util.List;
import java.util.Set;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
+import org.apache.hadoop.hdds.client.StorageTier;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.conf.StorageUnit;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -82,7 +84,7 @@ public void setup() throws IOException, NodeNotFoundException {
// Placement policy will always return EC number of random nodes.
when(placementPolicy.chooseDatanodes(anyList(),
anyList(), anyInt(), anyLong(),
- anyLong()))
+ anyLong(), any(StorageType.class)))
.thenAnswer(invocation -> {
List<DatanodeDetails> dns = new ArrayList<>();
for (int i = 0; i < (int) invocation.getArguments()[2]; i++) {
@@ -98,7 +100,7 @@ public void setup() throws IOException, NodeNotFoundException {
@Test
public void testSimplePipelineCanBeCreatedWithIndexes() throws IOException {
ECReplicationConfig ecConf = new ECReplicationConfig(3, 2);
- Pipeline pipeline = provider.create(ecConf);
+ Pipeline pipeline = provider.create(ecConf, StorageTier.getDefaultTier());
assertEquals(EC, pipeline.getType());
assertEquals(ecConf.getData() + ecConf.getParity(), pipeline.getNodes().size());
assertEquals(ALLOCATED, pipeline.getPipelineState());
@@ -195,12 +197,11 @@ public void testExcludedAndFavoredNodesPassedToPlacementPolicy()
List<DatanodeDetails> favoredNodes = new ArrayList<>();
favoredNodes.add(MockDatanodeDetails.randomDatanodeDetails());
- Pipeline pipeline = provider.create(ecConf, excludedNodes, favoredNodes);
+ Pipeline pipeline = provider.create(ecConf, excludedNodes, favoredNodes, StorageTier.getDefaultTier());
assertEquals(EC, pipeline.getType());
assertEquals(ecConf.getData() + ecConf.getParity(), pipeline.getNodes().size());
-
verify(placementPolicy).chooseDatanodes(excludedNodes, favoredNodes,
- ecConf.getRequiredNodes(), 0, containerSizeBytes);
+ ecConf.getRequiredNodes(), 0, containerSizeBytes, StorageType.DEFAULT);
}
private Set<ContainerReplica> createContainerReplicas(int number) {
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineDatanodesIntersection.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineDatanodesIntersection.java
index a947d65e0fe..6f95ba48ad4 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineDatanodesIntersection.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineDatanodesIntersection.java
@@ -25,6 +25,7 @@
import java.io.IOException;
import java.util.List;
import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.client.StorageTier;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
@@ -108,7 +109,7 @@ public void testPipelineDatanodesIntersection(int nodeCount,
while (!end && createdPipelineCount <= healthyNodeCount * nodeHeaviness) {
try {
Pipeline pipeline = provider.create(RatisReplicationConfig.getInstance(
- ReplicationFactor.THREE));
+ ReplicationFactor.THREE), StorageTier.getDefaultTier());
HddsProtos.Pipeline pipelineProto = pipeline.getProtobufMessage(
ClientVersion.CURRENT_VERSION);
stateManager.addPipeline(pipelineProto);
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 3ddb0361e08..6ddb2e8b6af 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
@@ -36,6 +36,7 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.conf.StorageUnit;
@@ -174,7 +175,7 @@ public void testDefaultPipelineProviderRackPlacement() throws Exception {
int nodeNum = 3;
List<DatanodeDetails> datanodeDetails =
- policy.chooseDatanodes(null, null, nodeNum, 15, 15);
+ policy.chooseDatanodes(null, null, nodeNum, 15, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertTrue(cluster.isSameParent(datanodeDetails.get(0),
datanodeDetails.get(2)));
@@ -200,7 +201,7 @@ public void testRackScatterPipelineProviderRackPlacement() throws Exception {
List<DatanodeDetails> favoredNodes = new ArrayList<>();
List<DatanodeDetails> datanodeDetails =
policy.chooseDatanodes(excludedNodes, excludedNodes, favoredNodes,
- nodeNum, 15, 15);
+ nodeNum, 15, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertFalse(cluster.isSameParent(datanodeDetails.get(0),
datanodeDetails.get(2)));
@@ -224,7 +225,7 @@ public void testPipelineProviderRackPlacementAnchorChange()
int nodeNum = 3;
List<DatanodeDetails> datanodeDetails =
- policy.chooseDatanodes(null, null, nodeNum, 15, 15);
+ policy.chooseDatanodes(null, null, nodeNum, 15, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
// First anchor will be Node0, Since there is no more node available
@@ -252,7 +253,7 @@ public void testPipelineProviderRackPlacementWithUsedNodes()
usedNodes.add(datanodes.get(0));
int nodeNum = 2;
List<DatanodeDetails> datanodeDetails =
- policy.chooseDatanodes(usedNodes, null, null, nodeNum, 15, 15);
+ policy.chooseDatanodes(usedNodes, null, null, nodeNum, 15, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
assertTrue(cluster.isSameParent(usedNodes.get(0),
@@ -266,7 +267,7 @@ public void testPipelineProviderRackPlacementWithUsedNodes()
nodeNum = 1;
datanodeDetails =
- policy.chooseDatanodes(usedNodes, null, null, nodeNum, 15, 15);
+ policy.chooseDatanodes(usedNodes, null, null, nodeNum, 15, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
// Node return by policy should have different parent as node0 and node1
assertFalse(cluster.isSameParent(usedNodes.get(0),
@@ -280,7 +281,7 @@ public void testPipelineProviderRackPlacementWithUsedNodes()
nodeNum = 1;
datanodeDetails =
- policy.chooseDatanodes(usedNodes, null, null, nodeNum, 15, 15);
+ policy.chooseDatanodes(usedNodes, null, null, nodeNum, 15, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
// Node return by policy should have same parent as node0 or node3
assertTrue(cluster.isSameParent(usedNodes.get(0),
@@ -291,7 +292,7 @@ public void testPipelineProviderRackPlacementWithUsedNodes()
usedNodes.clear();
nodeNum = 3;
datanodeDetails =
- policy.chooseDatanodes(usedNodes, null, null, nodeNum, 15, 15);
+ policy.chooseDatanodes(usedNodes, null, null, nodeNum, 15, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
}
@@ -312,7 +313,7 @@ public void testPipelineProviderRackPlacementWithUsedAndExcludeNodes()
excludeNodes.add(datanodes.get(3));
int nodeNum = 2;
List<DatanodeDetails> datanodeDetails =
- policy.chooseDatanodes(usedNodes, excludeNodes, null, nodeNum, 15, 15);
+ policy.chooseDatanodes(usedNodes, excludeNodes, null, nodeNum, 15, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
// policy should not return any of excluded node
assertNotSame(datanodeDetails.get(0).getID(), excludeNodes.get(0).getID());
@@ -332,7 +333,7 @@ public void testPipelineProviderRackPlacementWithUsedAndExcludeNodes()
excludeNodes.add(datanodes.get(2));
nodeNum = 1;
datanodeDetails =
- policy.chooseDatanodes(usedNodes, excludeNodes, null, nodeNum, 15, 15);
+ policy.chooseDatanodes(usedNodes, excludeNodes, null, nodeNum, 15, 15, StorageType.DEFAULT);
assertEquals(nodeNum, datanodeDetails.size());
// policy should not return any of excluded node
assertNotSame(datanodeDetails.get(0).getID(), excludeNodes.get(0).getID());
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 d76bf22b3fe..d91d608b588 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
@@ -43,6 +43,7 @@
import java.util.List;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.RatisReplicationConfig;
import org.apache.hadoop.hdds.client.ReplicationConfig;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -113,13 +114,14 @@ public class TestPipelinePlacementPolicy {
private List<DatanodeDetails> nodesWithOutRackAwareness = new ArrayList<>();
private List<DatanodeDetails> nodesWithRackAwareness = new ArrayList<>();
+ private final StorageType storageType = StorageType.DEFAULT;
@BeforeEach
public void init() throws Exception {
cluster = initTopology();
// start with nodes with rack awareness.
nodeManager = new MockNodeManager(cluster, getNodesWithRackAwareness(),
- false, PIPELINE_PLACEMENT_MAX_NODES_COUNT);
+ false, PIPELINE_PLACEMENT_MAX_NODES_COUNT, storageType);
conf = SCMTestUtils.getConf(testDir);
conf.setInt(OZONE_DATANODE_PIPELINE_LIMIT, PIPELINE_LOAD_LIMIT);
conf.setStorageSize(OZONE_DATANODE_RATIS_VOLUME_FREE_SPACE_MIN,
@@ -204,7 +206,7 @@ public void testChooseNodeWithSingleNodeRack() throws IOException {
datanodes.add(datanode);
}
MockNodeManager localNodeManager = new MockNodeManager(cluster,
- datanodes, false, datanodes.size());
+ datanodes, false, datanodes.size(), storageType);
PipelineStateManager tempPipelineStateManager = PipelineStateManagerImpl
.newBuilder().setNodeManager(localNodeManager)
@@ -219,7 +221,7 @@ public void testChooseNodeWithSingleNodeRack() throws IOException {
List<DatanodeDetails> results = localPlacementPolicy.chooseDatanodes(
new ArrayList<>(datanodes.size()),
new ArrayList<>(datanodes.size()),
- nodesRequired, 0, 0);
+ nodesRequired, 0, 0, storageType);
assertEquals(nodesRequired, results.size());
// 3 nodes should be on different racks.
@@ -241,7 +243,7 @@ public void testChooseNodeNotEnoughSpace() throws IOException {
datanodes.add(datanode);
}
MockNodeManager localNodeManager = new MockNodeManager(cluster,
- datanodes, false, datanodes.size());
+ datanodes, false, datanodes.size(), storageType);
PipelineStateManager tempPipelineStateManager = PipelineStateManagerImpl
.newBuilder().setNodeManager(localNodeManager)
@@ -262,13 +264,13 @@ public void testChooseNodeNotEnoughSpace() throws IOException {
SCMException ex =
assertThrows(SCMException.class,
() -> localPlacementPolicy.chooseDatanodes(new ArrayList<>(datanodes.size()),
- new ArrayList<>(datanodes.size()), nodesRequired, 0, 10 * OzoneConsts.TB));
+ new ArrayList<>(datanodes.size()), nodesRequired, 0, 10 * OzoneConsts.TB, storageType));
assertThat(ex.getMessage()).contains(expectedMessageSubstring);
// a huge free space min configured
ex = assertThrows(SCMException.class,
() -> localPlacementPolicy.chooseDatanodes(new ArrayList<>(datanodes.size()),
- new ArrayList<>(datanodes.size()), nodesRequired, 10 * OzoneConsts.TB, 0));
+ new ArrayList<>(datanodes.size()), nodesRequired, 10 * OzoneConsts.TB, 0, storageType));
assertThat(ex.getMessage()).contains(expectedMessageSubstring);
}
@@ -282,7 +284,7 @@ public void testPickLowestLoadAnchor() throws IOException, TimeoutException {
for (int i = 0; i < maxPipelineCount; i++) {
try {
List<DatanodeDetails> nodes = placementPolicy.chooseDatanodes(null,
- null, HddsProtos.ReplicationFactor.THREE.getNumber(), 0, 0);
+ null, HddsProtos.ReplicationFactor.THREE.getNumber(), 0, 0, storageType);
Pipeline pipeline = Pipeline.newBuilder()
.setId(PipelineID.randomId())
@@ -421,7 +423,7 @@ public void testHeavyNodeShouldBeExcludedWithMinorityHeavy()
List<DatanodeDetails> pickedNodes1 = placementPolicy.chooseDatanodes(
new ArrayList<>(PIPELINE_PLACEMENT_MAX_NODES_COUNT),
new ArrayList<>(PIPELINE_PLACEMENT_MAX_NODES_COUNT),
- nodesRequired, 0, 0);
+ nodesRequired, 0, 0, storageType);
// modify node to pipeline mapping.
insertHeavyNodesIntoNodeManager(healthyNodes, minorityHeavy);
// NODES should be sufficient.
@@ -437,7 +439,7 @@ public void testHeavyNodeShouldBeExcludedWithMinorityHeavy()
placementPolicy.chooseDatanodes(
new ArrayList<>(PIPELINE_PLACEMENT_MAX_NODES_COUNT),
new ArrayList<>(PIPELINE_PLACEMENT_MAX_NODES_COUNT),
- nodesRequired, 0, 0));
+ nodesRequired, 0, 0, storageType));
}
@Test
@@ -454,13 +456,13 @@ public void testHeavyNodeShouldBeExcludedWithMajorityHeavy()
placementPolicy.chooseDatanodes(
new ArrayList<>(PIPELINE_PLACEMENT_MAX_NODES_COUNT),
new ArrayList<>(PIPELINE_PLACEMENT_MAX_NODES_COUNT),
- nodesRequired, 0, 0));
+ nodesRequired, 0, 0, storageType));
}
@Test
public void testValidatePlacementPolicyOK() {
nodeManager = new MockNodeManager(cluster, getNodesWithRackAwareness(),
- false, PIPELINE_PLACEMENT_MAX_NODES_COUNT);
+ false, PIPELINE_PLACEMENT_MAX_NODES_COUNT, storageType);
placementPolicy = new PipelinePlacementPolicy(
nodeManager, stateManager, conf);
@@ -514,7 +516,7 @@ public void testValidatePlacementPolicySingleRackInCluster() {
NetworkTopologyImpl localCluster = initTopology();
nodeManager = new MockNodeManager(localCluster, new ArrayList<>(),
- false, PIPELINE_PLACEMENT_MAX_NODES_COUNT);
+ false, PIPELINE_PLACEMENT_MAX_NODES_COUNT, storageType);
placementPolicy = new PipelinePlacementPolicy(
nodeManager, stateManager, conf);
@@ -546,7 +548,7 @@ public void test3NodesInSameRackReturnedWhenOnlyOneHealthyRackIsPresent()
// As there is only 1 rack alive, the 3 DNs on /rack2 should be returned
List<DatanodeDetails> pickedDns = placementPolicy.chooseDatanodes(
- new ArrayList<>(), new ArrayList<>(), nodesRequired, 0, 0);
+ new ArrayList<>(), new ArrayList<>(), nodesRequired, 0, 0, storageType);
assertEquals(3, pickedDns.size());
assertThat(pickedDns).contains(dns.get(1));
@@ -567,7 +569,7 @@ public void testExceptionIsThrownWhenRackAwarePipelineCanNotBeCreated()
Throwable t = assertThrows(SCMException.class, () ->
placementPolicy.chooseDatanodes(
- new ArrayList<>(), new ArrayList<>(), nodesRequired, 0, 0));
+ new ArrayList<>(), new ArrayList<>(), nodesRequired, 0, 0, storageType));
assertEquals(PipelinePlacementPolicy.MULTIPLE_RACK_PIPELINE_MSG, t.getMessage());
}
@@ -586,7 +588,7 @@ public void testExceptionThrownRackAwarePipelineCanNotBeCreatedExcludedNode()
excluded.add(dns.get(0));
Throwable t = assertThrows(SCMException.class, () ->
placementPolicy.chooseDatanodes(
- excluded, new ArrayList<>(), nodesRequired, 0, 0));
+ excluded, new ArrayList<>(), nodesRequired, 0, 0, storageType));
assertEquals(PipelinePlacementPolicy.MULTIPLE_RACK_PIPELINE_MSG, t.getMessage());
}
@@ -602,7 +604,7 @@ private List<DatanodeDetails> setupSkewedRacks() {
.createDatanodeDetails("host4", "/rack2"));
nodeManager = new MockNodeManager(cluster, dns,
- false, PIPELINE_PLACEMENT_MAX_NODES_COUNT);
+ false, PIPELINE_PLACEMENT_MAX_NODES_COUNT, storageType);
placementPolicy = new PipelinePlacementPolicy(
nodeManager, stateManager, conf);
return dns;
@@ -770,7 +772,7 @@ public void testPipelinePlacementPolicyDefaultLimitFiltersNodeAtLimit()
// 3) Verifies node is filtered out when choosing nodes for new pipeline
int nodesRequired = HddsProtos.ReplicationFactor.THREE.getNumber();
List<DatanodeDetails> chosen = localPolicy.chooseDatanodes(
- new ArrayList<>(), new ArrayList<>(), nodesRequired, 0, 0);
+ new ArrayList<>(), new ArrayList<>(), nodesRequired, 0, 0, StorageType.DEFAULT);
assertEquals(nodesRequired, chosen.size());
assertThat(chosen).doesNotContain(target);
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 5017a65aca4..1ec92b31ea5 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
@@ -22,8 +22,10 @@
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_RATIS_VOLUME_FREE_SPACE_MIN;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_PIPELINE_PLACEMENT_IMPL_KEY;
+import static org.apache.hadoop.hdds.scm.exceptions.SCMException.ResultCodes.CANNOT_CREATE_PIPELINE_FOR_EMPTY_TIER;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -39,9 +41,13 @@
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.client.RatisReplicationConfig;
import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.client.StorageTier;
+import org.apache.hadoop.hdds.client.StorageTierUtil;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.conf.StorageUnit;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -71,6 +77,7 @@
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.MethodSource;
/**
* Test for {@link RatisPipelineProvider}.
@@ -88,25 +95,34 @@ public class TestRatisPipelineProvider {
private DBStore dbStore;
private int nodeCount = 10;
- public void init(int maxPipelinePerNode) throws Exception {
- init(maxPipelinePerNode, new OzoneConfiguration());
+ public void init(int maxPipelinePerNode, StorageTier storageTier) throws Exception {
+ init(maxPipelinePerNode, new OzoneConfiguration(), storageTier);
}
- public void init(int maxPipelinePerNode, OzoneConfiguration conf)
+ public void init(int maxPipelinePerNode, OzoneConfiguration conf, StorageTier storageTier)
throws Exception {
- init(maxPipelinePerNode, conf, testDir);
+ init(maxPipelinePerNode, conf, testDir, storageTier);
}
public void initWithNodes(int maxPipelinePerNode, OzoneConfiguration conf, List<DatanodeDetails> nodes, int count)
throws Exception {
conf.set(HddsConfigKeys.OZONE_METADATA_DIRS, testDir.getAbsolutePath());
- nodeManager = new MockNodeManager(new NetworkTopologyImpl(new OzoneConfiguration()), nodes, false, count);
+ StorageType storageType = StorageTierUtil.getStorageTypeForUniformStorageTier(
+ StorageTier.getDefaultTier(),
+ RatisReplicationConfig.getInstance(ReplicationFactor.ONE));
+ nodeManager = new MockNodeManager(new NetworkTopologyImpl(new OzoneConfiguration()), nodes, false, count,
+ storageType);
initializeCommonState(maxPipelinePerNode, conf);
}
- public void init(int maxPipelinePerNode, OzoneConfiguration conf, File dir) throws Exception {
+ public void init(int maxPipelinePerNode, OzoneConfiguration conf, File dir,
+ StorageTier storageTier) throws Exception {
+ assertTrue(storageTier.isUniform(), "Only support uniform StorageTier");
+ assertFalse(storageTier.equals(StorageTier.EMPTY), "not support the EMPTY StorageTier");
+ StorageType storageType = storageTier.getStorageTypes(
+ RatisReplicationConfig.getInstance(ReplicationFactor.ONE).getRequiredNodes()).get(0);
conf.set(HddsConfigKeys.OZONE_METADATA_DIRS, dir.getAbsolutePath());
- nodeManager = new MockNodeManager(true, nodeCount);
+ nodeManager = new MockNodeManager(true, nodeCount, storageType);
initializeCommonState(maxPipelinePerNode, conf);
}
@@ -118,7 +134,7 @@ private void initializeCommonState(int maxPipelinePerNode, OzoneConfiguration co
ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE_DEFAULT, StorageUnit.BYTES);
nodeManager.setPendingContainerMaxSize(containerSize);
SCMHAManager scmhaManager = SCMHAManagerStub.getInstance(true);
- conf.setInt(ScmConfigKeys.OZONE_DATANODE_PIPELINE_LIMIT,
+ conf.setInt(OZONE_DATANODE_PIPELINE_LIMIT,
maxPipelinePerNode);
stateManager = PipelineStateManagerImpl.newBuilder()
.setPipelineStore(SCMDBDefinition.PIPELINES.getTable(dbStore))
@@ -149,10 +165,10 @@ private static void assertPipelineProperties(
}
private void createPipelineAndAssertions(
- HddsProtos.ReplicationFactor factor)
+ HddsProtos.ReplicationFactor factor, StorageTier storageTier)
throws IOException, TimeoutException {
Pipeline pipeline = provider.create(RatisReplicationConfig
- .getInstance(factor));
+ .getInstance(factor), storageTier);
assertPipelineProperties(pipeline, factor, REPLICATION_TYPE,
Pipeline.PipelineState.ALLOCATED);
HddsProtos.Pipeline pipelineProto = pipeline.getProtobufMessage(
@@ -161,7 +177,7 @@ private void createPipelineAndAssertions(
nodeManager.addPipeline(pipeline);
Pipeline pipeline1 = provider.create(RatisReplicationConfig
- .getInstance(factor));
+ .getInstance(factor), storageTier);
HddsProtos.Pipeline pipelineProto1 = pipeline1.getProtobufMessage(
ClientVersion.CURRENT_VERSION);
assertPipelineProperties(pipeline1, factor, REPLICATION_TYPE,
@@ -176,16 +192,18 @@ private void createPipelineAndAssertions(
nodeManager.addPipeline(pipeline1);
}
- @Test
- public void testCreatePipelineWithFactorThree() throws Exception {
- init(1);
- createPipelineAndAssertions(HddsProtos.ReplicationFactor.THREE);
+ @ParameterizedTest
+ @MethodSource("storageTiers")
+ public void testCreatePipelineWithFactorThree(StorageTier storageTier) throws Exception {
+ init(1, storageTier);
+ createPipelineAndAssertions(HddsProtos.ReplicationFactor.THREE, storageTier);
}
- @Test
- public void testCreatePipelineWithFactorOne() throws Exception {
- init(1);
- createPipelineAndAssertions(HddsProtos.ReplicationFactor.ONE);
+ @ParameterizedTest
+ @MethodSource("storageTiers")
+ public void testCreatePipelineWithFactorOne(StorageTier storageTier) throws Exception {
+ init(1, storageTier);
+ createPipelineAndAssertions(HddsProtos.ReplicationFactor.ONE, storageTier);
}
private List<DatanodeDetails> createListOfNodes(int count) {
@@ -196,12 +214,13 @@ private List<DatanodeDetails> createListOfNodes(int count) {
return nodes;
}
- @Test
- public void testCreatePipelineWithFactor() throws Exception {
- init(1);
+ @ParameterizedTest
+ @MethodSource("storageTiers")
+ public void testCreatePipelineWithFactor(StorageTier storageTier) throws Exception {
+ init(1, storageTier);
HddsProtos.ReplicationFactor factor = HddsProtos.ReplicationFactor.THREE;
Pipeline pipeline = provider.create(RatisReplicationConfig
- .getInstance(factor));
+ .getInstance(factor), storageTier);
assertPipelineProperties(pipeline, factor, REPLICATION_TYPE,
Pipeline.PipelineState.ALLOCATED);
HddsProtos.Pipeline pipelineProto = pipeline.getProtobufMessage(
@@ -210,7 +229,7 @@ public void testCreatePipelineWithFactor() throws Exception {
factor = HddsProtos.ReplicationFactor.ONE;
Pipeline pipeline1 = provider.create(RatisReplicationConfig
- .getInstance(factor));
+ .getInstance(factor), storageTier);
assertPipelineProperties(pipeline1, factor, REPLICATION_TYPE,
Pipeline.PipelineState.ALLOCATED);
HddsProtos.Pipeline pipelineProto1 = pipeline1.getProtobufMessage(
@@ -223,7 +242,7 @@ public void testCreatePipelineWithFactor() throws Exception {
@Test
public void testCreatePipelineWithNodes() throws Exception {
- init(1);
+ init(1, StorageTier.getDefaultTier());
HddsProtos.ReplicationFactor factor = HddsProtos.ReplicationFactor.THREE;
Pipeline pipeline =
provider.create(RatisReplicationConfig.getInstance(factor),
@@ -241,7 +260,7 @@ public void testCreatePipelineWithNodes() throws Exception {
@Test
public void testCreateFactorTHREEPipelineWithSameDatanodes()
throws Exception {
- init(2);
+ init(2, StorageTier.getDefaultTier());
List<DatanodeDetails> healthyNodes = nodeManager
.getNodes(NodeStatus.inServiceHealthy()).stream()
.limit(3).collect(Collectors.toList());
@@ -259,6 +278,7 @@ public void testCreateFactorTHREEPipelineWithSameDatanodes()
assertEquals(pipeline1.getNodeSet(), pipeline2.getNodeSet());
assertEquals(pipeline2.getNodeSet(), pipeline3.getNodeSet());
+ assertEquals(pipeline1.getNodeSet(), pipeline2.getNodeSet());
}
private DatanodeDetails createDatanodeDetails(boolean supportRatisStreaming) {
@@ -301,7 +321,8 @@ public void testCreatePipelinePrioritizesRatisStreamingNodes() throws Exception
}
initWithNodes(1, conf, nodes, 3);
- Pipeline pipeline = provider.create(RatisReplicationConfig.getInstance(ReplicationFactor.THREE));
+ Pipeline pipeline = provider.create(RatisReplicationConfig.getInstance(ReplicationFactor.THREE),
+ StorageTier.getDefaultTier());
assertEquals(3, pipeline.getNodes().size());
for (DatanodeDetails dn : pipeline.getNodes()) {
assertTrue(dn.hasPort(DatanodeDetails.Port.Name.RATIS_DATASTREAM),
@@ -323,7 +344,8 @@ public void testCreatePipelineFallsBackWhenNotEnoughRatisStreamingNodes() throws
nodes.add(createDatanodeDetails(false));
initWithNodes(1, conf, nodes, 3);
- Pipeline pipeline = provider.create(RatisReplicationConfig.getInstance(ReplicationFactor.THREE));
+ Pipeline pipeline = provider.create(RatisReplicationConfig.getInstance(ReplicationFactor.THREE),
+ StorageTier.getDefaultTier());
assertEquals(3, pipeline.getNodes().size());
long streamingNodeCount = pipeline.getNodes().stream()
@@ -334,11 +356,12 @@ public void testCreatePipelineFallsBackWhenNotEnoughRatisStreamingNodes() throws
"Pipeline should contain exactly 2 nodes with RATIS_DATASTREAM as fallback was required");
}
- @Test
- public void testCreatePipelinesDnExclude() throws Exception {
+ @ParameterizedTest
+ @MethodSource("storageTiers")
+ public void testCreatePipelinesDnExclude(StorageTier storageTier) throws Exception {
int maxPipelinePerNode = 2;
- init(maxPipelinePerNode);
+ init(maxPipelinePerNode, storageTier);
List<DatanodeDetails> healthyNodes =
nodeManager.getNodes(NodeStatus.inServiceHealthy());
@@ -363,7 +386,7 @@ public void testCreatePipelinesDnExclude() throws Exception {
// only 2 healthy DNs left that are not part of any pipeline
Pipeline pipeline = provider.create(
- RatisReplicationConfig.getInstance(factor));
+ RatisReplicationConfig.getInstance(factor), storageTier);
assertPipelineProperties(pipeline, factor, REPLICATION_TYPE,
Pipeline.PipelineState.ALLOCATED);
HddsProtos.Pipeline pipelineProto = pipeline.getProtobufMessage(
@@ -388,7 +411,7 @@ public void testCreatePipelinesDnExclude() throws Exception {
// favored nodes.
public void testCreateFactorTHREEPipelineWithExcludedDatanodes()
throws Exception {
- init(1);
+ init(1, StorageTier.getDefaultTier());
int healthyCount = nodeManager.getNodes(NodeStatus.inServiceHealthy())
.size();
// Add all but 3 nodes to the exclude list and ensure that the 3 picked
@@ -399,33 +422,36 @@ public void testCreateFactorTHREEPipelineWithExcludedDatanodes()
Pipeline pipeline1 = provider.create(
RatisReplicationConfig.getInstance(ReplicationFactor.THREE),
- excludedNodes, Collections.EMPTY_LIST);
+ excludedNodes, Collections.EMPTY_LIST, StorageTier.getDefaultTier());
for (DatanodeDetails dn : pipeline1.getNodes()) {
assertThat(excludedNodes).doesNotContain(dn);
}
}
- @Test
+ @ParameterizedTest
+ @MethodSource("storageTiers")
// Test pipeline provider with RackScatter policy cannot create
// pipeline due to nodes with full pipeline engagement.
- public void testFactorTHREEPipelineRackScatterEngagement()
+ public void testFactorTHREEPipelineRackScatterEngagement(StorageTier storageTier)
throws Exception {
OzoneConfiguration conf = new OzoneConfiguration();
conf.set(OZONE_SCM_PIPELINE_PLACEMENT_IMPL_KEY,
SCMContainerPlacementRackScatter.class.getCanonicalName());
conf.set(OZONE_DATANODE_PIPELINE_LIMIT, "0");
- init(0, conf);
+ init(0, conf, storageTier);
List<DatanodeDetails> excludedNodes = new ArrayList<>();
assertThrows(SCMException.class, () ->
provider.create(RatisReplicationConfig
.getInstance(ReplicationFactor.THREE),
- excludedNodes, Collections.EMPTY_LIST));
+ excludedNodes, Collections.EMPTY_LIST, StorageTier.getDefaultTier()));
}
- @Test
- public void testCreatePipelinesWhenNotEnoughSpace(@TempDir File tempDir) throws Exception {
+ @ParameterizedTest
+ @MethodSource("storageTiers")
+ public void testCreatePipelinesWhenNotEnoughSpace(StorageTier storageTier,
+ @TempDir File tempDir) throws Exception {
String expectedErrorSubstring = "Unable to find enough" +
" nodes that meet the space requirement";
@@ -433,26 +459,27 @@ public void testCreatePipelinesWhenNotEnoughSpace(@TempDir File tempDir) throws
// enough space to hold one.
OzoneConfiguration largeContainerConf = new OzoneConfiguration();
largeContainerConf.set(OZONE_SCM_CONTAINER_SIZE, "300TB");
- init(1, largeContainerConf);
+ init(1, largeContainerConf, storageTier);
for (ReplicationFactor factor: ReplicationFactor.values()) {
if (factor == ReplicationFactor.ZERO) {
continue;
}
- SCMException ex =
- assertThrows(SCMException.class, () -> provider.create(RatisReplicationConfig.getInstance(factor)),
+ SCMException ex = assertThrows(SCMException.class,
+ () -> provider.create(RatisReplicationConfig.getInstance(factor), storageTier),
"Expected SCMException for large container size with replication factor " + factor.toString());
assertThat(ex.getMessage()).contains(expectedErrorSubstring);
}
OzoneConfiguration largeMetadataConf = new OzoneConfiguration();
largeMetadataConf.set(OZONE_DATANODE_RATIS_VOLUME_FREE_SPACE_MIN, "300TB");
- init(1, largeMetadataConf, tempDir);
+ init(1, largeMetadataConf, tempDir, storageTier);
for (ReplicationFactor factor: ReplicationFactor.values()) {
if (factor == ReplicationFactor.ZERO) {
continue;
}
SCMException ex =
- assertThrows(SCMException.class, () -> provider.create(RatisReplicationConfig.getInstance(factor)),
+ assertThrows(SCMException.class,
+ () -> provider.create(RatisReplicationConfig.getInstance(factor), storageTier),
"Expected SCMException for large metadata size with replication factor " + factor.toString());
assertThat(ex.getMessage()).contains(expectedErrorSubstring);
}
@@ -493,14 +520,14 @@ public void testCreatePipelineWithDefaultLimit() throws Exception {
for (int i = 0; i < maxPipelines; i++) {
Pipeline p = provider.create(
RatisReplicationConfig.getInstance(ReplicationFactor.THREE),
- new ArrayList<>(), new ArrayList<>());
+ new ArrayList<>(), new ArrayList<>(), StorageTier.getDefaultTier());
stateManager.addPipeline(p.getProtobufMessage(ClientVersion.CURRENT_VERSION));
}
// Next pipeline creation should fail with default limit message.
SCMException ex = assertThrows(SCMException.class, () ->
provider.create(RatisReplicationConfig.getInstance(ReplicationFactor.THREE),
- new ArrayList<>(), new ArrayList<>())
+ new ArrayList<>(), new ArrayList<>(), StorageTier.getDefaultTier())
);
assertThat(ex.getMessage())
@@ -513,20 +540,21 @@ public void testCreatePipelineWithDefaultLimit() throws Exception {
public void testCreatePipelineThrowErrorWithDataNodeLimit(int limit, int pipelineCount) throws Exception {
// increasing node count to avoid intermittent failures due to unhealthy nodes.
nodeCount = 13;
- init(limit, new OzoneConfiguration(), testDir);
+ init(limit, new OzoneConfiguration(), testDir, StorageTier.getDefaultTier());
// Create pipelines up to the limit (3 for limit=1, 6 for limit=2).
for (int i = 0; i < pipelineCount; i++) {
stateManager.addPipeline(
provider.create(RatisReplicationConfig.getInstance(ReplicationFactor.THREE),
- new ArrayList<>(), new ArrayList<>()).getProtobufMessage(ClientVersion.CURRENT_VERSION)
+ new ArrayList<>(), new ArrayList<>(), StorageTier.getDefaultTier())
+ .getProtobufMessage(ClientVersion.CURRENT_VERSION)
);
}
// Verify that creating an additional pipeline throws an exception.
SCMException exception = assertThrows(SCMException.class, () ->
provider.create(RatisReplicationConfig.getInstance(ReplicationFactor.THREE),
- new ArrayList<>(), new ArrayList<>())
+ new ArrayList<>(), new ArrayList<>(), StorageTier.getDefaultTier())
);
// Validate exception message.
@@ -535,6 +563,21 @@ public void testCreatePipelineThrowErrorWithDataNodeLimit(int limit, int pipelin
assertEquals(expectedError, exception.getMessage());
}
+ @Test
+ public void testCreatePipelinesInEmptyTier() throws Exception {
+ init(1, StorageTier.getDefaultTier());
+ for (ReplicationFactor factor: ReplicationFactor.values()) {
+ if (factor == ReplicationFactor.ZERO) {
+ continue;
+ }
+ SCMException ex =
+ assertThrows(SCMException.class, () ->
+ provider.create(RatisReplicationConfig.getInstance(factor), StorageTier.EMPTY),
+ "Expected SCMException for empty StorageTier" + factor.toString());
+ assertEquals(CANNOT_CREATE_PIPELINE_FOR_EMPTY_TIER, ex.getResult());
+ }
+ }
+
private void addPipeline(
List<DatanodeDetails> dns,
Pipeline.PipelineState open, ReplicationConfig replicationConfig)
@@ -571,4 +614,12 @@ private Set<ContainerReplica> createContainerReplicas(
}
return replicas;
}
+
+ static Stream<StorageTier> storageTiers() {
+ return Stream.of(
+ StorageTier.DISK,
+ StorageTier.SSD,
+ StorageTier.ARCHIVE
+ );
+ }
}
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSimplePipelineProvider.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSimplePipelineProvider.java
index 338d9129512..42aa1943aa7 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSimplePipelineProvider.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSimplePipelineProvider.java
@@ -17,18 +17,27 @@
package org.apache.hadoop.hdds.scm.pipeline;
+import static org.apache.hadoop.hdds.scm.exceptions.SCMException.ResultCodes.CANNOT_CREATE_PIPELINE_FOR_EMPTY_TIER;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.File;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import java.util.stream.Stream;
+import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.RatisReplicationConfig;
import org.apache.hadoop.hdds.client.StandaloneReplicationConfig;
+import org.apache.hadoop.hdds.client.StorageTier;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.MockDatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.scm.container.MockNodeManager;
+import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.hdds.scm.ha.SCMHAManager;
import org.apache.hadoop.hdds.scm.ha.SCMHAManagerStub;
import org.apache.hadoop.hdds.scm.metadata.SCMDBDefinition;
@@ -41,6 +50,8 @@
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
/**
* Test for SimplePipelineProvider.
@@ -54,10 +65,19 @@ public class TestSimplePipelineProvider {
private DBStore dbStore;
@BeforeEach
- public void init() throws Exception {
- NodeManager nodeManager = new MockNodeManager(true, 10);
- final OzoneConfiguration conf = SCMTestUtils.getConf(testDir);
+ public void startup() throws Exception {
+ OzoneConfiguration conf = SCMTestUtils.getConf(testDir);
+ conf.set(HddsConfigKeys.OZONE_METADATA_DIRS,
+ new File(testDir, "metadata").getAbsolutePath());
dbStore = DBStoreBuilder.createDBStore(conf, SCMDBDefinition.get());
+ }
+
+ public void init(StorageTier storageTier) throws Exception {
+ assertTrue(storageTier.isUniform(), "Only support uniform StorageTier");
+ assertFalse(storageTier.equals(StorageTier.EMPTY), "not support the EMPTY StorageTier");
+ StorageType storageType = storageTier.getStorageTypes(
+ RatisReplicationConfig.getInstance(HddsProtos.ReplicationFactor.ONE).getRequiredNodes()).get(0);
+ NodeManager nodeManager = new MockNodeManager(true, 10, storageType);
SCMHAManager scmhaManager = SCMHAManagerStub.getInstance(true);
stateManager = PipelineStateManagerImpl.newBuilder()
.setPipelineStore(SCMDBDefinition.PIPELINES.getTable(dbStore))
@@ -75,11 +95,13 @@ public void cleanup() throws Exception {
}
}
- @Test
- public void testCreatePipelineWithFactor() throws Exception {
+ @ParameterizedTest
+ @MethodSource("storageTiers")
+ public void testCreatePipelineWithFactor(StorageTier storageTier) throws Exception {
+ init(storageTier);
HddsProtos.ReplicationFactor factor = HddsProtos.ReplicationFactor.THREE;
Pipeline pipeline =
- provider.create(StandaloneReplicationConfig.getInstance(factor));
+ provider.create(StandaloneReplicationConfig.getInstance(factor), storageTier);
HddsProtos.Pipeline pipelineProto = pipeline.getProtobufMessage(
ClientVersion.CURRENT_VERSION);
stateManager.addPipeline(pipelineProto);
@@ -90,7 +112,7 @@ public void testCreatePipelineWithFactor() throws Exception {
factor = HddsProtos.ReplicationFactor.ONE;
Pipeline pipeline1 =
- provider.create(StandaloneReplicationConfig.getInstance(factor));
+ provider.create(StandaloneReplicationConfig.getInstance(factor), storageTier);
HddsProtos.Pipeline pipelineProto1 = pipeline1.getProtobufMessage(
ClientVersion.CURRENT_VERSION);
stateManager.addPipeline(pipelineProto1);
@@ -111,7 +133,9 @@ private List<DatanodeDetails> createListOfNodes(int nodeCount) {
}
@Test
- public void testCreatePipelineWithNodes() throws IOException {
+ public void testCreatePipelineWithNodes()
+ throws Exception {
+ init(StorageTier.getDefaultTier());
HddsProtos.ReplicationFactor factor = HddsProtos.ReplicationFactor.THREE;
Pipeline pipeline =
provider.create(StandaloneReplicationConfig.getInstance(factor),
@@ -134,4 +158,26 @@ public void testCreatePipelineWithNodes() throws IOException {
assertEquals(pipeline.getPipelineState(), Pipeline.PipelineState.OPEN);
assertEquals(pipeline.getNodes().size(), factor.getNumber());
}
+
+ @Test
+ public void testCreatePipelinesInEmptyTier() throws Exception {
+ init(StorageTier.getDefaultTier());
+ for (HddsProtos.ReplicationFactor factor: HddsProtos.ReplicationFactor.values()) {
+ if (factor == HddsProtos.ReplicationFactor.ZERO) {
+ continue;
+ }
+ SCMException ex = assertThrows(SCMException.class, () ->
+ provider.create(StandaloneReplicationConfig.getInstance(factor), StorageTier.EMPTY),
+ "Expected SCMException for empty StorageTier" + factor.toString());
+ assertEquals(ex.getResult(), CANNOT_CREATE_PIPELINE_FOR_EMPTY_TIER);
+ }
+ }
+
+ static Stream<StorageTier> storageTiers() {
+ return Stream.of(
+ StorageTier.DISK,
+ StorageTier.SSD,
+ StorageTier.ARCHIVE
+ );
+ }
}
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/placement/TestContainerPlacement.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/placement/TestContainerPlacement.java
index c57009a5a71..439311f61be 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/placement/TestContainerPlacement.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/placement/TestContainerPlacement.java
@@ -25,6 +25,7 @@
import java.util.List;
import java.util.Random;
import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.scm.container.MockNodeManager;
@@ -91,12 +92,12 @@ public void testCapacityPlacementYieldsBetterDataDistribution() throws
long metadataSize = random.nextInt(10) * OzoneConsts.GB;
List<DatanodeDetails> nodesCapacity =
capacityPlacer.chooseDatanodes(new ArrayList<>(), null, nodesRequired,
- metadataSize, containerSize);
+ metadataSize, containerSize, StorageType.DEFAULT);
assertEquals(nodesRequired, nodesCapacity.size());
List<DatanodeDetails> nodesRandom =
randomPlacer.chooseDatanodes(nodesCapacity, null, nodesRequired,
- metadataSize, containerSize);
+ metadataSize, containerSize, StorageType.DEFAULT);
// One fifth of all calls are delete
if (x % 5 == 0) {
diff --git a/hadoop-ozone/common/src/test/java/org/apache/hadoop/hdds/protocol/StorageTierTest.java b/hadoop-ozone/common/src/test/java/org/apache/hadoop/hdds/protocol/StorageTierTest.java
index 58907bc2650..0630dd45cd3 100644
--- a/hadoop-ozone/common/src/test/java/org/apache/hadoop/hdds/protocol/StorageTierTest.java
+++ b/hadoop-ozone/common/src/test/java/org/apache/hadoop/hdds/protocol/StorageTierTest.java
@@ -35,9 +35,9 @@ class StorageTierTest {
@Test
void testUniformStorageType() {
- assertTrue(StorageTier.SSD.isUniformStorageType());
- assertTrue(StorageTier.DISK.isUniformStorageType());
- assertTrue(StorageTier.ARCHIVE.isUniformStorageType());
+ assertTrue(StorageTier.SSD.isUniform());
+ assertTrue(StorageTier.DISK.isUniform());
+ assertTrue(StorageTier.ARCHIVE.isUniform());
}
@Test
@@ -51,15 +51,15 @@ void testGetStorageTypesWithReplicationConfig() {
// Assert uniform storage types
Arrays.asList(StorageTier.SSD, StorageTier.DISK, StorageTier.ARCHIVE, StorageTier.EMPTY)
- .forEach(tier -> assertTrue(tier.isUniformStorageType()));
+ .forEach(tier -> assertTrue(tier.isUniform()));
for (StorageTier tier : StorageTier.values()) {
- if (!tier.isUniformStorageType()) {
+ if (!tier.isUniform()) {
return;
}
for (ReplicationConfig replicationConfig : Arrays.asList(ratisOne, ratisThree,
standaloneOne, standaloneThree)) {
- List<StorageType> storageTypes = tier.getStorageTypes(replicationConfig);
+ List<StorageType> storageTypes = tier.getStorageTypes(replicationConfig.getRequiredNodes());
if (tier.equals(StorageTier.EMPTY)) {
assertEquals(0, storageTypes.size());
} else {
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconPipelineFactory.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconPipelineFactory.java
index 154849c5a45..55fee8f8653 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconPipelineFactory.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconPipelineFactory.java
@@ -21,6 +21,7 @@
import java.util.Set;
import org.apache.commons.collections4.map.DefaultedMap;
import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.client.StorageTier;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.scm.container.ContainerReplica;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
@@ -43,7 +44,8 @@ static class ReconPipelineProvider extends
PipelineProvider<ReplicationConfig> {
@Override
- public Pipeline create(ReplicationConfig config) {
+ public Pipeline create(ReplicationConfig config,
+ StorageTier storageTier) {
// We don't expect this to be called at all. But adding this as a red
// flag for troubleshooting.
throw new UnsupportedOperationException(
@@ -53,7 +55,7 @@ public Pipeline create(ReplicationConfig config) {
@Override
public Pipeline create(ReplicationConfig config,
List<DatanodeDetails> excludedNodes,
- List<DatanodeDetails> favoredNodes) {
+ List<DatanodeDetails> favoredNodes, StorageTier storageTier) {
// We don't expect this to be called at all. But adding this as a red
// flag for troubleshooting.
throw new UnsupportedOperationException(
@@ -66,7 +68,7 @@ public Pipeline create(ReplicationConfig config,
throw new UnsupportedOperationException(
"Trying to create pipeline in Recon, which is prohibited!");
}
-
+
@Override
public Pipeline createForRead(ReplicationConfig config,
Set<ContainerReplica> replicas) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org