This is an automated email from the ASF dual-hosted git repository.
xichen01 pushed a commit to branch HDDS-11233
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/HDDS-11233 by this push:
new 2b371c6075a HDDS-15241. Datanode Support create Container use specific
StorageType (#10362)
2b371c6075a is described below
commit 2b371c6075a58d6140325b149105e77d09c37a20
Author: XiChen <[email protected]>
AuthorDate: Fri May 29 00:17:46 2026 +0800
HDDS-15241. Datanode Support create Container use specific StorageType
(#10362)
---
.../hadoop/hdds/scm/storage/BlockOutputStream.java | 5 +-
.../hadoop/hdds/client/StorageTypeUtils.java | 102 ++++++++
.../hdds/scm/storage/ContainerProtocolCalls.java | 58 +++--
.../ozone/container/common/impl/ContainerData.java | 1 +
.../container/common/impl/HddsDispatcher.java | 23 +-
.../container/common/interfaces/Container.java | 14 +-
.../common/interfaces/VolumeChoosingPolicy.java | 21 +-
.../volume/AbstractStorageTypeChoosingPolicy.java | 6 -
.../container/keyvalue/KeyValueContainer.java | 23 +-
.../ozone/container/keyvalue/KeyValueHandler.java | 18 +-
.../container/replication/ContainerImporter.java | 4 +-
.../ozone/container/common/ContainerTestUtils.java | 3 +-
.../container/common/TestBlockDeletingService.java | 3 +-
.../TestSchemaTwoBackwardsCompatibility.java | 3 +-
...stStaleRecoveringContainerScrubbingService.java | 6 +-
.../common/impl/TestContainerPersistence.java | 3 +-
.../container/common/impl/TestHddsDispatcher.java | 259 +++++++++++++++++++--
.../volume/TestCapacityVolumeChoosingPolicy.java | 7 +-
.../volume/TestRoundRobinVolumeChoosingPolicy.java | 15 +-
.../common/volume/TestVolumeSetDiskChecks.java | 5 +-
.../diskbalancer/TestDiskBalancerTask.java | 5 +-
.../keyvalue/TestKeyValueBlockIterator.java | 3 +-
.../container/keyvalue/TestKeyValueContainer.java | 48 ++--
.../TestKeyValueContainerIntegrityChecks.java | 3 +-
.../TestKeyValueContainerMarkUnhealthy.java | 8 +-
.../container/keyvalue/TestKeyValueHandler.java | 5 +-
.../keyvalue/impl/AbstractTestChunkManager.java | 6 +-
.../keyvalue/impl/TestBlockManagerImpl.java | 6 +-
.../container/ozoneimpl/TestContainerReader.java | 22 +-
.../container/ozoneimpl/TestOzoneContainer.java | 5 +-
.../replication/TestContainerImporter.java | 15 ++
.../replication/TestGrpcReplicationService.java | 3 +-
.../src/main/proto/DatanodeClientProtocol.proto | 6 +
.../schemaupgrade/TestUpgradeContainerSchema.java | 6 +-
.../hadoop/hdds/scm/TestContainerSmallFile.java | 9 +-
.../hadoop/ozone/freon/ChunkManagerDiskWrite.java | 3 +-
.../containergenerator/GeneratorDatanode.java | 3 +-
37 files changed, 559 insertions(+), 176 deletions(-)
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java
index d7076df3ba0..432fee81d19 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java
@@ -963,8 +963,11 @@ private CompletableFuture<PutBlockResult>
writeChunkToContainer(
byteBufferList = null;
}
+ // TODO: Pass the requested storage type from allocation/storage policy
+ // once that context is wired through BlockOutputStream. Null preserves
+ // the current any-volume behavior.
asyncReply = writeChunkAsync(xceiverClient, chunkInfo,
- blockID.get(), data, tokenString, replicationIndex, blockData,
close);
+ blockID.get(), data, tokenString, replicationIndex, blockData,
close, null);
CompletableFuture<ContainerCommandResponseProto>
respFuture = asyncReply.getResponse();
validateFuture = respFuture.thenApplyAsync(e -> {
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
new file mode 100644
index 00000000000..0956581ff87
--- /dev/null
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTypeUtils.java
@@ -0,0 +1,102 @@
+/*
+ * 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 jakarta.annotation.Nonnull;
+import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.StorageTypeProto;
+
+/**
+ * Utility class for converting between Hadoop's {@link StorageType} and
+ * Ozone's protobuf representation {@link HddsProtos.StorageTypeProto}.
+ */
+public final class StorageTypeUtils {
+ private StorageTypeUtils() {
+ }
+
+ public static HddsProtos.StorageTypeProto getStorageTypeProto(@Nonnull
StorageType type)
+ throws IllegalArgumentException {
+ switch (type) {
+ case SSD:
+ return HddsProtos.StorageTypeProto.SSD;
+ case DISK:
+ return HddsProtos.StorageTypeProto.DISK;
+ case ARCHIVE:
+ return HddsProtos.StorageTypeProto.ARCHIVE;
+ case PROVIDED:
+ return HddsProtos.StorageTypeProto.PROVIDED;
+ case RAM_DISK:
+ return HddsProtos.StorageTypeProto.RAM_DISK;
+ default:
+ throw new IllegalArgumentException("Illegal Storage Type specified");
+ }
+ }
+
+ public static StorageType getFromProtobuf(@Nonnull
HddsProtos.StorageTypeProto proto) throws
+ IllegalArgumentException {
+ switch (proto) {
+ case SSD:
+ return StorageType.SSD;
+ case DISK:
+ return StorageType.DISK;
+ case ARCHIVE:
+ return StorageType.ARCHIVE;
+ case PROVIDED:
+ return StorageType.PROVIDED;
+ case RAM_DISK:
+ return StorageType.RAM_DISK;
+ default:
+ throw new IllegalArgumentException("Illegal Storage Type specified");
+ }
+ }
+
+ /**
+ * Returns Filesystem StorageType enum value corresponding to the int ID
value.
+ *
+ * @param storageTypeID StorageType int ID value
+ * @return StorageType
+ */
+ public static StorageType getStorageTypeFromID(int storageTypeID) throws
+ IllegalArgumentException {
+ if (StorageTypeProto.forNumber(storageTypeID) == null) {
+ throw new IllegalArgumentException("Illegal storageTypeID " +
storageTypeID);
+ }
+ return getFromProtobuf(StorageTypeProto.forNumber(storageTypeID));
+ }
+
+ /**
+ * Returns integer representation of protobuf StorageType.
+ *
+ * @return storageType int ID value
+ */
+ public static int getIDFromProtobuf(@Nonnull HddsProtos.StorageTypeProto
proto) throws
+ IllegalArgumentException {
+ switch (proto) {
+ case SSD:
+ case DISK:
+ case ARCHIVE:
+ case PROVIDED:
+ case RAM_DISK:
+ 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/storage/ContainerProtocolCalls.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java
index 15879fb4764..44363806255 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java
@@ -33,6 +33,7 @@
import java.util.function.Function;
import org.apache.hadoop.hdds.annotation.InterfaceStability;
import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.client.StorageTypeUtils;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.BlockData;
@@ -62,6 +63,7 @@
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadContainerResponseProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.WriteChunkRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.scm.XceiverClientReply;
import org.apache.hadoop.hdds.scm.XceiverClientSpi;
import org.apache.hadoop.hdds.scm.XceiverClientSpi.Validator;
@@ -430,23 +432,23 @@ static long getLen(ReadChunkResponseProto response) {
* @param blockID ID of the block
* @param data the data of the chunk to write
* @param tokenString serialized block token
+ * @param storageType - the type of storage that is required, if the
storageType
+ * is null, any storageType will be considered.
* @throws IOException if there is an I/O error while performing the call
*/
@SuppressWarnings("parameternumber")
public static XceiverClientReply writeChunkAsync(
XceiverClientSpi xceiverClient, ChunkInfo chunk, BlockID blockID,
ByteString data, String tokenString,
- int replicationIndex, BlockData blockData, boolean close)
+ int replicationIndex, BlockData blockData, boolean close,
+ HddsProtos.StorageTypeProto storageType)
throws IOException, ExecutionException, InterruptedException {
+ DatanodeBlockID datanodeBlockID = getDatanodeBlockID(
+ blockID, replicationIndex, storageType);
WriteChunkRequestProto.Builder writeChunkRequest =
WriteChunkRequestProto.newBuilder()
- .setBlockID(DatanodeBlockID.newBuilder()
- .setContainerID(blockID.getContainerID())
- .setLocalID(blockID.getLocalID())
- .setBlockCommitSequenceId(blockID.getBlockCommitSequenceId())
- .setReplicaIndex(replicationIndex)
- .build())
+ .setBlockID(datanodeBlockID)
.setChunkData(chunk)
.setData(data);
if (blockData != null) {
@@ -477,18 +479,23 @@ public static XceiverClientReply writeChunkAsync(
* using a single RPC. This API is designed to be used for files which are
* smaller than 1 MB.
*
- * @param client - client that communicates with the container.
- * @param blockID - ID of the block
- * @param data - Data to be written into the container.
- * @param token a token for this block (may be null)
+ * @param client - client that communicates with the container.
+ * @param blockID - ID of the block
+ * @param data - Data to be written into the container.
+ * @param token a token for this block (may be null)
+ * @param storageType - the type of storage that is required, if the
storageType
+ * is null, any storageType will be considered.
* @return container protocol writeSmallFile response
*/
public static PutSmallFileResponseProto writeSmallFile(
XceiverClientSpi client, BlockID blockID, byte[] data,
- Token<OzoneBlockTokenIdentifier> token) throws IOException {
+ Token<OzoneBlockTokenIdentifier> token,
+ HddsProtos.StorageTypeProto storageType) throws IOException {
+ DatanodeBlockID datanodeBlockID = getDatanodeBlockID(
+ blockID, null, storageType);
BlockData containerBlockData =
- BlockData.newBuilder().setBlockID(blockID.getDatanodeBlockIDProtobuf())
+ BlockData.newBuilder().setBlockID(datanodeBlockID)
.build();
PutBlockRequestProto.Builder createBlockRequest =
PutBlockRequestProto.newBuilder()
@@ -509,10 +516,10 @@ public static PutSmallFileResponseProto writeSmallFile(
.setChecksumData(checksumData.getProtoBufMessage())
.build();
- PutSmallFileRequestProto putSmallFileRequest =
+ PutSmallFileRequestProto.Builder putSmallFileBuilder =
PutSmallFileRequestProto.newBuilder().setChunkInfo(chunk)
- .setBlock(createBlockRequest).setData(ByteString.copyFrom(data))
- .build();
+ .setBlock(createBlockRequest)
+ .setData(ByteString.copyFrom(data));
String id = client.getPipeline().getFirstNode().getUuidString();
ContainerCommandRequestProto.Builder builder =
@@ -520,7 +527,7 @@ public static PutSmallFileResponseProto writeSmallFile(
.setCmdType(Type.PutSmallFile)
.setContainerID(blockID.getContainerID())
.setDatanodeUuid(id)
- .setPutSmallFile(putSmallFileRequest);
+ .setPutSmallFile(putSmallFileBuilder);
if (token != null) {
builder.setEncodedToken(token.encodeToUrlString());
}
@@ -573,6 +580,7 @@ public static void createContainer(XceiverClientSpi client,
throws IOException {
ContainerProtos.CreateContainerRequestProto.Builder createRequest =
ContainerProtos.CreateContainerRequestProto.newBuilder();
+ // TODO StoragePolicy Support createContainer Command
createRequest
.setContainerType(ContainerProtos.ContainerType.KeyValueContainer);
if (state != null) {
@@ -937,4 +945,20 @@ static DatanodeBlockID getDatanodeBlockID(BlockID blockID,
DatanodeDetails datan
}
return b.build();
}
+
+ private static DatanodeBlockID getDatanodeBlockID(BlockID blockID,
+ Integer replicationIndex, HddsProtos.StorageTypeProto storageType) {
+ DatanodeBlockID.Builder blockIDBuilder = DatanodeBlockID.newBuilder()
+ .setContainerID(blockID.getContainerID())
+ .setLocalID(blockID.getLocalID())
+ .setBlockCommitSequenceId(blockID.getBlockCommitSequenceId());
+ if (replicationIndex != null) {
+ blockIDBuilder.setReplicaIndex(replicationIndex);
+ }
+ if (storageType != null) {
+ blockIDBuilder.setStorageTypeID(
+ StorageTypeUtils.getIDFromProtobuf(storageType));
+ }
+ return blockIDBuilder.build();
+ }
}
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerData.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerData.java
index 5e6bba1189a..1d9c1f0ef20 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerData.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerData.java
@@ -292,6 +292,7 @@ public void setStorageType(StorageType type) {
storageType = type;
}
+ @Nullable
public StorageType getStorageType() {
return storageType;
}
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java
index a109c02e5a1..a095b7d2a28 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java
@@ -45,6 +45,7 @@
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerType;
+import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.DatanodeBlockID;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type;
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction;
@@ -499,16 +500,28 @@ ContainerCommandResponseProto createContainer(
createRequest.setContainerType(containerType);
if (containerRequest.hasWriteChunk()) {
- createRequest.setReplicaIndex(
- containerRequest.getWriteChunk().getBlockID().getReplicaIndex());
+ DatanodeBlockID blockID = containerRequest.getWriteChunk().getBlockID();
+ createRequest.setReplicaIndex(blockID.getReplicaIndex());
+ if (blockID.hasStorageTypeID() && blockID.getStorageTypeID() > 0) {
+ createRequest.setStorageTypeID(blockID.getStorageTypeID());
+ }
}
if (containerRequest.hasPutBlock()) {
- createRequest.setReplicaIndex(
- containerRequest.getPutBlock().getBlockData().getBlockID()
- .getReplicaIndex());
+ DatanodeBlockID blockID =
containerRequest.getPutBlock().getBlockData().getBlockID();
+ createRequest.setReplicaIndex(blockID.getReplicaIndex());
+ if (blockID.hasStorageTypeID() && blockID.getStorageTypeID() > 0) {
+ createRequest.setStorageTypeID(blockID.getStorageTypeID());
+ }
}
+ if (containerRequest.hasPutSmallFile()) {
+ // PutSmallFile Not support EC yet
+ DatanodeBlockID blockID =
containerRequest.getPutSmallFile().getBlock().getBlockData().getBlockID();
+ if (blockID.hasStorageTypeID() && blockID.getStorageTypeID() > 0) {
+ createRequest.setStorageTypeID(blockID.getStorageTypeID());
+ }
+ }
ContainerCommandRequestProto.Builder requestBuilder =
ContainerCommandRequestProto.newBuilder()
.setCmdType(Type.CreateContainer)
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Container.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Container.java
index ddb7de4140b..07d5db708c3 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Container.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Container.java
@@ -40,23 +40,13 @@
*/
public interface Container<CONTAINERDATA extends ContainerData> {
- /**
- * Creates a container.
- *
- * @throws StorageContainerException
- */
- void create(VolumeSet volumeSet, VolumeChoosingPolicy volumeChoosingPolicy,
- String scmId) throws StorageContainerException;
-
/**
* Creates a container and optionally constrains the destination volume to a
* specific storage type.
*/
- default void create(VolumeSet volumeSet,
+ void create(VolumeSet volumeSet,
VolumeChoosingPolicy volumeChoosingPolicy, String scmId,
- StorageType storageType) throws StorageContainerException {
- create(volumeSet, volumeChoosingPolicy, scmId);
- }
+ StorageType storageType) throws StorageContainerException;
/**
* Deletes the container.
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/VolumeChoosingPolicy.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/VolumeChoosingPolicy.java
index 63d9dccc74b..0e80787510b 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/VolumeChoosingPolicy.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/VolumeChoosingPolicy.java
@@ -17,6 +17,7 @@
package org.apache.hadoop.ozone.container.common.interfaces;
+import jakarta.annotation.Nullable;
import java.io.IOException;
import java.util.List;
import org.apache.hadoop.fs.StorageType;
@@ -29,22 +30,6 @@
@InterfaceAudience.Private
public interface VolumeChoosingPolicy {
- /**
- * Choose a volume to place a container,
- * given a list of volumes and the max container size sought for storage.
- *
- * The implementations of this interface must be thread-safe.
- *
- * @param volumes - a list of available volumes.
- * @param maxContainerSize - the maximum size of the container for which a
- * volume is sought.
- * @return the chosen volume.
- * @throws IOException when disks are unavailable or are full.
- */
- @Deprecated
- HddsVolume chooseVolume(List<HddsVolume> volumes, long maxContainerSize)
- throws IOException;
-
/**
* Choose a volume to place a container, optionally constraining the choice
* to a specific storage type.
@@ -57,6 +42,6 @@ HddsVolume chooseVolume(List<HddsVolume> volumes, long
maxContainerSize)
* @return the chosen volume.
* @throws IOException when disks are unavailable or are full.
*/
- HddsVolume chooseVolume(List<HddsVolume> volumes,
- long maxContainerSize, StorageType storageType) throws IOException;
+ HddsVolume chooseVolume(List<HddsVolume> volumes, long maxContainerSize,
+ @Nullable StorageType storageType) throws IOException;
}
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/AbstractStorageTypeChoosingPolicy.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/AbstractStorageTypeChoosingPolicy.java
index d73ab885e5f..5a013892760 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/AbstractStorageTypeChoosingPolicy.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/AbstractStorageTypeChoosingPolicy.java
@@ -31,12 +31,6 @@ public abstract class AbstractStorageTypeChoosingPolicy
@Override
@Deprecated
- public HddsVolume chooseVolume(List<HddsVolume> volumes,
- long maxContainerSize) throws IOException {
- return chooseVolume(volumes, maxContainerSize, null);
- }
-
- @Override
public HddsVolume chooseVolume(List<HddsVolume> volumes,
long maxContainerSize, StorageType storageType) throws IOException {
List<HddsVolume> candidates = volumes;
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java
index 727d7d14705..640d81df8ed 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java
@@ -146,13 +146,6 @@ public void setCheckChunksFilePath(boolean
bCheckChunksDirFilePath) {
this.bCheckChunksFilePath = bCheckChunksDirFilePath;
}
- @Override
- @Deprecated
- public void create(VolumeSet volumeSet, VolumeChoosingPolicy
- volumeChoosingPolicy, String clusterId) throws StorageContainerException
{
- create(volumeSet, volumeChoosingPolicy, clusterId, null);
- }
-
@Override
public void create(VolumeSet volumeSet, VolumeChoosingPolicy
volumeChoosingPolicy, String clusterId, StorageType storageType)
@@ -172,11 +165,7 @@ public void create(VolumeSet volumeSet,
VolumeChoosingPolicy
HddsVolume containerVolume;
String hddsVolumeDir;
try {
- // TODO Use the `chooseVolume` that supports `storageType`
- containerVolume = storageType == null
- ? volumeChoosingPolicy.chooseVolume(volumes, maxSize)
- : volumeChoosingPolicy.chooseVolume(
- volumes, maxSize, storageType);
+ containerVolume = volumeChoosingPolicy.chooseVolume(volumes,
maxSize, storageType);
hddsVolumeDir = containerVolume.getHddsRootDir().toString();
// Set volume before getContainerDBFile(), because we may need the
// volume to deduce the db file.
@@ -186,10 +175,12 @@ public void create(VolumeSet volumeSet,
VolumeChoosingPolicy
containerData.setCommittedSpace(true);
} catch (DiskOutOfSpaceException ex) {
throw new StorageContainerException("Container creation failed, " +
- "due to disk out of space", ex, DISK_OUT_OF_SPACE);
+ "due to disk out of space" + storageTypeMessage(storageType),
+ ex, DISK_OUT_OF_SPACE);
} catch (IOException ex) {
throw new StorageContainerException(
- "Container creation failed. " + ex.getMessage(), ex,
+ "Container creation failed" + storageTypeMessage(storageType) +
+ ". " + ex.getMessage(), ex,
CONTAINER_INTERNAL_ERROR);
}
@@ -353,6 +344,10 @@ private void updateContainerFile(File containerFile)
writeToContainerFile(containerFile, false);
}
+ private static String storageTypeMessage(StorageType storageType) {
+ return storageType == null ? "" : " on StorageType: " + storageType;
+ }
+
@Override
public void delete() throws StorageContainerException {
try {
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java
index 6f93bcef4fd..bf7346014f9 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java
@@ -94,8 +94,10 @@
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.HddsUtils;
import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.client.StorageTypeUtils;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.conf.StorageUnit;
@@ -469,6 +471,18 @@ ContainerCommandResponseProto handleCreateContainer(
}
newContainerData.setReplicaIndex(request.getCreateContainer()
.getReplicaIndex());
+ StorageType storageType = null;
+ if (request.getCreateContainer().hasStorageTypeID()) {
+ try {
+ storageType = StorageTypeUtils.getStorageTypeFromID(
+ request.getCreateContainer().getStorageTypeID());
+ } catch (IllegalArgumentException ex) {
+ return ContainerUtils.logAndReturnError(LOG,
+ new StorageContainerException(ex.getMessage(), ex,
+ INVALID_ARGUMENT), request);
+ }
+ }
+ newContainerData.setStorageType(storageType);
// TODO: Add support to add metadataList to ContainerData. Add metadata
// to container during creation.
@@ -480,7 +494,7 @@ ContainerCommandResponseProto handleCreateContainer(
containerIdLock.lock();
try {
if (containerSet.getContainer(containerID) == null) {
- newContainer.create(volumeSet, volumeChoosingPolicy, clusterId);
+ newContainer.create(volumeSet, volumeChoosingPolicy, clusterId,
storageType);
if (RECOVERING == newContainer.getContainerState()) {
created =
containerSet.addContainerByOverwriteMissingContainer(newContainer);
} else {
@@ -515,6 +529,7 @@ private void populateContainerPathFields(KeyValueContainer
container,
HddsVolume hddsVolume) throws IOException {
volumeSet.readLock();
try {
+ // TODO StoragePolicy Check whether need to adapt storageType
String idDir = VersionedDatanodeFeatures.ScmHA.chooseContainerPathID(
hddsVolume, clusterId);
container.populatePathFields(idDir, hddsVolume);
@@ -541,7 +556,6 @@ ContainerCommandResponseProto handleReadContainer(
request, containerData.getProtoBufMessage());
}
-
/**
* Handles Update Container Request. If successful, the container metadata
* is updated.
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerImporter.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerImporter.java
index 7b42006b229..682e119c553 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerImporter.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerImporter.java
@@ -149,9 +149,11 @@ private static void deleteFileQuietely(Path tarFilePath) {
HddsVolume chooseNextVolume(long spaceToReserve) throws IOException {
// Choose volume that can hold both container in tmp and dest directory
LOG.debug("Choosing volume to reserve space : {}", spaceToReserve);
+ // TODO: Use the target container storage type once replication/import
+ // requests carry it. Null preserves the existing any-volume behavior.
return volumeChoosingPolicy.chooseVolume(
StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()),
- spaceToReserve);
+ spaceToReserve, null);
}
public static Path getUntarDirectory(HddsVolume hddsVolume)
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/ContainerTestUtils.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/ContainerTestUtils.java
index bdef642204f..15c5884900f 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/ContainerTestUtils.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/ContainerTestUtils.java
@@ -36,6 +36,7 @@
import java.util.concurrent.atomic.AtomicLong;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.StorageUnit;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -384,7 +385,7 @@ public static KeyValueContainer addContainerToVolumeDir(
KeyValueContainer container =
new KeyValueContainer(keyValueContainerData, conf);
- container.create(volume.getVolumeSet(), volumeChoosingPolicy, clusterId);
+ container.create(volume.getVolumeSet(), volumeChoosingPolicy, clusterId,
StorageType.DISK);
container.close();
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestBlockDeletingService.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestBlockDeletingService.java
index 4f885961b27..996226c92bf 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestBlockDeletingService.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestBlockDeletingService.java
@@ -58,6 +58,7 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
@@ -189,7 +190,7 @@ private KeyValueContainerData
createToDeleteBlocks(ContainerSet containerSet,
data.setSchemaVersion(schemaVersion);
KeyValueContainer container = new KeyValueContainer(data, conf);
container.create(volumeSet,
- new RoundRobinVolumeChoosingPolicy(), scmId);
+ new RoundRobinVolumeChoosingPolicy(), scmId, StorageType.DISK);
containerSet.addContainer(container);
data = (KeyValueContainerData) containerSet.getContainer(
containerID).getContainerData();
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestSchemaTwoBackwardsCompatibility.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestSchemaTwoBackwardsCompatibility.java
index 1bdac646b73..a1b32ac15bd 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestSchemaTwoBackwardsCompatibility.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestSchemaTwoBackwardsCompatibility.java
@@ -37,6 +37,7 @@
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeoutException;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -272,7 +273,7 @@ private KeyValueContainer createTestContainer() throws
IOException {
cData.setSchemaVersion(OzoneConsts.SCHEMA_V2);
KeyValueContainer container = new KeyValueContainer(cData, conf);
container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(),
- clusterID);
+ clusterID, StorageType.DISK);
// populate with some blocks
// metadata will be updated here, too
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestStaleRecoveringContainerScrubbingService.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestStaleRecoveringContainerScrubbingService.java
index a21813956f5..a0064de6884 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestStaleRecoveringContainerScrubbingService.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestStaleRecoveringContainerScrubbingService.java
@@ -24,6 +24,7 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.Mockito.anyList;
import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -42,6 +43,7 @@
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.apache.hadoop.conf.StorageUnit;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
@@ -103,7 +105,7 @@ private void init() throws IOException {
volumeSet = mock(MutableVolumeSet.class);
volumeChoosingPolicy = mock(RoundRobinVolumeChoosingPolicy.class);
- when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong()))
+ when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong(),
eq(StorageType.DISK)))
.thenReturn(hddsVolume);
}
@@ -133,7 +135,7 @@ private List<Long> createTestContainers(
new KeyValueContainer(recoveringContainerData,
conf);
recoveringKeyValueContainer.create(
- volumeSet, volumeChoosingPolicy, clusterID);
+ volumeSet, volumeChoosingPolicy, clusterID, StorageType.DISK);
containerSet.addContainer(recoveringKeyValueContainer);
createdIds.add((long) containerIdNum);
}
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerPersistence.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerPersistence.java
index 0498d849961..c2c2f1b7e89 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerPersistence.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerPersistence.java
@@ -52,6 +52,7 @@
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
@@ -205,7 +206,7 @@ private KeyValueContainer addContainer(ContainerSet cSet,
long cID)
KeyValueContainer container = new KeyValueContainer(data, conf);
commitBytesBefore =
StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()).get(0).getCommittedBytes();
- container.create(volumeSet, volumeChoosingPolicy, SCM_ID);
+ container.create(volumeSet, volumeChoosingPolicy, SCM_ID,
StorageType.DISK);
cSet.addContainer(container);
commitBytesAfter = container.getContainerData()
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java
index da84d23d3e5..8dc29deb52a 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java
@@ -19,6 +19,7 @@
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.hadoop.hdds.fs.MockSpaceUsagePersistence.inMemory;
+import static
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.DISK_OUT_OF_SPACE;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.HDDS_DATANODE_DIR_KEY;
import static
org.apache.hadoop.hdds.scm.protocolPB.ContainerCommandResponseBuilders.getContainerCommandResponse;
import static
org.apache.hadoop.ozone.container.common.ContainerTestUtils.COMMIT_STAGE;
@@ -40,10 +41,13 @@
import com.google.common.collect.Maps;
import java.io.File;
import java.io.IOException;
+import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -51,6 +55,7 @@
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.hadoop.conf.StorageUnit;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.fs.MockSpaceUsageCheckFactory;
@@ -64,7 +69,9 @@
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerType;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.WriteChunkRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import
org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
import org.apache.hadoop.hdds.security.token.TokenVerifier;
import org.apache.hadoop.ozone.OzoneConfigKeys;
@@ -167,9 +174,9 @@ public void testContainerCloseActionWhenFull(
StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList())
.forEach(hddsVolume -> hddsVolume.setDbParentDir(tempDir.toFile()));
container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(),
- scmId.toString());
+ scmId.toString(), StorageType.DISK);
container2.create(volumeSet, new RoundRobinVolumeChoosingPolicy(),
- scmId.toString());
+ scmId.toString(), StorageType.DISK);
containerSet.addContainer(container);
containerSet.addContainer(container2);
ContainerMetrics metrics = ContainerMetrics.create(conf);
@@ -185,7 +192,7 @@ public void testContainerCloseActionWhenFull(
conf, containerSet, volumeSet, handlers, context, metrics, null);
hddsDispatcher.setClusterId(scmId.toString());
ContainerCommandResponseProto responseOne = hddsDispatcher
- .dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 1L), null);
+ .dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 1L, null),
null);
assertEquals(ContainerProtos.Result.SUCCESS,
responseOne.getResult());
verify(context, times(0))
@@ -196,9 +203,9 @@ public void testContainerCloseActionWhenFull(
containerData2.getStatistics().setBlockBytesForTesting(Double.valueOf(
StorageUnit.MB.toBytes(950)).longValue());
ContainerCommandResponseProto responseTwo = hddsDispatcher
- .dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 2L), null);
+ .dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 2L, null),
null);
ContainerCommandResponseProto responseThree = hddsDispatcher
- .dispatch(getWriteChunkRequest(dd.getUuidString(), 2L, 1L), null);
+ .dispatch(getWriteChunkRequest(dd.getUuidString(), 2L, 1L, null),
null);
assertEquals(ContainerProtos.Result.SUCCESS,
responseTwo.getResult());
assertEquals(ContainerProtos.Result.SUCCESS, responseThree.getResult());
@@ -211,7 +218,7 @@ public void testContainerCloseActionWhenFull(
// if we write again to container 1, the container action should get
added but heartbeat should not get triggered
// again because of throttling
- hddsDispatcher.dispatch(getWriteChunkRequest(dd.getUuidString(), 1L,
3L), null);
+ hddsDispatcher.dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 3L,
null), null);
verify(context,
times(3)).addContainerActionIfAbsent(any(ContainerAction.class));
verify(stateMachine, times(2)).triggerHeartbeat(); // was called twice
before
} finally {
@@ -326,7 +333,7 @@ public void testContainerCloseActionWhenVolumeFull(
StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList())
.forEach(hddsVolume -> hddsVolume.setDbParentDir(tempDir.toFile()));
container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(),
- scmId.toString());
+ scmId.toString(), StorageType.DISK);
containerSet.addContainer(container);
ContainerMetrics metrics = ContainerMetrics.create(conf);
Map<ContainerType, Handler> handlers = Maps.newHashMap();
@@ -343,15 +350,15 @@ public void testContainerCloseActionWhenVolumeFull(
containerData.getVolume().incrementUsedSpace(60);
usedSpace.addAndGet(60);
ContainerCommandResponseProto response = hddsDispatcher
- .dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 1L), null);
- assertEquals(ContainerProtos.Result.DISK_OUT_OF_SPACE,
response.getResult());
+ .dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 1L, null),
null);
+ assertEquals(DISK_OUT_OF_SPACE, response.getResult());
verify(context, times(1))
.addContainerActionIfAbsent(any(ContainerAction.class));
// verify that immediate heartbeat is triggered
verify(stateMachine, times(1)).triggerHeartbeat();
// the volume has reached the min free space boundary but this time the
heartbeat should not be triggered because
// of throttling
- hddsDispatcher.dispatch(getWriteChunkRequest(dd.getUuidString(), 1L,
2L), null);
+ hddsDispatcher.dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 2L,
null), null);
verify(context,
times(2)).addContainerActionIfAbsent(any(ContainerAction.class));
verify(stateMachine, times(1)).triggerHeartbeat(); // was called once
before
@@ -366,8 +373,8 @@ public void testContainerCloseActionWhenVolumeFull(
StorageContainerException scException =
assertThrows(StorageContainerException.class,
() -> container2.create(volumeSet,
- new RoundRobinVolumeChoosingPolicy(), scmId.toString()));
- assertEquals("Container creation failed, due to disk out of space",
+ new RoundRobinVolumeChoosingPolicy(), scmId.toString(),
StorageType.DISK));
+ assertEquals("Container creation failed, due to disk out of space on
StorageType: DISK",
scException.getMessage());
} finally {
volumeSet.shutdown();
@@ -386,7 +393,7 @@ public void testCreateContainerWithWriteChunk() throws
IOException {
DatanodeDetails dd = randomDatanodeDetails();
HddsDispatcher hddsDispatcher = createDispatcher(dd, scmId, conf);
ContainerCommandRequestProto writeChunkRequest =
- getWriteChunkRequest(dd.getUuidString(), 1L, 1L);
+ getWriteChunkRequest(dd.getUuidString(), 1L, 1L, null);
// send read chunk request and make sure container does not exist
ContainerCommandResponseProto response =
hddsDispatcher.dispatch(getReadChunkRequest(writeChunkRequest),
null);
@@ -428,6 +435,151 @@ public void testCreateContainerWithWriteChunk() throws
IOException {
}
}
+ @Test
+ public void testCreateContainerWithWriteChunkWithStorageType() throws
IOException {
+ int volumeNum = 2;
+ // Create two volume one is SSD volume another is DISK volume
+ StringBuilder hddsDirs = new StringBuilder();
+ File ssdVolume = Files.createTempDirectory(tempDir, "ssd").toFile();
+ File diskVolume = Files.createTempDirectory(tempDir, "disk").toFile();
+
hddsDirs.append("[SSD]").append(ssdVolume).append(',').append("[DISK]").append(diskVolume);
+ UUID scmId = UUID.randomUUID();
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.set(ScmConfigKeys.HDDS_DATANODE_DIR_KEY, hddsDirs.toString());
+ DatanodeDetails dd = randomDatanodeDetails();
+ HddsDispatcher dispatcher = createDispatcher(dd, scmId, conf);
+ long containerID = 1L;
+ long localId = 1L;
+
+ ContainerCommandRequestProto writeSSDChunkReq1 =
+ getWriteChunkRequest(dd.getUuidString(), containerID++, localId++,
HddsProtos.StorageTypeProto.SSD);
+ ContainerCommandRequestProto writeSSDChunkReq2 =
+ getWriteChunkRequest(dd.getUuidString(), containerID++, localId++,
HddsProtos.StorageTypeProto.SSD);
+ ContainerCommandRequestProto writeDISKChunkReq1 =
+ getWriteChunkRequest(dd.getUuidString(), containerID++, localId++,
HddsProtos.StorageTypeProto.DISK);
+ ContainerCommandRequestProto writeDISKChunkReq2 =
+ getWriteChunkRequest(dd.getUuidString(), containerID++, localId++,
HddsProtos.StorageTypeProto.DISK);
+
+ assertContainerDoNotExist(dispatcher, writeSSDChunkReq1);
+ assertContainerDoNotExist(dispatcher, writeSSDChunkReq2);
+ assertContainerDoNotExist(dispatcher, writeDISKChunkReq1);
+ assertContainerDoNotExist(dispatcher, writeDISKChunkReq2);
+
+ assertContainerCreateAtSpecificVolume(dispatcher, writeSSDChunkReq1,
Collections.singletonList(ssdVolume));
+ assertContainerCreateAtSpecificVolume(dispatcher, writeSSDChunkReq2,
Collections.singletonList(ssdVolume));
+ assertContainerCreateAtSpecificVolume(dispatcher, writeDISKChunkReq1,
Collections.singletonList(diskVolume));
+ assertContainerCreateAtSpecificVolume(dispatcher, writeDISKChunkReq2,
Collections.singletonList(diskVolume));
+
+ // Except the DISK and SSD volume, cannot create Container on other
StorageType Volume,
+ // because there is no other StorageType Volume
+ ContainerCommandRequestProto writeToRAMDISKChunkRequest =
+ getWriteChunkRequest(dd.getUuidString(), containerID++, localId++,
HddsProtos.StorageTypeProto.RAM_DISK);
+ assertContainerDoNotExist(dispatcher, writeToRAMDISKChunkRequest);
+ ContainerCommandResponseProto response =
dispatcher.dispatch(writeToRAMDISKChunkRequest, null);
+ assertEquals(DISK_OUT_OF_SPACE, response.getResult());
+
+ // If set the StorageType to null, then the Container can be created on
any StorageType Volume
+ ContainerCommandRequestProto writeAnyStorageTypeChunkRequest1 =
+ getWriteChunkRequest(dd.getUuidString(), containerID++, localId++,
null);
+ ContainerCommandRequestProto writeAnyStorageTypeChunkRequest2 =
+ getWriteChunkRequest(dd.getUuidString(), containerID++, localId++,
null);
+ ContainerCommandRequestProto writeAnyStorageTypeChunkRequest3 =
+ getWriteChunkRequest(dd.getUuidString(), containerID++, localId++,
null);
+ ContainerCommandRequestProto writeAnyStorageTypeChunkRequest4 =
+ getWriteChunkRequest(dd.getUuidString(), containerID++, localId++,
null);
+
+ assertContainerCreateAtSpecificVolume(dispatcher,
writeAnyStorageTypeChunkRequest1,
+ Arrays.asList(ssdVolume, diskVolume));
+ assertContainerCreateAtSpecificVolume(dispatcher,
writeAnyStorageTypeChunkRequest2,
+ Arrays.asList(ssdVolume, diskVolume));
+ assertContainerCreateAtSpecificVolume(dispatcher,
writeAnyStorageTypeChunkRequest3,
+ Arrays.asList(ssdVolume, diskVolume));
+ assertContainerCreateAtSpecificVolume(dispatcher,
writeAnyStorageTypeChunkRequest4,
+ Arrays.asList(ssdVolume, diskVolume));
+ }
+
+ @Test
+ public void testCreateContainerWithPutSmallFileWithStorageType() throws
IOException {
+ StringBuilder hddsDirs = new StringBuilder();
+ File ssdVolume = Files.createTempDirectory(tempDir, "ssd").toFile();
+ File diskVolume = Files.createTempDirectory(tempDir, "disk").toFile();
+
hddsDirs.append("[SSD]").append(ssdVolume).append(',').append("[DISK]").append(diskVolume);
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.set(ScmConfigKeys.HDDS_DATANODE_DIR_KEY, hddsDirs.toString());
+ HddsDispatcher dispatcher = createDispatcher(randomDatanodeDetails(),
UUID.randomUUID(), conf);
+ long containerID = 1L;
+ long localId = 1L;
+
+ ContainerCommandRequestProto putSmallFileToSSD =
+ newPutSmallFile(containerID++, localId++,
HddsProtos.StorageTypeProto.SSD);
+ ContainerCommandRequestProto putSmallFileToDISK =
+ newPutSmallFile(containerID++, localId++,
HddsProtos.StorageTypeProto.DISK);
+ ContainerCommandRequestProto putSmallFileToRAMDISK =
+ newPutSmallFile(containerID++, localId++,
HddsProtos.StorageTypeProto.RAM_DISK);
+
+ assertContainerDoNotExist(dispatcher, putSmallFileToSSD);
+ assertContainerDoNotExist(dispatcher, putSmallFileToDISK);
+ assertContainerDoNotExist(dispatcher, putSmallFileToRAMDISK);
+
+ assertContainerCreateAtSpecificVolume(dispatcher, putSmallFileToSSD,
+ Collections.singletonList(ssdVolume));
+ assertContainerCreateAtSpecificVolume(dispatcher, putSmallFileToDISK,
+ Collections.singletonList(diskVolume));
+
+ ContainerCommandResponseProto response =
dispatcher.dispatch(putSmallFileToRAMDISK, null);
+ assertEquals(DISK_OUT_OF_SPACE, response.getResult());
+ }
+
+ @Test
+ public void testCreateContainerRejectsInvalidStorageType() throws
IOException {
+ File diskVolume = Files.createTempDirectory(tempDir, "disk").toFile();
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.set(ScmConfigKeys.HDDS_DATANODE_DIR_KEY,
diskVolume.getAbsolutePath());
+ DatanodeDetails dd = randomDatanodeDetails();
+ HddsDispatcher dispatcher = createDispatcher(dd, UUID.randomUUID(), conf);
+
+ ContainerCommandRequestProto writeChunkRequest =
+ getWriteChunkRequest(dd.getUuidString(), 1L, 1L, null);
+ WriteChunkRequestProto writeChunk = writeChunkRequest.getWriteChunk();
+ ContainerCommandRequestProto requestWithInvalidStorageType =
+ writeChunkRequest.toBuilder()
+ .setWriteChunk(writeChunk.toBuilder()
+ .setBlockID(writeChunk.getBlockID().toBuilder()
+ .setStorageTypeID(999)))
+ .build();
+
+ ContainerCommandResponseProto response =
+ dispatcher.dispatch(requestWithInvalidStorageType, null);
+ assertEquals(ContainerProtos.Result.INVALID_ARGUMENT,
response.getResult());
+ }
+
+ private void assertContainerDoNotExist(HddsDispatcher hddsDispatcher,
+ ContainerCommandRequestProto request) {
+ ContainerCommandResponseProto response =
+ hddsDispatcher.dispatch(getReadContainerRequest(request), null);
+ assertEquals(response.getResult(),
+ ContainerProtos.Result.CONTAINER_NOT_FOUND);
+ }
+
+ private void assertContainerCreateAtSpecificVolume(HddsDispatcher
hddsDispatcher,
+ ContainerCommandRequestProto request, List<File> exceptionVolumePaths) {
+
+ // Send data command without sending create container.
+ ContainerCommandResponseProto response = hddsDispatcher.dispatch(request,
null);
+ // Container should be created as part of the data command.
+ assertEquals(ContainerProtos.Result.SUCCESS, response.getResult());
+
+ response = hddsDispatcher.dispatch(getReadContainerRequest(request), null);
+ assertEquals(ContainerProtos.Result.SUCCESS, response.getResult());
+
+ // Check if any of the exceptionVolumePaths match
+ String containerPath =
response.getReadContainer().getContainerData().getContainerPath();
+ boolean isMatched = exceptionVolumePaths.stream()
+ .anyMatch(path -> containerPath.startsWith(path.getAbsolutePath()));
+
+ assertTrue(isMatched);
+ }
+
@Test
public void testContainerNotFoundWithCommitChunk() throws IOException {
String testDirPath = testDir.getPath();
@@ -439,7 +591,7 @@ public void testContainerNotFoundWithCommitChunk() throws
IOException {
DatanodeDetails dd = randomDatanodeDetails();
HddsDispatcher hddsDispatcher = createDispatcher(dd, scmId, conf);
ContainerCommandRequestProto writeChunkRequest =
- getWriteChunkRequest(dd.getUuidString(), 1L, 1L);
+ getWriteChunkRequest(dd.getUuidString(), 1L, 1L, null);
// send read chunk request and make sure container does not exist
ContainerCommandResponseProto response =
@@ -473,12 +625,12 @@ public void testWriteChunkWithCreateContainerFailure()
throws IOException {
DatanodeDetails dd = randomDatanodeDetails();
HddsDispatcher hddsDispatcher = createDispatcher(dd, scmId, conf);
ContainerCommandRequestProto writeChunkRequest = getWriteChunkRequest(
- dd.getUuidString(), 1L, 1L);
+ dd.getUuidString(), 1L, 1L, null);
HddsDispatcher mockDispatcher = spy(hddsDispatcher);
ContainerCommandResponseProto.Builder builder =
getContainerCommandResponse(writeChunkRequest,
- ContainerProtos.Result.DISK_OUT_OF_SPACE, "");
+ DISK_OUT_OF_SPACE, "");
// Return DISK_OUT_OF_SPACE response when writing chunk
// with container creation.
doReturn(builder.build()).when(mockDispatcher)
@@ -509,7 +661,7 @@ public void
testCreateContainerWhenAlreadyExistsDoesNotMarkUnhealthy() throws IO
// Create container via WriteChunk
ContainerCommandRequestProto writeChunkRequest =
- getWriteChunkRequest(dd.getUuidString(), 1L, 1L);
+ getWriteChunkRequest(dd.getUuidString(), 1L, 1L, null);
ContainerCommandResponseProto initialResponse =
hddsDispatcher.dispatch(writeChunkRequest, null);
assertEquals(ContainerProtos.Result.SUCCESS,
initialResponse.getResult());
@@ -549,7 +701,7 @@ public void testDuplicateWriteChunkAndPutBlockRequest()
throws IOException {
DatanodeDetails dd = randomDatanodeDetails();
HddsDispatcher hddsDispatcher = createDispatcher(dd, scmId, conf);
ContainerCommandRequestProto writeChunkRequest = getWriteChunkRequest(
- dd.getUuidString(), 1L, 1L);
+ dd.getUuidString(), 1L, 1L, null);
//Send same WriteChunkRequest
ContainerCommandResponseProto response;
hddsDispatcher.dispatch(writeChunkRequest, null);
@@ -657,7 +809,7 @@ private static DatanodeDetails randomDatanodeDetails() {
}
private ContainerCommandRequestProto getWriteChunkRequest(
- String datanodeId, Long containerId, Long localId) {
+ String datanodeId, Long containerId, Long localId,
HddsProtos.StorageTypeProto storageType) {
ByteString data = ByteString.copyFrom(
UUID.randomUUID().toString().getBytes(UTF_8));
@@ -671,10 +823,16 @@ private ContainerCommandRequestProto getWriteChunkRequest(
.setChecksumData(Checksum.getNoChecksumDataProto())
.build();
+ ContainerProtos.DatanodeBlockID.Builder blockID =
+ new BlockID(containerId, localId).getDatanodeBlockIDProtobufBuilder();
+ // TODO: Pass the real storage type from the write path once that BlockID
support StorageType
+ if (storageType != null) {
+ blockID.setStorageTypeID(storageType.getNumber());
+ }
+
WriteChunkRequestProto.Builder writeChunkRequest = WriteChunkRequestProto
.newBuilder()
- .setBlockID(new BlockID(containerId, localId)
- .getDatanodeBlockIDProtobuf())
+ .setBlockID(blockID)
.setChunkData(chunk)
.setData(data);
@@ -728,15 +886,50 @@ private ContainerCommandRequestProto
getWriteChunkRequest0(
}
static ContainerCommandRequestProto newPutSmallFile(Long containerId, Long
localId) {
+ return newPutSmallFile(containerId, localId, null);
+ }
+
+ static ContainerCommandRequestProto newPutBlock(Long containerId, Long
localId,
+ HddsProtos.StorageTypeProto storageType) {
+ BlockID blockID = new BlockID(containerId, localId);
+ ContainerProtos.DatanodeBlockID.Builder datanodeBlockID =
+ blockID.getDatanodeBlockIDProtobuf().toBuilder();
+ if (storageType != null) {
+ datanodeBlockID.setStorageTypeID(storageType.getNumber());
+ }
+ ContainerProtos.BlockData.Builder blockData =
+ ContainerProtos.BlockData.newBuilder()
+ .setBlockID(datanodeBlockID);
+ return ContainerCommandRequestProto.newBuilder()
+ .setCmdType(ContainerProtos.Type.PutBlock)
+ .setContainerID(containerId)
+ .setDatanodeUuid(UUID.randomUUID().toString())
+ .setPutBlock(ContainerProtos.PutBlockRequestProto.newBuilder()
+ .setBlockData(blockData))
+ .build();
+ }
+
+ static ContainerCommandRequestProto newPutSmallFile(Long containerId, Long
localId,
+ HddsProtos.StorageTypeProto storageType) {
ByteString chunkData =
ByteString.copyFrom(RandomUtils.secure().randomBytes(32));
- return newPutSmallFile(new BlockID(containerId, localId), chunkData);
+ return newPutSmallFile(new BlockID(containerId, localId), chunkData,
storageType);
}
static ContainerCommandRequestProto newPutSmallFile(
BlockID blockID, ByteString data) {
+ return newPutSmallFile(blockID, data, null);
+ }
+
+ static ContainerCommandRequestProto newPutSmallFile(
+ BlockID blockID, ByteString data, HddsProtos.StorageTypeProto
storageType) {
+ ContainerProtos.DatanodeBlockID.Builder datanodeBlockID =
+ blockID.getDatanodeBlockIDProtobuf().toBuilder();
+ if (storageType != null) {
+ datanodeBlockID.setStorageTypeID(storageType.getNumber());
+ }
final ContainerProtos.BlockData.Builder blockData
= ContainerProtos.BlockData.newBuilder()
- .setBlockID(blockID.getDatanodeBlockIDProtobuf());
+ .setBlockID(datanodeBlockID);
final ContainerProtos.PutBlockRequestProto.Builder putBlockRequest
= ContainerProtos.PutBlockRequestProto.newBuilder()
.setBlockData(blockData);
@@ -789,6 +982,24 @@ private ContainerCommandRequestProto getReadChunkRequest(
.build();
}
+ /**
+ * Creates container read chunk request using input container write chunk
+ * request.
+ *
+ * @param writeChunkRequest - Input container write chunk request
+ * @return container read chunk request
+ */
+ private ContainerCommandRequestProto getReadContainerRequest(
+ ContainerCommandRequestProto request) {
+ return ContainerCommandRequestProto.newBuilder()
+ .setCmdType(ContainerProtos.Type.ReadContainer)
+ .setContainerID(request.getContainerID())
+ .setTraceID(request.getTraceID())
+ .setDatanodeUuid(request.getDatanodeUuid())
+
.setReadContainer(ContainerProtos.ReadContainerRequestProto.newBuilder())
+ .build();
+ }
+
@Test
public void testValidateToken() throws Exception {
try {
@@ -819,7 +1030,7 @@ public void verify(Token<?> token,
};
final ContainerCommandRequestProto request = getWriteChunkRequest(
- dd.getUuidString(), 1L, 1L);
+ dd.getUuidString(), 1L, 1L, HddsProtos.StorageTypeProto.DISK);
final HddsDispatcher dispatcher = createDispatcher(
dd, scmId, conf, tokenVerifier);
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestCapacityVolumeChoosingPolicy.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestCapacityVolumeChoosingPolicy.java
index deae4f83951..161aef6cefa 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestCapacityVolumeChoosingPolicy.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestCapacityVolumeChoosingPolicy.java
@@ -29,6 +29,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.fs.MockSpaceUsageCheckFactory;
import org.apache.hadoop.hdds.fs.MockSpaceUsageSource;
@@ -113,7 +114,7 @@ public void testCapacityVolumeChoosingPolicy() throws
Exception {
// Test 1000 rounds of volume choosing
for (int i = 0; i < 1000; i++) {
- HddsVolume volume = policy.chooseVolume(volumes, 0);
+ HddsVolume volume = policy.chooseVolume(volumes, 0, StorageType.DISK);
chooseCount.put(volume, chooseCount.get(volume) + 1);
}
@@ -124,7 +125,7 @@ public void testCapacityVolumeChoosingPolicy() throws
Exception {
@Test
public void throwsDiskOutOfSpaceIfRequestMoreThanAvailable() {
Exception e = assertThrows(DiskOutOfSpaceException.class,
- () -> policy.chooseVolume(volumes, 500));
+ () -> policy.chooseVolume(volumes, 500, StorageType.DISK));
String msg = e.getMessage();
assertThat(msg)
@@ -155,7 +156,7 @@ public void testVolumeCommittedSpace() throws Exception {
volumes.forEach(vol ->
initialCommittedSpace.put(vol, vol.getCommittedBytes()));
- HddsVolume selectedVolume = policy.chooseVolume(volumes, 50);
+ HddsVolume selectedVolume = policy.chooseVolume(volumes, 50,
StorageType.DISK);
assertEquals(initialCommittedSpace.get(selectedVolume) + 50,
selectedVolume.getCommittedBytes());
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestRoundRobinVolumeChoosingPolicy.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestRoundRobinVolumeChoosingPolicy.java
index 36fabff1fe8..e101b43eb41 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestRoundRobinVolumeChoosingPolicy.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestRoundRobinVolumeChoosingPolicy.java
@@ -28,6 +28,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.fs.MockSpaceUsageCheckFactory;
import org.apache.hadoop.hdds.fs.MockSpaceUsageSource;
@@ -96,21 +97,21 @@ public void testRRVolumeChoosingPolicy() throws Exception {
assertEquals(200L, hddsVolume2.getCurrentUsage().getAvailable());
// Test two rounds of round-robin choosing
- assertEquals(hddsVolume1, policy.chooseVolume(volumes, 0));
- assertEquals(hddsVolume2, policy.chooseVolume(volumes, 0));
- assertEquals(hddsVolume1, policy.chooseVolume(volumes, 0));
- assertEquals(hddsVolume2, policy.chooseVolume(volumes, 0));
+ assertEquals(hddsVolume1, policy.chooseVolume(volumes, 0,
StorageType.DISK));
+ assertEquals(hddsVolume2, policy.chooseVolume(volumes, 0,
StorageType.DISK));
+ assertEquals(hddsVolume1, policy.chooseVolume(volumes, 0,
StorageType.DISK));
+ assertEquals(hddsVolume2, policy.chooseVolume(volumes, 0,
StorageType.DISK));
// The first volume has only 100L space, so the policy should
// choose the second one in case we ask for more.
assertEquals(hddsVolume2,
- policy.chooseVolume(volumes, 120));
+ policy.chooseVolume(volumes, 120, StorageType.DISK));
}
@Test
public void throwsDiskOutOfSpaceIfRequestMoreThanAvailable() {
Exception e = assertThrows(DiskOutOfSpaceException.class,
- () -> policy.chooseVolume(volumes, 300));
+ () -> policy.chooseVolume(volumes, 300, StorageType.DISK));
String msg = e.getMessage();
assertThat(msg).contains("No volumes have enough space for a new
container. " +
@@ -123,7 +124,7 @@ public void testVolumeCommittedSpace() throws Exception {
volumes.forEach(vol ->
initialCommittedSpace.put(vol, vol.getCommittedBytes()));
- HddsVolume selectedVolume = policy.chooseVolume(volumes, 50);
+ HddsVolume selectedVolume = policy.chooseVolume(volumes, 50,
StorageType.DISK);
assertEquals(initialCommittedSpace.get(selectedVolume) + 50,
selectedVolume.getCommittedBytes());
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestVolumeSetDiskChecks.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestVolumeSetDiskChecks.java
index e310eaf3dea..02aa6d379f7 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestVolumeSetDiskChecks.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestVolumeSetDiskChecks.java
@@ -42,6 +42,7 @@
import java.util.Set;
import java.util.UUID;
import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -301,14 +302,14 @@ public void testVolumeFailure() throws IOException {
StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList())
.forEach(hddsVolume -> hddsVolume.setDbParentDir(tempDir.toFile()));
container.create(volumeSet,
- new RoundRobinVolumeChoosingPolicy(), UUID.randomUUID().toString());
+ new RoundRobinVolumeChoosingPolicy(), UUID.randomUUID().toString(),
StorageType.DISK);
conSet.addContainer(container);
KeyValueContainer container1 = new KeyValueContainer(data1, conf);
StorageVolumeUtil.getHddsVolumesList(volumeSet1.getVolumesList())
.forEach(hddsVolume -> hddsVolume.setDbParentDir(tempDir.toFile()));
container1.create(volumeSet1,
- new RoundRobinVolumeChoosingPolicy(), UUID.randomUUID().toString());
+ new RoundRobinVolumeChoosingPolicy(), UUID.randomUUID().toString(),
StorageType.DISK);
conSet.addContainer(container1);
DatanodeStateMachine datanodeStateMachineMock =
mock(DatanodeStateMachine.class);
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/diskbalancer/TestDiskBalancerTask.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/diskbalancer/TestDiskBalancerTask.java
index dc2cd03655d..d1ff42b4906 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/diskbalancer/TestDiskBalancerTask.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/diskbalancer/TestDiskBalancerTask.java
@@ -49,6 +49,7 @@
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.fs.MockSpaceUsageCheckFactory;
@@ -686,9 +687,9 @@ private KeyValueContainer createContainer(long containerId,
HddsVolume vol, Stat
KeyValueContainer container = new KeyValueContainer(containerData, conf);
VolumeChoosingPolicy policy = mock(VolumeChoosingPolicy.class);
- when(policy.chooseVolume(any(List.class), any(Long.class)))
+ when(policy.chooseVolume(any(List.class), any(Long.class),
any(StorageType.class)))
.thenReturn(vol);
- container.create((VolumeSet) volumeSet, policy, scmId);
+ container.create((VolumeSet) volumeSet, policy, scmId, StorageType.DISK);
containerSet.addContainer(container);
// Manually update volume usage for test purposes
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueBlockIterator.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueBlockIterator.java
index 4d24f199581..d2cc55e55b4 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueBlockIterator.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueBlockIterator.java
@@ -35,6 +35,7 @@
import java.util.NoSuchElementException;
import java.util.UUID;
import org.apache.hadoop.conf.StorageUnit;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
@@ -116,7 +117,7 @@ public void setup() throws Exception {
// Init the container.
KeyValueContainer container = new KeyValueContainer(containerData, conf);
container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(),
- clusterID);
+ clusterID, StorageType.DISK);
db = BlockUtils.getDB(containerData, conf);
}
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainer.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainer.java
index 7c14200e6c8..137b677ad0d 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainer.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainer.java
@@ -161,7 +161,7 @@ private void init(ContainerTestVersionInfo versionInfo)
throws Exception {
.map(v -> (StorageVolume) v)
.collect(Collectors.toList()));
when(volumeChoosingPolicy
- .chooseVolume(anyList(), anyLong())).thenAnswer(
+ .chooseVolume(anyList(), anyLong(), eq(StorageType.DISK))).thenAnswer(
invocation -> {
List<HddsVolume> volumes = invocation.getArgument(0);
return volumes.get(0);
@@ -458,7 +458,7 @@ public void
testContainerImportExport(ContainerTestVersionInfo versionInfo)
KeyValueContainer container = new KeyValueContainer(containerData, CONF);
HddsVolume containerVolume = volumeChoosingPolicy.chooseVolume(
- StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), 1);
+ StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), 1,
StorageType.DISK);
container.populatePathFields(scmId, containerVolume);
try (InputStream fis = Files.newInputStream(folderToExport.toPath())) {
@@ -504,7 +504,7 @@ public void
testContainerImportExport(ContainerTestVersionInfo versionInfo)
container = new KeyValueContainer(containerData, CONF);
containerVolume = volumeChoosingPolicy.chooseVolume(
- StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), 1);
+ StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), 1,
StorageType.DISK);
container.populatePathFields(scmId, containerVolume);
KeyValueContainer finalContainer1 = container;
assertThrows(IOException.class, () -> {
@@ -537,7 +537,7 @@ private void
checkContainerFilesPresent(KeyValueContainerData data,
* Create the container on disk.
*/
private void createContainer() throws StorageContainerException {
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK);
keyValueContainerData = keyValueContainer.getContainerData();
}
@@ -635,9 +635,9 @@ public void concurrentExport(ContainerTestVersionInfo
versionInfo)
public void testDuplicateContainer(ContainerTestVersionInfo versionInfo)
throws Exception {
init(versionInfo);
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK);
StorageContainerException exception =
assertThrows(StorageContainerException.class, () ->
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId));
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK));
assertEquals(ContainerProtos.Result.CONTAINER_ALREADY_EXISTS,
exception.getResult());
assertThat(exception).hasMessage("Container creation failed because
ContainerFile already exists");
}
@@ -647,13 +647,13 @@ public void testDiskFullExceptionCreateContainer(
ContainerTestVersionInfo versionInfo) throws Exception {
init(versionInfo);
reset(volumeChoosingPolicy);
- when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong()))
+ when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong(),
eq(StorageType.DISK)))
.thenThrow(DiskChecker.DiskOutOfSpaceException.class);
StorageContainerException exception =
assertThrows(StorageContainerException.class, () ->
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId));
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK));
assertEquals(ContainerProtos.Result.DISK_OUT_OF_SPACE,
exception.getResult());
- assertThat(exception).hasMessage("Container creation failed, due to disk
out of space");
+ assertThat(exception).hasMessage("Container creation failed, due to disk
out of space on StorageType: DISK");
}
@ContainerTestVersionInfo.ContainerTest
@@ -663,7 +663,7 @@ public void testDeleteContainer(ContainerTestVersionInfo
versionInfo)
closeContainer();
keyValueContainer = new KeyValueContainer(
keyValueContainerData, CONF);
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK);
KeyValueContainerUtil.removeContainer(
keyValueContainer.getContainerData(), CONF);
keyValueContainer.delete();
@@ -690,7 +690,7 @@ public void testDeleteContainer(ContainerTestVersionInfo
versionInfo)
public void testCloseContainer(ContainerTestVersionInfo versionInfo)
throws Exception {
init(versionInfo);
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK);
keyValueContainer.close();
keyValueContainerData = keyValueContainer
@@ -712,7 +712,7 @@ public void testCloseContainer(ContainerTestVersionInfo
versionInfo)
public void testReportOfUnhealthyContainer(
ContainerTestVersionInfo versionInfo) throws Exception {
init(versionInfo);
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK);
assertNotNull(keyValueContainer.getContainerReport());
keyValueContainer.markContainerUnhealthy();
File containerFile = keyValueContainer.getContainerFile();
@@ -727,7 +727,7 @@ public void testReportOfUnhealthyContainer(
public void testUpdateContainer(ContainerTestVersionInfo versionInfo)
throws Exception {
init(versionInfo);
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK);
Map<String, String> metadata = new HashMap<>();
metadata.put(OzoneConsts.VOLUME, OzoneConsts.OZONE);
metadata.put(OzoneConsts.OWNER, OzoneConsts.OZONE_SIMPLE_HDFS_USER);
@@ -756,7 +756,7 @@ public void testUpdateContainerUnsupportedRequest(
StorageContainerException exception =
assertThrows(StorageContainerException.class, () -> {
keyValueContainer = new KeyValueContainer(keyValueContainerData, CONF);
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK);
Map<String, String> metadata = new HashMap<>();
metadata.put(OzoneConsts.VOLUME, OzoneConsts.OZONE);
keyValueContainer.update(metadata, false);
@@ -773,7 +773,7 @@ public void testContainerRocksDB(ContainerTestVersionInfo
versionInfo)
closeContainer();
keyValueContainer = new KeyValueContainer(
keyValueContainerData, CONF);
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK);
try (DBHandle db = BlockUtils.getDB(keyValueContainerData, CONF)) {
RDBStore store = (RDBStore) db.getStore().getStore();
@@ -827,7 +827,7 @@ public void testDBProfileAffectsDBOptions(
ContainerTestVersionInfo versionInfo) throws Exception {
init(versionInfo);
// Create Container 1
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK);
DatanodeDBProfile outProfile1;
try (DBHandle db1 =
@@ -848,7 +848,7 @@ public void testDBProfileAffectsDBOptions(
(long) StorageUnit.GB.toBytes(5), UUID.randomUUID().toString(),
datanodeId.toString());
keyValueContainer = new KeyValueContainer(keyValueContainerData,
otherConf);
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK);
DatanodeDBProfile outProfile2;
try (DBHandle db2 =
@@ -928,7 +928,7 @@ void testAutoCompactionSmallSstFile(
List<File> exportFiles = new ArrayList<>();
for (HddsVolume volume: volumeList) {
reset(volumeChoosingPolicy);
- when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong()))
+ when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong(),
eq(StorageType.DISK)))
.thenReturn(volume);
for (int index = 0; index < count; index++, containerId++) {
// Create new container
@@ -936,7 +936,7 @@ void testAutoCompactionSmallSstFile(
(long) StorageUnit.GB.toBytes(5), UUID.randomUUID().toString(),
datanodeId.toString());
container = new KeyValueContainer(containerData, CONF);
- container.create(volumeSet, volumeChoosingPolicy, scmId);
+ container.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK);
containerData = container.getContainerData();
containerData.setState(ContainerProtos.ContainerDataProto.State.CLOSED);
populate(container, numberOfKeysToWrite);
@@ -1043,7 +1043,7 @@ public void testIsEmptyContainerStateWhileImport(
KeyValueContainer container = new KeyValueContainer(containerData, CONF);
HddsVolume containerVolume = volumeChoosingPolicy.chooseVolume(
- StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), 1);
+ StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), 1,
StorageType.DISK);
container.populatePathFields(scmId, containerVolume);
try (InputStream fis = Files.newInputStream(folderToExport.toPath())) {
@@ -1091,7 +1091,7 @@ public void
testIsEmptyContainerStateWhileImportWithoutBlock(
KeyValueContainer container = new KeyValueContainer(containerData, CONF);
HddsVolume containerVolume = volumeChoosingPolicy.chooseVolume(
- StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), 1);
+ StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), 1,
StorageType.DISK);
container.populatePathFields(scmId, containerVolume);
try (InputStream fis = Files.newInputStream(folderToExport.toPath())) {
@@ -1146,7 +1146,7 @@ private void testMixedSchemaImport(String dir,
ContainerTestHelper.CONTAINER_MAX_SIZE, UUID.randomUUID().toString(),
UUID.randomUUID().toString());
KeyValueContainer container = new KeyValueContainer(data, conf);
- container.create(volumeSet, volumeChoosingPolicy, scmId);
+ container.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK);
long pendingDeleteBlockCount = 20;
try (DBHandle meta = BlockUtils.getDB(data, conf)) {
Table<String, Long> metadataTable = meta.getStore().getMetadataTable();
@@ -1208,10 +1208,10 @@ public void testContainerCreationCommitSpaceReserve(
keyValueContainer = new KeyValueContainer(keyValueContainerData, CONF);
keyValueContainer = spy(keyValueContainer);
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK);
// verify that
- verify(volumeChoosingPolicy).chooseVolume(anyList(), anyLong()); // this
would reserve commit space
+ verify(volumeChoosingPolicy).chooseVolume(anyList(), anyLong(),
eq(StorageType.DISK));
assertTrue(keyValueContainerData.isCommittedSpace());
}
}
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerIntegrityChecks.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerIntegrityChecks.java
index adc4b35ba1a..a1b44f4a44c 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerIntegrityChecks.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerIntegrityChecks.java
@@ -30,6 +30,7 @@
import java.util.List;
import java.util.UUID;
import org.apache.commons.lang3.RandomStringUtils;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
@@ -132,7 +133,7 @@ protected KeyValueContainer createContainerWithBlocks(long
containerId,
UUID.randomUUID().toString(), UUID.randomUUID().toString());
KeyValueContainer container = new KeyValueContainer(containerData, conf);
container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(),
- clusterID);
+ clusterID, StorageType.DISK);
try (DBHandle metadataStore = BlockUtils.getDB(containerData,
conf)) {
assertNotNull(containerData.getChunksPath());
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerMarkUnhealthy.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerMarkUnhealthy.java
index 3606f80d13f..77aa70b8fd7 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerMarkUnhealthy.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerMarkUnhealthy.java
@@ -23,6 +23,7 @@
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.anyList;
import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -32,6 +33,7 @@
import java.nio.file.Path;
import java.util.UUID;
import org.apache.hadoop.conf.StorageUnit;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import
org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
import org.apache.hadoop.ozone.container.common.impl.ContainerDataYaml;
@@ -79,7 +81,7 @@ public void setup() throws Exception {
volumeSet = mock(MutableVolumeSet.class);
volumeChoosingPolicy = mock(RoundRobinVolumeChoosingPolicy.class);
- when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong()))
+ when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong(),
eq(StorageType.DISK)))
.thenReturn(hddsVolume);
keyValueContainerData = new KeyValueContainerData(1L,
@@ -145,7 +147,7 @@ public void
testMarkClosedContainerAsUnhealthy(ContainerLayoutVersion layoutVers
initTestData(layoutVersion);
// We need to create the container so the compact-on-close operation
// does not NPE.
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK);
keyValueContainer.close();
keyValueContainer.markContainerUnhealthy();
assertThat(keyValueContainerData.getState()).isEqualTo(UNHEALTHY);
@@ -159,7 +161,7 @@ public void
testMarkQuasiClosedContainerAsUnhealthy(ContainerLayoutVersion layou
initTestData(layoutVersion);
// We need to create the container so the sync-on-quasi-close operation
// does not NPE.
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK);
keyValueContainer.quasiClose();
keyValueContainer.markContainerUnhealthy();
assertThat(keyValueContainerData.getState()).isEqualTo(UNHEALTHY);
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java
index 0385564ebf7..7eb3166df4b 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java
@@ -72,6 +72,7 @@
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.StorageUnit;
import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -876,7 +877,7 @@ public void testICRsOnContainerClose(ContainerLayoutVersion
containerLayoutVersi
// markContainerForClose - OPEN -> CLOSING (should send immediate ICR)
containerData.setState(ContainerProtos.ContainerDataProto.State.OPEN);
KeyValueContainer container = new KeyValueContainer(containerData, conf);
- container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(),
CLUSTER_ID);
+ container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(),
CLUSTER_ID, StorageType.DISK);
containerSet.addContainer(container);
kvHandler.markContainerForClose(container);
@@ -1023,7 +1024,7 @@ public void testReadBlockMetrics() throws Exception {
(long) StorageUnit.GB.toBytes(1), UUID.randomUUID().toString(),
DATANODE_UUID);
KeyValueContainer container = new KeyValueContainer(containerData, conf);
- container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(),
CLUSTER_ID);
+ container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(),
CLUSTER_ID, StorageType.DISK);
containerSet.addContainer(container);
BlockID blockID = ContainerTestHelper.getTestBlockID(containerID);
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/AbstractTestChunkManager.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/AbstractTestChunkManager.java
index 6995b75f530..1e09faea8d9 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/AbstractTestChunkManager.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/AbstractTestChunkManager.java
@@ -24,6 +24,7 @@
import static org.junit.jupiter.api.Assumptions.abort;
import static org.mockito.Mockito.anyList;
import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -35,6 +36,7 @@
import java.nio.ByteBuffer;
import java.util.UUID;
import org.apache.hadoop.conf.StorageUnit;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.ozone.container.common.helpers.ChunkInfo;
@@ -93,7 +95,7 @@ public final void setUp(@TempDir File confDir) throws
Exception {
RoundRobinVolumeChoosingPolicy volumeChoosingPolicy =
mock(RoundRobinVolumeChoosingPolicy.class);
- when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong()))
+ when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong(),
eq(StorageType.DISK)))
.thenReturn(hddsVolume);
keyValueContainerData = new KeyValueContainerData(1L,
@@ -104,7 +106,7 @@ public final void setUp(@TempDir File confDir) throws
Exception {
keyValueContainer = new KeyValueContainer(keyValueContainerData, config);
keyValueContainer.create(volumeSet, volumeChoosingPolicy,
- UUID.randomUUID().toString());
+ UUID.randomUUID().toString(), StorageType.DISK);
header = "my header".getBytes(UTF_8);
byte[] bytes = "testing write chunks".getBytes(UTF_8);
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestBlockManagerImpl.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestBlockManagerImpl.java
index eb58ac3ce84..43d012c068c 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestBlockManagerImpl.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestBlockManagerImpl.java
@@ -26,6 +26,7 @@
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -35,6 +36,7 @@
import java.util.List;
import java.util.UUID;
import org.apache.hadoop.conf.StorageUnit;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
@@ -95,7 +97,7 @@ private void initialize() throws Exception {
VolumeSet volumeSet = mock(MutableVolumeSet.class);
RoundRobinVolumeChoosingPolicy volumeChoosingPolicy =
mock(RoundRobinVolumeChoosingPolicy.class);
- when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong()))
+ when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong(),
eq(StorageType.DISK)))
.thenReturn(hddsVolume);
KeyValueContainerData keyValueContainerData = new KeyValueContainerData(1L,
@@ -106,7 +108,7 @@ private void initialize() throws Exception {
keyValueContainer = new KeyValueContainer(
keyValueContainerData, config);
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK);
// Creating BlockData
BlockID blockID = new BlockID(1L, 1L);
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestContainerReader.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestContainerReader.java
index 234fda2cd00..4b796376fd0 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestContainerReader.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestContainerReader.java
@@ -33,6 +33,7 @@
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.anyList;
import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -45,6 +46,7 @@
import java.util.List;
import java.util.UUID;
import org.apache.hadoop.conf.StorageUnit;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
@@ -130,7 +132,7 @@ private void setup(ContainerTestVersionInfo versionInfo)
throws Exception {
volumeSet = mock(MutableVolumeSet.class);
volumeChoosingPolicy = mock(RoundRobinVolumeChoosingPolicy.class);
- when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong()))
+ when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong(),
eq(StorageType.DISK)))
.thenReturn(hddsVolume);
for (int i = 0; i < 2; i++) {
@@ -141,7 +143,7 @@ private void setup(ContainerTestVersionInfo versionInfo)
throws Exception {
KeyValueContainer keyValueContainer =
new KeyValueContainer(keyValueContainerData,
conf);
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, clusterId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, clusterId,
StorageType.DISK);
List<Long> blkNames;
@@ -299,7 +301,7 @@ public void testContainerReader(ContainerTestVersionInfo
versionInfo)
new KeyValueContainer(recoveringContainerData,
conf);
recoveringKeyValueContainer.create(
- volumeSet, volumeChoosingPolicy, clusterId);
+ volumeSet, volumeChoosingPolicy, clusterId, StorageType.DISK);
thread = new Thread(containerReader);
thread.start();
@@ -359,7 +361,7 @@ public void testContainerReaderWithLoadException(
StorageVolumeUtil.checkVolume(hddsVolume1, clusterId, clusterId, conf,
null, null);
volumeChoosingPolicy1 = mock(RoundRobinVolumeChoosingPolicy.class);
- when(volumeChoosingPolicy1.chooseVolume(anyList(), anyLong()))
+ when(volumeChoosingPolicy1.chooseVolume(anyList(), anyLong(),
eq(StorageType.DISK)))
.thenReturn(hddsVolume1);
int containerCount = 3;
@@ -370,7 +372,7 @@ public void testContainerReaderWithLoadException(
datanodeId.toString());
KeyValueContainer keyValueContainer =
new KeyValueContainer(keyValueContainerData, conf);
- keyValueContainer.create(volumeSet1, volumeChoosingPolicy1, clusterId);
+ keyValueContainer.create(volumeSet1, volumeChoosingPolicy1, clusterId,
StorageType.DISK);
if (i == 0) {
// rename first container directory name
@@ -417,7 +419,7 @@ public void testContainerReaderWithInvalidDbPath(
StorageVolumeUtil.checkVolume(hddsVolume1, clusterId, clusterId, conf,
null, null);
volumeChoosingPolicy1 = mock(RoundRobinVolumeChoosingPolicy.class);
- when(volumeChoosingPolicy1.chooseVolume(anyList(), anyLong()))
+ when(volumeChoosingPolicy1.chooseVolume(anyList(), anyLong(),
eq(StorageType.DISK)))
.thenReturn(hddsVolume1);
List<File> dbPathList = new ArrayList<>();
@@ -429,7 +431,7 @@ public void testContainerReaderWithInvalidDbPath(
datanodeId.toString());
KeyValueContainer keyValueContainer =
new KeyValueContainer(keyValueContainerData, conf);
- keyValueContainer.create(volumeSet1, volumeChoosingPolicy1, clusterId);
+ keyValueContainer.create(volumeSet1, volumeChoosingPolicy1, clusterId,
StorageType.DISK);
dbPathList.add(keyValueContainerData.getDbFile());
}
ContainerCache.getInstance(conf).shutdownCache();
@@ -642,7 +644,7 @@ private KeyValueContainer createContainerWithId(int id,
VolumeSet volSet,
KeyValueContainer keyValueContainer =
new KeyValueContainer(keyValueContainerData,
conf);
- keyValueContainer.create(volSet, policy, clusterId);
+ keyValueContainer.create(volSet, policy, clusterId, StorageType.DISK);
List<Long> blkNames;
if (id % 2 == 0) {
@@ -680,7 +682,7 @@ public void testMarkedDeletedContainerCleared(
KeyValueContainer kvContainer =
new KeyValueContainer(containerData, conf);
kvContainer.create(
- volumeSet, volumeChoosingPolicy, clusterId);
+ volumeSet, volumeChoosingPolicy, clusterId, StorageType.DISK);
long baseCount = 0;
if (containerData.hasSchema(OzoneConsts.SCHEMA_V3)) {
// add db entry for the container ID 101 for V3
@@ -849,7 +851,7 @@ private KeyValueContainer createContainer(long containerId)
throws Exception {
(long) StorageUnit.GB.toBytes(5), UUID.randomUUID().toString(),
datanodeId.toString());
containerData.setState(ContainerProtos.ContainerDataProto.State.CLOSED);
KeyValueContainer container = new KeyValueContainer(containerData, conf);
- container.create(volumeSet, volumeChoosingPolicy, clusterId);
+ container.create(volumeSet, volumeChoosingPolicy, clusterId,
StorageType.DISK);
return container;
}
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainer.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainer.java
index 91c3f8ed58c..ecac604cde6 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainer.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainer.java
@@ -41,6 +41,7 @@
import java.util.concurrent.ConcurrentSkipListSet;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.StorageUnit;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -168,7 +169,7 @@ public void testBuildContainerMap(ContainerTestVersionInfo
versionInfo)
containerDatas.add(keyValueContainerData);
keyValueContainer = new KeyValueContainer(
keyValueContainerData, conf);
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, clusterId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, clusterId,
StorageType.DISK);
myVolume = keyValueContainer.getContainerData().getVolume();
// Track container in mock volume
@@ -286,7 +287,7 @@ public void
testContainerCreateDiskFull(ContainerTestVersionInfo versionInfo)
StorageContainerException e = assertThrows(
StorageContainerException.class,
() -> keyValueContainer.
- create(volumeSet, volumeChoosingPolicy, clusterId)
+ create(volumeSet, volumeChoosingPolicy, clusterId,
StorageType.DISK)
);
assertEquals(DISK_OUT_OF_SPACE, e.getResult());
}
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerImporter.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerImporter.java
index e594dad3e58..553fb267882 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerImporter.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerImporter.java
@@ -23,6 +23,8 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.anyLong;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.doNothing;
@@ -234,6 +236,19 @@ public void testImportContainerResetsLastScanTime() throws
Exception {
assertEquals(Optional.empty(), containerData.lastDataScanTime());
}
+ @Test
+ public void testChooseNextVolumeStorageType() throws Exception {
+ VolumeChoosingPolicy policy = mock(VolumeChoosingPolicy.class);
+ HddsVolume expectedVolume = mock(HddsVolume.class);
+ long spaceToReserve = 100L;
+ when(policy.chooseVolume(anyList(), anyLong(), isNull()))
+ .thenReturn(expectedVolume);
+ ContainerImporter importer = new ContainerImporter(conf, containerSet,
+ controllerMock, volumeSet, policy);
+
+ assertEquals(expectedVolume, importer.chooseNextVolume(spaceToReserve));
+ }
+
private File containerTarFile(long id, ContainerData data) throws
IOException {
File yamlFile = new File(tempDir, "container.yaml");
ContainerDataYaml.createContainerFile(data, yamlFile);
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestGrpcReplicationService.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestGrpcReplicationService.java
index 8b831fa0646..0fc8b1a5554 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestGrpcReplicationService.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestGrpcReplicationService.java
@@ -40,6 +40,7 @@
import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;
+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.datanode.proto.ContainerProtos;
@@ -139,7 +140,7 @@ public void init() throws Exception {
StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList())
.forEach(hddsVolume -> hddsVolume.setDbParentDir(tempDir.toFile()));
container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(),
- "test-replication");
+ "test-replication", StorageType.DISK);
containerSet.addContainer(container);
container.close();
diff --git
a/hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto
b/hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto
index 05c94624c99..faff319310f 100644
--- a/hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto
+++ b/hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto
@@ -150,6 +150,8 @@ message DatanodeBlockID {
optional uint64 blockCommitSequenceId = 3 [default = 0];
optional int32 replicaIndex = 4;
+ optional int32 storageTypeID = 5;
+
}
message KeyValue {
@@ -271,6 +273,10 @@ message CreateContainerRequestProto {
optional ContainerType containerType = 3 [default = KeyValueContainer];
optional int32 replicaIndex = 4;
optional ContainerDataProto.State state = 5;
+
+ // If we do not set storageTypeID, it means that we allow Containers to be
+ // created on Volumes of any StorageType.
+ optional int32 storageTypeID = 6;
}
message CreateContainerResponseProto {
diff --git
a/hadoop-ozone/cli-repair/src/test/java/org/apache/hadoop/ozone/repair/datanode/schemaupgrade/TestUpgradeContainerSchema.java
b/hadoop-ozone/cli-repair/src/test/java/org/apache/hadoop/ozone/repair/datanode/schemaupgrade/TestUpgradeContainerSchema.java
index f8983e23a9e..e51e50ffe3b 100644
---
a/hadoop-ozone/cli-repair/src/test/java/org/apache/hadoop/ozone/repair/datanode/schemaupgrade/TestUpgradeContainerSchema.java
+++
b/hadoop-ozone/cli-repair/src/test/java/org/apache/hadoop/ozone/repair/datanode/schemaupgrade/TestUpgradeContainerSchema.java
@@ -36,6 +36,7 @@
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.anyList;
import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -53,6 +54,7 @@
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.cli.GenericCli;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -159,7 +161,7 @@ private void initDatanode(HDDSLayoutFeature layoutFeature)
throws IOException {
volumeChoosingPolicy = mock(RoundRobinVolumeChoosingPolicy.class);
final AtomicInteger loopCount = new AtomicInteger(0);
- when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong()))
+ when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong(),
eq(StorageType.DISK)))
.thenAnswer(invocation -> {
final int ii = loopCount.getAndIncrement() % volumes.size();
return volumes.get(ii);
@@ -318,7 +320,7 @@ private void putChunksInBlock(
data.setSchemaVersion(SCHEMA_V2);
KeyValueContainer container = new KeyValueContainer(data, conf);
- container.create(volumeSet, volumeChoosingPolicy, SCM_ID);
+ container.create(volumeSet, volumeChoosingPolicy, SCM_ID,
StorageType.DISK);
containerSet.addContainer(container);
data = (KeyValueContainerData) containerSet.getContainer(containerId)
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestContainerSmallFile.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestContainerSmallFile.java
index 8140ea3eb45..c624b091f0f 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestContainerSmallFile.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestContainerSmallFile.java
@@ -81,7 +81,7 @@ public void testAllocateWrite() throws Exception {
BlockID blockID = ContainerTestHelper.getTestBlockID(
container.getContainerInfo().getContainerID());
ContainerProtocolCalls.writeSmallFile(client, blockID,
- "data123".getBytes(UTF_8), null);
+ "data123".getBytes(UTF_8), null, null);
ContainerProtos.GetSmallFileResponseProto response =
ContainerProtocolCalls.readSmallFile(client, blockID, null);
String readData = response.getData().getDataBuffers().getBuffersList()
@@ -124,7 +124,7 @@ public void testInvalidContainerRead() throws Exception {
BlockID blockID = ContainerTestHelper.getTestBlockID(
container.getContainerInfo().getContainerID());
ContainerProtocolCalls.writeSmallFile(client, blockID,
- "data123".getBytes(UTF_8), null);
+ "data123".getBytes(UTF_8), null, null);
assertThrowsExactly(StorageContainerException.class,
() -> ContainerProtocolCalls.readSmallFile(client,
@@ -149,7 +149,8 @@ public void testReadWriteWithBCSId() throws Exception {
container.getContainerInfo().getContainerID());
ContainerProtos.PutSmallFileResponseProto responseProto =
ContainerProtocolCalls
- .writeSmallFile(client, blockID1, "data123".getBytes(UTF_8), null);
+ .writeSmallFile(client, blockID1, "data123".getBytes(UTF_8), null,
+ null);
long bcsId = responseProto.getCommittedBlockLength().getBlockID()
.getBlockCommitSequenceId();
@@ -164,7 +165,7 @@ public void testReadWriteWithBCSId() throws Exception {
BlockID blockID2 = ContainerTestHelper
.getTestBlockID(container.getContainerInfo().getContainerID());
ContainerProtocolCalls
- .writeSmallFile(client, blockID2, "data123".getBytes(UTF_8), null);
+ .writeSmallFile(client, blockID2, "data123".getBytes(UTF_8), null,
null);
blockID1.setBlockCommitSequenceId(bcsId + 1);
//read a file with higher bcsId than the committed bcsId for the block
diff --git
a/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/ChunkManagerDiskWrite.java
b/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/ChunkManagerDiskWrite.java
index b7ca060419d..885ff965520 100644
---
a/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/ChunkManagerDiskWrite.java
+++
b/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/ChunkManagerDiskWrite.java
@@ -28,6 +28,7 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.lang3.RandomStringUtils;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.cli.HddsVersionProvider;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -130,7 +131,7 @@ public Void call() throws Exception {
KeyValueContainer keyValueContainer =
new KeyValueContainer(keyValueContainerData, ozoneConfiguration);
- keyValueContainer.create(volumeSet, volumeChoicePolicy, "scmid");
+ keyValueContainer.create(volumeSet, volumeChoicePolicy, "scmid",
StorageType.DISK);
containersPerThread.put(i, keyValueContainer);
}
diff --git
a/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/containergenerator/GeneratorDatanode.java
b/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/containergenerator/GeneratorDatanode.java
index 10f47fe2090..239026ddc52 100644
---
a/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/containergenerator/GeneratorDatanode.java
+++
b/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/containergenerator/GeneratorDatanode.java
@@ -34,6 +34,7 @@
import java.util.SplittableRandom;
import java.util.concurrent.Callable;
import java.util.stream.Stream;
+import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdds.cli.HddsVersionProvider;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
@@ -329,7 +330,7 @@ private KeyValueContainer createContainer(long containerId)
new KeyValueContainer(keyValueContainerData, config);
try {
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
+ keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId,
StorageType.DISK);
} catch (StorageContainerException ex) {
throw new RuntimeException(ex);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
