This is an automated email from the ASF dual-hosted git repository.
ivandika3 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 8acc0413a95 HDDS-14812. Support datanode topology in MiniOzoneCluster
(#10598)
8acc0413a95 is described below
commit 8acc0413a9548462df8a699133d009a230dff2c5
Author: Han-Wen Hsu <[email protected]>
AuthorDate: Thu Jul 30 09:19:58 2026 +0800
HDDS-14812. Support datanode topology in MiniOzoneCluster (#10598)
---
.../dev-support/findbugsExcludeFile.xml | 12 +
.../hadoop/hdds/scm/TestRackAwarePlacement.java | 412 +++++++++++++++++++++
.../org/apache/hadoop/ozone/FixedHostMapping.java | 153 ++++++++
.../org/apache/hadoop/ozone/MiniOzoneCluster.java | 50 +++
.../apache/hadoop/ozone/MiniOzoneClusterImpl.java | 92 +++++
.../hadoop/ozone/MiniOzoneHAClusterImpl.java | 1 +
6 files changed, 720 insertions(+)
diff --git a/hadoop-ozone/integration-test/dev-support/findbugsExcludeFile.xml
b/hadoop-ozone/integration-test/dev-support/findbugsExcludeFile.xml
index 55abc263017..bbc0b6b17af 100644
--- a/hadoop-ozone/integration-test/dev-support/findbugsExcludeFile.xml
+++ b/hadoop-ozone/integration-test/dev-support/findbugsExcludeFile.xml
@@ -16,4 +16,16 @@
limitations under the License.
-->
<FindBugsFilter>
+ <Match>
+ <Class
name="org.apache.hadoop.hdds.scm.TestRackAwarePlacement$WithRacksAndHosts"/>
+ <Bug pattern="SIC_INNER_SHOULD_BE_STATIC"/>
+ </Match>
+ <Match>
+ <Class
name="org.apache.hadoop.hdds.scm.TestRackAwarePlacement$WithRacksOnly"/>
+ <Bug pattern="SIC_INNER_SHOULD_BE_STATIC"/>
+ </Match>
+ <Match>
+ <Class
name="org.apache.hadoop.hdds.scm.TestRackAwarePlacement$WithHostsOnly"/>
+ <Bug pattern="SIC_INNER_SHOULD_BE_STATIC"/>
+ </Match>
</FindBugsFilter>
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestRackAwarePlacement.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestRackAwarePlacement.java
new file mode 100644
index 00000000000..c8258eb0e43
--- /dev/null
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestRackAwarePlacement.java
@@ -0,0 +1,412 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.scm;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
+import org.apache.hadoop.hdds.scm.container.ContainerID;
+import org.apache.hadoop.hdds.scm.container.ContainerInfo;
+import org.apache.hadoop.hdds.scm.container.ContainerReplica;
+import org.apache.hadoop.hdds.scm.node.NodeManager;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.apache.hadoop.net.NetworkTopology;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.client.ObjectStore;
+import org.apache.hadoop.ozone.client.OzoneBucket;
+import org.apache.hadoop.ozone.client.OzoneClient;
+import org.apache.hadoop.ozone.client.OzoneVolume;
+import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
+import org.apache.ozone.test.GenericTestUtils;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+/**
+ * Integration tests that verify rack/host topology is correctly propagated
+ * to SCM and that pipeline and container placement respect rack boundaries.
+ *
+ */
+public class TestRackAwarePlacement {
+
+ private static final String RACK0 = "/rack0";
+ private static final String RACK1 = "/rack1";
+
+ private static final String[] RACKS = {
+ RACK0, RACK0, RACK0,
+ RACK1, RACK1, RACK1
+ };
+
+ private static final String[] HOSTS = {
+ "host0.test", "host1.test", "host2.test",
+ "host3.test", "host4.test", "host5.test"
+ };
+
+ private static void applyReplicationSpeedupConfig(OzoneConfiguration conf) {
+ conf.setTimeDuration(ScmConfigKeys.OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL,
+ 100, TimeUnit.MILLISECONDS);
+ conf.setTimeDuration(ScmConfigKeys.OZONE_SCM_STALENODE_INTERVAL,
+ 3, TimeUnit.SECONDS);
+ conf.setTimeDuration(ScmConfigKeys.OZONE_SCM_DEADNODE_INTERVAL,
+ 6, TimeUnit.SECONDS);
+ conf.setTimeDuration("hdds.scm.replication.thread.interval",
+ 1, TimeUnit.SECONDS);
+ conf.setTimeDuration("hdds.scm.replication.under.replicated.interval",
+ 5, TimeUnit.SECONDS);
+ conf.setTimeDuration("hdds.scm.replication.over.replicated.interval",
+ 5, TimeUnit.SECONDS);
+ }
+
+ static Stream<Arguments> rackAwarePolicies() {
+ return Stream.of(
+ Arguments.of(
+ "org.apache.hadoop.hdds.scm.container.placement.algorithms"
+ + ".SCMContainerPlacementRackAware"),
+ Arguments.of(
+ "org.apache.hadoop.hdds.scm.container.placement.algorithms"
+ + ".SCMContainerPlacementRackScatter")
+ );
+ }
+
+ @ParameterizedTest
+ @MethodSource("rackAwarePolicies")
+ void testContainerPlacementWithPolicy(
+ String placementClassName) throws Exception {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.set(ScmConfigKeys.OZONE_SCM_CONTAINER_PLACEMENT_IMPL_KEY,
+ placementClassName);
+ applyReplicationSpeedupConfig(conf);
+
+ try (MiniOzoneCluster cluster = MiniOzoneCluster.newBuilder(conf)
+ .setNumDatanodes(RACKS.length)
+ .setRacks(RACKS)
+ .setHosts(HOSTS)
+ .build()) {
+ cluster.waitForClusterToBeReady();
+ cluster.waitForPipelineTobeReady(ReplicationFactor.THREE, 60_000);
+
+ StorageContainerManager scm = cluster.getStorageContainerManager();
+ PlacementPolicy actualPolicy = scm.getContainerPlacementPolicy();
+ assertEquals(placementClassName, actualPolicy.getClass().getName(),
+ "Placement policy was not set correctly");
+
+ assertPipelinesSpanMultipleRacks(cluster);
+ assertContainerReplicationIsRackAware(cluster);
+ }
+ }
+
+ @Nested
+ @TestInstance(TestInstance.Lifecycle.PER_CLASS)
+ class WithRacksAndHosts {
+
+ private MiniOzoneCluster cluster;
+
+ @BeforeAll
+ void init() throws Exception {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ applyReplicationSpeedupConfig(conf);
+ cluster = MiniOzoneCluster.newBuilder(conf)
+ .setNumDatanodes(RACKS.length)
+ .setRacks(RACKS)
+ .setHosts(HOSTS)
+ .build();
+ cluster.waitForClusterToBeReady();
+ cluster.waitForPipelineTobeReady(ReplicationFactor.THREE, 60_000);
+ }
+
+ @AfterAll
+ void tearDown() {
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ }
+
+ @Test
+ void testDatanodesHaveCorrectRack() {
+ assertRackAssignments(cluster, RACKS);
+ }
+
+ @Test
+ void testDatanodesHaveCorrectHostname() {
+ assertHostnameAssignments(cluster, HOSTS);
+ }
+
+ @Test
+ void testRatisPipelineSpansMultipleRacks() {
+ assertPipelinesSpanMultipleRacks(cluster);
+ }
+
+ @Test
+ void testContainerReplicationIsRackAware() throws Exception {
+ assertContainerReplicationIsRackAware(cluster);
+ }
+ }
+
+ @Nested
+ @TestInstance(TestInstance.Lifecycle.PER_CLASS)
+ class WithRacksOnly {
+
+ private MiniOzoneCluster cluster;
+
+ @BeforeAll
+ void init() throws Exception {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ applyReplicationSpeedupConfig(conf);
+ cluster = MiniOzoneCluster.newBuilder(conf)
+ .setNumDatanodes(RACKS.length)
+ .setRacks(RACKS)
+ .build();
+ cluster.waitForClusterToBeReady();
+ cluster.waitForPipelineTobeReady(ReplicationFactor.THREE, 60_000);
+ }
+
+ @AfterAll
+ void tearDown() {
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ }
+
+ @Test
+ void testDatanodesHaveCorrectRack() {
+ assertRackAssignments(cluster, RACKS);
+ }
+
+ @Test
+ void testRatisPipelineSpansMultipleRacks() {
+ assertPipelinesSpanMultipleRacks(cluster);
+ }
+
+ @Test
+ void testContainerReplicationIsRackAware() throws Exception {
+ assertContainerReplicationIsRackAware(cluster);
+ }
+ }
+
+ @Nested
+ @TestInstance(TestInstance.Lifecycle.PER_CLASS)
+ class WithHostsOnly {
+
+ private MiniOzoneCluster cluster;
+
+ @BeforeAll
+ void init() throws Exception {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ cluster = MiniOzoneCluster.newBuilder(conf)
+ .setNumDatanodes(HOSTS.length)
+ .setHosts(HOSTS)
+ .build();
+ cluster.waitForClusterToBeReady();
+ cluster.waitForPipelineTobeReady(ReplicationFactor.THREE, 60_000);
+ }
+
+ @AfterAll
+ void tearDown() {
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ }
+
+ @Test
+ void testDatanodesHaveCorrectHostname() {
+ assertHostnameAssignments(cluster, HOSTS);
+ }
+
+ @Test
+ void testDatanodesAllInDefaultRack() {
+ NodeManager nodeManager =
+ cluster.getStorageContainerManager().getScmNodeManager();
+ List<? extends DatanodeDetails> allNodes = nodeManager.getAllNodes();
+
+ for (DatanodeDetails dn : allNodes) {
+ assertEquals(NetworkTopology.DEFAULT_RACK, dn.getNetworkLocation(),
+ "Datanode " + dn.getHostName()
+ + " should be in default rack when no racks are configured");
+ }
+ }
+ }
+
+ private static void assertContainerReplicationIsRackAware(
+ MiniOzoneCluster cluster) throws Exception {
+ StorageContainerManager scm = cluster.getStorageContainerManager();
+
+ try (OzoneClient client = cluster.newClient()) {
+ ObjectStore store = client.getObjectStore();
+ store.createVolume("testvol");
+ OzoneVolume volume = store.getVolume("testvol");
+ volume.createBucket("testbucket");
+ OzoneBucket bucket = volume.getBucket("testbucket");
+
+ byte[] data = "test-data".getBytes(StandardCharsets.UTF_8);
+ try (OzoneOutputStream out = bucket.createKey(
+ "testkey", data.length,
+ RatisReplicationConfig.getInstance(ReplicationFactor.THREE),
+ new HashMap<>())) {
+ out.write(data);
+ }
+ }
+
+ ContainerInfo targetContainer = null;
+ Set<ContainerReplica> replicas = null;
+ for (ContainerInfo c : scm.getContainerManager().getContainers()) {
+ Set<ContainerReplica> r =
+ scm.getContainerManager().getContainerReplicas(c.containerID());
+ if (r.size() >= 3) {
+ targetContainer = c;
+ replicas = r;
+ break;
+ }
+ }
+ assertNotNull(targetContainer,
+ "Should find a container with 3 replicas");
+ ContainerID containerID = targetContainer.containerID();
+
+ DatanodeDetails stoppedDn =
+ replicas.iterator().next().getDatanodeDetails();
+ cluster.shutdownHddsDatanode(stoppedDn);
+
+ GenericTestUtils.waitFor(() -> {
+ try {
+ return scm.getScmNodeManager()
+ .getNodeStatus(stoppedDn)
+ .getHealth() == HddsProtos.NodeState.DEAD;
+ } catch (Exception e) {
+ return false;
+ }
+ }, 500, 30_000);
+
+ GenericTestUtils.waitFor(() -> {
+ try {
+ return scm.getContainerManager()
+ .getContainerReplicas(containerID)
+ .size() >= 3;
+ } catch (Exception e) {
+ return false;
+ }
+ }, 1_000, 60_000);
+
+ Set<String> racks = scm.getContainerManager()
+ .getContainerReplicas(containerID)
+ .stream()
+ .map(r -> r.getDatanodeDetails().getNetworkLocation())
+ .collect(Collectors.toSet());
+
+ assertTrue(racks.size() >= 2,
+ "Container replicas after re-replication should span at least "
+ + "2 racks, but were on: " + racks);
+ }
+
+ private static void assertRackAssignments(MiniOzoneCluster cluster,
+ String[] expectedRacks) {
+ NodeManager nodeManager =
+ cluster.getStorageContainerManager().getScmNodeManager();
+ List<? extends DatanodeDetails> allNodes = nodeManager.getAllNodes();
+
+ assertEquals(expectedRacks.length, allNodes.size(),
+ "Number of registered datanodes should match number of configured
racks");
+
+ long actualRack0 = allNodes.stream()
+ .filter(dn -> RACK0.equals(dn.getNetworkLocation()))
+ .count();
+ long actualRack1 = allNodes.stream()
+ .filter(dn -> RACK1.equals(dn.getNetworkLocation()))
+ .count();
+
+ long expectedRack0 =
+ Arrays.stream(expectedRacks).filter(RACK0::equals).count();
+ long expectedRack1 =
+ Arrays.stream(expectedRacks).filter(RACK1::equals).count();
+
+ assertEquals(expectedRack0, actualRack0,
+ "Expected " + expectedRack0 + " datanodes on " + RACK0);
+ assertEquals(expectedRack1, actualRack1,
+ "Expected " + expectedRack1 + " datanodes on " + RACK1);
+
+ for (DatanodeDetails dn : allNodes) {
+ String location = dn.getNetworkLocation();
+ assertNotNull(location,
+ "Network location must not be null for " + dn.getHostName());
+ assertTrue(location.equals(RACK0) || location.equals(RACK1),
+ "Unexpected rack for datanode " + dn.getHostName()
+ + ": " + location);
+ }
+ }
+
+ private static void assertHostnameAssignments(MiniOzoneCluster cluster,
+ String[] expectedHosts) {
+ NodeManager nodeManager =
+ cluster.getStorageContainerManager().getScmNodeManager();
+ List<? extends DatanodeDetails> allNodes = nodeManager.getAllNodes();
+
+ assertEquals(expectedHosts.length, allNodes.size(),
+ "Number of registered datanodes should match number of configured
hosts");
+
+ Set<String> actual = allNodes.stream()
+ .map(DatanodeDetails::getHostName)
+ .collect(Collectors.toSet());
+
+ Set<String> expected = Arrays.stream(expectedHosts)
+ .collect(Collectors.toSet());
+
+ assertEquals(expected, actual,
+ "Registered datanode hostnames should match configured hosts");
+ }
+
+ private static void assertPipelinesSpanMultipleRacks(
+ MiniOzoneCluster cluster) {
+ List<Pipeline> pipelines = cluster.getStorageContainerManager()
+ .getPipelineManager()
+ .getPipelines(
+ RatisReplicationConfig.getInstance(ReplicationFactor.THREE),
+ Pipeline.PipelineState.OPEN);
+
+ assertFalse(pipelines.isEmpty(),
+ "There should be at least one open RATIS THREE pipeline");
+
+ for (Pipeline pipeline : pipelines) {
+ Set<String> racks = pipeline.getNodes().stream()
+ .map(DatanodeDetails::getNetworkLocation)
+ .collect(Collectors.toSet());
+
+ assertTrue(racks.size() >= 2,
+ "Pipeline " + pipeline.getId()
+ + " should span at least 2 racks, but spans: " + racks);
+ }
+ }
+}
diff --git
a/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/FixedHostMapping.java
b/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/FixedHostMapping.java
new file mode 100644
index 00000000000..801af9fb1cf
--- /dev/null
+++
b/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/FixedHostMapping.java
@@ -0,0 +1,153 @@
+/*
+ * 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.ozone;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import org.apache.hadoop.net.CachedDNSToSwitchMapping;
+import org.apache.hadoop.net.DNSToSwitchMapping;
+import org.apache.hadoop.net.NetworkTopology;
+
+/**
+ * A {@link CachedDNSToSwitchMapping} implementation that resolves hostnames
+ * to rack locations using a statically configured map, bypassing DNS lookups.
+ *
+ * <p>This is intended for use in test environments (e.g. {@code
MiniOzoneCluster})
+ * where DataNode hostnames may be synthetic or unresolvable via DNS. The
standard
+ * {@link CachedDNSToSwitchMapping} performs DNS normalization before rack
resolution,
+ * which can cause synthetic hostnames to be incorrectly resolved to a real IP
address,
+ * leading to rack mapping failures. This class avoids that by resolving
directly
+ * against the registered hostname.
+ *
+ * <p>The mapping is stored in a JVM-wide static map. Callers must invoke
+ * {@link #addNode(String, String)} before cluster startup to register
hostname-to-rack
+ * entries, and should call {@link #clear()} after each test to avoid
cross-test pollution.
+ *
+ * <p>Usage:
+ * <pre>{@code
+ * FixedHostMapping.addNode("dn-0.test", "/rack1");
+ * FixedHostMapping.addNode("dn-1.test", "/rack1");
+ * FixedHostMapping.addNode("dn-2.test", "/rack2");
+ *
+ * conf.setClass(
+ * CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
+ * FixedHostMapping.class,
+ * DNSToSwitchMapping.class);
+ * }</pre>
+ */
+public class FixedHostMapping extends CachedDNSToSwitchMapping {
+
+ private static final Map<String, String> RACK_MAP = new
ConcurrentHashMap<>();
+
+ /**
+ * Constructs a {@code FixedHostMapping} with a no-op raw mapping.
+ * The raw mapping is unused since {@link #resolve(List)} is fully
overridden.
+ */
+ public FixedHostMapping() {
+ super(new NoOpMapping());
+ }
+
+ /**
+ * Constructs a {@code FixedHostMapping} with the given raw mapping.
+ * The raw mapping is unused since {@link #resolve(List)} is fully
overridden,
+ * but is accepted to satisfy {@link CachedDNSToSwitchMapping} constructor
requirements.
+ *
+ * @param rawMapping the raw DNS mapping (not used for resolution)
+ */
+ public FixedHostMapping(DNSToSwitchMapping rawMapping) {
+ super(rawMapping);
+ }
+
+ /**
+ * Registers a hostname-to-rack mapping entry.
+ * Must be called before cluster startup for the mapping to take effect
during
+ * DataNode registration.
+ *
+ * @param host the DataNode hostname as it will appear in {@link
#resolve(List)}
+ * @param rack the rack path (e.g. {@code "/rack1"})
+ */
+ public static void addNode(String host, String rack) {
+ RACK_MAP.put(host, rack);
+ }
+
+ /**
+ * Clears all registered hostname-to-rack mappings.
+ * Should be called in test teardown (e.g. {@code @AfterEach}) to prevent
+ * cross-test pollution of the JVM-wide static map.
+ */
+ public static void clear() {
+ RACK_MAP.clear();
+ }
+
+ /**
+ * Resolves a list of hostnames to their rack locations using the static map.
+ * Hostnames not present in the map are assigned {@link
NetworkTopology#DEFAULT_RACK}.
+ * Unlike the parent class, this method does not perform DNS normalization.
+ *
+ * @param names the list of hostnames to resolve
+ * @return a list of rack paths in the same order as the input
+ */
+ @Override
+ public List<String> resolve(List<String> names) {
+ return names.stream()
+ .map(name -> RACK_MAP.getOrDefault(name, NetworkTopology.DEFAULT_RACK))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * No-op: this implementation does not maintain a cache.
+ */
+ @Override
+ public void reloadCachedMappings() {
+ }
+
+ /**
+ * No-op: this implementation does not maintain a cache.
+ *
+ * @param names the hostnames whose cached mappings should be reloaded
(ignored)
+ */
+ @Override
+ public void reloadCachedMappings(List<String> names) {
+ }
+
+ /**
+ * A no-op {@link DNSToSwitchMapping} used as a placeholder raw mapping.
+ * All hostnames are mapped to {@link NetworkTopology#DEFAULT_RACK}.
+ * This is never invoked during normal resolution since {@link
FixedHostMapping#resolve(List)}
+ * is fully overridden.
+ */
+ private static class NoOpMapping implements DNSToSwitchMapping {
+
+ @Override
+ public List<String> resolve(List<String> names) {
+ return names.stream()
+ .map(n -> NetworkTopology.DEFAULT_RACK)
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public void reloadCachedMappings() {
+ }
+
+ @Override
+ public void reloadCachedMappings(List<String> names) {
+ }
+ }
+}
diff --git
a/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneCluster.java
b/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneCluster.java
index d768e44756b..8765c2aaaae 100644
---
a/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneCluster.java
+++
b/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneCluster.java
@@ -19,9 +19,11 @@
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeoutException;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -263,6 +265,8 @@ abstract class Builder {
protected CertificateClient certClient;
protected SecretKeyClient secretKeyClient;
protected DatanodeFactory dnFactory =
UniformDatanodesFactory.newBuilder().build();
+ protected String[] racks;
+ protected String[] hosts;
private final List<Service> services = new ArrayList<>();
protected Builder(OzoneConfiguration conf) {
@@ -289,6 +293,15 @@ protected void prepareForNextBuild() {
conf.unset(OMConfigKeys.OZONE_OM_DB_DIRS);
conf.unset(OMConfigKeys.OZONE_OM_SNAPSHOT_DIFF_DB_DIR);
+ // dn rack configs
+ if (racks != null) {
+
conf.unset(CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY);
+ conf.unset(HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME);
+ conf.unset("hadoop.configured.node.mapping");
+ racks = null;
+ hosts = null;
+ }
+
setClusterId();
}
@@ -369,6 +382,43 @@ public Builder setDatanodeFactory(DatanodeFactory factory)
{
return this;
}
+ /**
+ * Sets the rack location for each datanode. Each entry is a rack path
+ * such as {@code "/rack0"}. The length of the array must match the
+ * number of datanodes.
+ *
+ * @param racks rack path per datanode
+ * @return this Builder
+ */
+ public Builder setRacks(String[] racks) {
+ this.racks = Arrays.copyOf(racks, racks.length);
+ return this;
+ }
+
+ /**
+ * Sets the hostname for each datanode. When used together with
+ * {@link #setRacks}, the hostnames are used as keys in the
+ * {@code StaticMapping} instead of the default synthetic names
+ * ({@code "dn-0"}, {@code "dn-1"}, …). The length of the array must
+ * match the number of datanodes.
+ *
+ * @param hosts hostname per datanode
+ * @return this Builder
+ */
+ public Builder setHosts(String[] hosts) {
+ this.hosts = Arrays.copyOf(hosts, hosts.length);
+ return this;
+ }
+
+ protected void validateDatanodeConfiguration() {
+ if (racks != null && racks.length != numOfDatanodes) {
+ throw new IllegalArgumentException("Number of racks must match the
number of datanodes");
+ }
+ if (hosts != null && hosts.length != numOfDatanodes) {
+ throw new IllegalArgumentException("Number of hosts must match the
number of datanodes");
+ }
+ }
+
public Builder addService(Service service) {
services.add(service);
return this;
diff --git
a/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java
b/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java
index b0e2c1efbdb..e13719dde6e 100644
---
a/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java
+++
b/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java
@@ -39,12 +39,14 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.HddsUtils;
import org.apache.hadoop.hdds.annotation.InterfaceAudience;
import org.apache.hadoop.hdds.client.RatisReplicationConfig;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.DatanodeID;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.scm.HddsTestUtils;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
@@ -54,6 +56,7 @@
import org.apache.hadoop.hdds.scm.node.NodeStatus;
import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineManager;
import
org.apache.hadoop.hdds.scm.protocolPB.StorageContainerLocationProtocolClientSideTranslatorPB;
import org.apache.hadoop.hdds.scm.proxy.SCMClientConfig;
import
org.apache.hadoop.hdds.scm.proxy.SCMContainerLocationFailoverProxyProvider;
@@ -64,14 +67,17 @@
import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
import org.apache.hadoop.hdds.security.symmetric.SecretKeyClient;
import
org.apache.hadoop.hdds.security.x509.certificate.client.CertificateClient;
+import org.apache.hadoop.hdds.utils.HddsServerUtil;
import org.apache.hadoop.hdds.utils.IOUtils;
import org.apache.hadoop.hdds.utils.db.CodecBuffer;
import org.apache.hadoop.hdds.utils.db.CodecTestUtil;
import org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectMetrics;
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
+import org.apache.hadoop.net.DNSToSwitchMapping;
import org.apache.hadoop.ozone.client.OzoneClient;
import org.apache.hadoop.ozone.client.OzoneClientFactory;
import org.apache.hadoop.ozone.common.Storage.StorageState;
+import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils;
import org.apache.hadoop.ozone.container.common.utils.ContainerCache;
import org.apache.hadoop.ozone.container.common.utils.DatanodeStoreCache;
import org.apache.hadoop.ozone.om.OMConfigKeys;
@@ -505,6 +511,7 @@ public Builder(OzoneConfiguration conf) {
@Override
public MiniOzoneCluster build() throws IOException {
+ validateDatanodeConfiguration();
DefaultMetricsSystem.setMiniClusterMode(true);
DatanodeStoreCache.setMiniClusterMode();
initializeConfiguration();
@@ -527,6 +534,12 @@ public MiniOzoneCluster build() throws IOException {
cluster.startHddsDatanodes();
}
+ // Recreate the Ratis pipeline to prevent imbalanced node placement
across racks
+ // caused by asynchronous DN registration.
+ if (racks != null && startDataNodes) {
+ resetPipelinesForRackAwareness(cluster);
+ }
+
prepareForNextBuild();
return cluster;
} catch (Exception ex) {
@@ -559,6 +572,33 @@ protected void setClients(OzoneManager om) throws
IOException {
}
}
+ /**
+ * Waits for all DNs to be healthy, then removes any pipelines that
+ * were created before the full rack topology was visible, and creates one
+ * fresh rack-aware pipeline directly (bypassing the background timer).
+ */
+ private void resetPipelinesForRackAwareness(MiniOzoneClusterImpl cluster)
+ throws IOException {
+ try {
+ cluster.waitForClusterToBeReady();
+ } catch (TimeoutException | InterruptedException e) {
+ throw new IOException(
+ "Timed out waiting for rack-aware cluster to be ready", e);
+ }
+ RatisReplicationConfig threeWay =
+
RatisReplicationConfig.getInstance(HddsProtos.ReplicationFactor.THREE);
+ PipelineManager pm =
+ cluster.getStorageContainerManager().getPipelineManager();
+ for (Pipeline p : pm.getPipelines(threeWay)) {
+ if (!p.isClosed()) {
+ pm.closePipeline(p.getId());
+ }
+ pm.deletePipeline(p.getId());
+ }
+
+ pm.createPipeline(threeWay);
+ }
+
/**
* Initializes the configuration required for starting MiniOzoneCluster.
*/
@@ -577,6 +617,40 @@ protected void initializeConfiguration() throws
IOException {
// pipeline.
conf.setInt(HddsConfigKeys.HDDS_SCM_SAFEMODE_MIN_DATANODE,
numOfDatanodes >= 3 ? 3 : 1);
+
+ configureHostAndRackTopology();
+ }
+
+ private void configureHostAndRackTopology() throws IOException {
+ FixedHostMapping.clear();
+ if (racks == null && hosts == null) {
+ return;
+ }
+
+ conf.setBoolean(HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME, true);
+
+ if (hosts == null) {
+ hosts = new String[racks.length];
+ for (int i = 0; i < racks.length; i++) {
+ hosts[i] = "host" + i + ".foo.com";
+ }
+ }
+
+ if (racks != null) {
+
+ if (hosts.length != racks.length) {
+ throw new IllegalArgumentException(
+ "The length of hosts [" + hosts.length
+ + "] must match the length of racks [" + racks.length +
"].");
+ }
+
+
conf.setClass(CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
+ FixedHostMapping.class, DNSToSwitchMapping.class);
+
+ for (int i = 0; i < racks.length; i++) {
+ FixedHostMapping.addNode(hosts[i], racks[i]);
+ }
+ }
}
void removeConfiguration() {
@@ -698,14 +772,32 @@ protected List<HddsDatanodeService> createHddsDatanodes()
for (int i = 0; i < numOfDatanodes; i++) {
OzoneConfiguration dnConf = dnFactory.apply(conf);
+ if (hosts != null) {
+ dnConf.set(HddsConfigKeys.HDDS_DATANODE_HOST_NAME_KEY, hosts[i]);
+ }
+ // Bypass InetAddress.getName() resolution for custom hostnames by
starting DN via YAML.
+ confDatanodeViaYaml(dnConf);
HddsDatanodeService datanode = new HddsDatanodeService(NO_ARGS);
+ dnConf.setStrings(ScmConfigKeys.OZONE_SCM_NAMES,
conf.getStrings(ScmConfigKeys.OZONE_SCM_NAMES));
datanode.setConfiguration(dnConf);
hddsDatanodes.add(datanode);
}
+
return hddsDatanodes;
}
+ private void confDatanodeViaYaml(OzoneConfiguration dnConf) throws
IOException {
+ DatanodeDetails datanodeDetails = DatanodeDetails.newBuilder()
+ .setID(DatanodeID.randomID())
+ .setHostName(dnConf.get(HddsConfigKeys.HDDS_DATANODE_HOST_NAME_KEY))
+ .setIpAddress("127.0.0.1")
+ .build();
+ datanodeDetails.setNetworkName(datanodeDetails.getUuidString());
+ String dnFilePath = HddsServerUtil.getDatanodeIdFilePath(dnConf);
+ ContainerUtils.writeDatanodeDetailsTo(datanodeDetails, new
File(dnFilePath), dnConf);
+ }
+
protected void configureSCM(boolean isHA) throws IOException {
conf.set(ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY,
localhostWithFreePort());
diff --git
a/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneHAClusterImpl.java
b/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneHAClusterImpl.java
index 303a16a1a63..507777d8bfe 100644
---
a/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneHAClusterImpl.java
+++
b/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneHAClusterImpl.java
@@ -464,6 +464,7 @@ public Builder setSCMServiceId(String serviceId) {
@Override
public MiniOzoneHAClusterImpl build() throws IOException {
+ validateDatanodeConfiguration();
if (numOfActiveOMs > numOfOMs) {
throw new IllegalArgumentException("Number of active OMs cannot be " +
"more than the total number of OMs");
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
