From 091b2683d56b34c6f68ea6f7183868b69d376dac Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Sat, 12 Sep 2026 20:29:57 +0800 Subject: [PATCH 1/6] HDDS-15941. Return null from sortDatanodesForWrite when the sort is skipped --- .../hadoop/ozone/TestOMSortDatanodes.java | 20 +++++--------- .../apache/hadoop/ozone/om/KeyManager.java | 6 ++--- .../hadoop/ozone/om/KeyManagerImpl.java | 10 ++++--- .../ozone/om/request/key/OMKeyRequest.java | 10 +++---- .../key/TestOMAllocateBlockRequest.java | 27 +++++++++++-------- 5 files changed, 37 insertions(+), 36 deletions(-) diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java index a206276ce8fd..ca983dc86992 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java @@ -22,7 +22,7 @@ import static org.apache.hadoop.hdds.scm.net.NetConstants.ROOT_LEVEL; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotEquals; -import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.mockito.Mockito.mock; import com.google.common.collect.ImmutableMap; @@ -204,28 +204,22 @@ public void sortDatanodesForWriteSortsRpcDeserializedPipeline() { } @Test - public void sortDatanodesForWriteKeepsOrderForStaleTopology() { + public void sortDatanodesForWriteReturnsNullForStaleTopology() { List nodes = new ArrayList<>(); nodes.add(randomDatanodeDetails()); nodes.addAll(nodeManager.getAllNodes()); - List sorted = - keyManager.sortDatanodesForWrite(nodes, "edge0", om.getClusterMap()); - - assertSame(nodes, sorted, - "Pipeline order should be preserved when a node is missing from the OM topology"); + assertNull(keyManager.sortDatanodesForWrite(nodes, "edge0", om.getClusterMap()), + "Sort must be skipped when a node is missing from the OM topology"); } @Test - public void sortDatanodesForWriteKeepsOrderWhenClientUnresolved() { + public void sortDatanodesForWriteReturnsNullWhenClientUnresolved() { List nodes = nodeManager.getAllNodes(); - List original = new ArrayList<>(nodes); // A client that resolves to no known rack must NOT trigger a shuffle. String unresolved = nodes.get(0).getIpAddress() + "X"; - List result = - keyManager.sortDatanodesForWrite(nodes, unresolved, om.getClusterMap()); - assertEquals(original, result, - "Write pipeline order must be preserved when client is unresolved"); + assertNull(keyManager.sortDatanodesForWrite(nodes, unresolved, om.getClusterMap()), + "Sort must be skipped when the client is unresolved"); } private String nodeAddress(DatanodeDetails dn) { diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java index 2c959cb4c018..e1b6bee142ca 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java @@ -410,9 +410,9 @@ DeleteKeysResult getPendingDeletionSubFiles(long volumeId, long bucketId, OmKeyI * @param nodes the pipeline nodes to sort * @param clientMachine client address (IP or hostname) * @param clusterMap OM's cached cluster map used to resolve topology distance - * @return nodes sorted nearest-first, or the original {@code nodes} list - * instance unchanged when sorting is skipped (client unresolved or stale - * topology); callers may use reference equality to detect a skipped sort + * @return nodes sorted nearest-first, or null when the sort is skipped + * (client unresolved, or a pipeline node missing from the cluster map); + * callers must leave the pipeline order unchanged in that case */ List sortDatanodesForWrite( List nodes, String clientMachine, NetworkTopology clusterMap); diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java index d74614579e24..443a0d571157 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java @@ -2322,9 +2322,9 @@ public List sortDatanodesForWrite( metrics.getAllocateBlockSortDatanodesLatencyNs(), () -> { final Node client = getClientNode(clientMachine, nodes, clusterMap); if (client == null) { - // Preserve pipeline order for writes: the first node is the write - // primary, so do not shuffle when the client cannot be resolved. - return nodes; + // Skip the sort for writes: the first node is the write primary, so + // do not shuffle when the client cannot be resolved. + return null; } return sortByClusterMapDistance(clusterMap, client, nodes); }); @@ -2342,6 +2342,8 @@ public boolean isSortDatanodesForWriteEnabled() { * {@link Integer#MAX_VALUE}) and the order comes out random. Look each node * up in OM's cluster map to get the topology-linked instance, sort those, * then map the order back to the original nodes. + * + * @return the sorted nodes, or null when a node is missing from the cluster map */ private List sortByClusterMapDistance( NetworkTopology clusterMap, Node client, @@ -2351,7 +2353,7 @@ private List sortByClusterMapDistance( for (DatanodeDetails node : nodes) { final Node resolved = clusterMap.getNode(node.getNetworkFullPath()); if (resolved == null) { - return nodes; + return null; } topologyNodes.add(resolved); nodeByPath.put(resolved.getNetworkFullPath(), node); diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java index 3d478e461396..a0ce7d4b882a 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java @@ -251,14 +251,14 @@ protected List allocateBlock( List sorted = sortedByNodes.get(uuidSet); if (sorted == null) { sorted = keyManager.sortDatanodesForWrite(nodes, omClientMachine, clusterMap); - // Cache only a freshly sorted order, not an input list returned - // unchanged when the client is unresolved: that order is per-pipeline - // and must not be reused for another pipeline with the same node set. - if (sorted != nodes) { + // A skipped sort returns null and is not cached: that pipeline keeps + // its own order, which must not be reused for another pipeline with + // the same node set. + if (sorted != null) { sortedByNodes.put(uuidSet, sorted); } } - if (!Objects.equals(sorted, pipeline.getNodesInOrder())) { + if (sorted != null && !Objects.equals(sorted, pipeline.getNodesInOrder())) { pipeline = pipeline.copyWithNodesInOrder(sorted); } } diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java index f389717907c4..784728ab3701 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java @@ -371,20 +371,26 @@ public void testAllocateBlockSortsSharedPipelineOnce() throws Exception { @Test public void testAllocateBlockKeepsPerPipelineOrderWhenSortSkipped() throws Exception { // Two pipelines share the same datanode set but in a different order. When - // the sort is skipped (sortDatanodesForWrite returns the input unchanged), - // each pipeline must keep its own order: the unsorted result must not be - // cached under the node set and reused for the other pipeline. + // the sort is skipped (sortDatanodesForWrite returns null), each pipeline + // must keep its own order: null must not be cached under the node set and + // reused for the other pipeline. DatanodeDetails a = MockDatanodeDetails.randomDatanodeDetails(); DatanodeDetails b = MockDatanodeDetails.randomDatanodeDetails(); DatanodeDetails c = MockDatanodeDetails.randomDatanodeDetails(); List nodes1 = Arrays.asList(a, b, c); List nodes2 = Arrays.asList(c, b, a); + // nodesInOrder deliberately differs from getNodes() so the test catches an + // implementation that overwrites an existing order with getNodes() when the + // sort is skipped. + List inOrder1 = Arrays.asList(b, c, a); + List inOrder2 = Arrays.asList(a, c, b); Pipeline pipeline1 = Pipeline.newBuilder() .setState(Pipeline.PipelineState.OPEN) .setId(PipelineID.randomId()) .setReplicationConfig( StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE)) .setNodes(nodes1) + .setNodesInOrder(inOrder1) .build(); Pipeline pipeline2 = Pipeline.newBuilder() .setState(Pipeline.PipelineState.OPEN) @@ -392,6 +398,7 @@ public void testAllocateBlockKeepsPerPipelineOrderWhenSortSkipped() throws Excep .setReplicationConfig( StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE)) .setNodes(nodes2) + .setNodesInOrder(inOrder2) .build(); AllocatedBlock block1 = new AllocatedBlock.Builder().setPipeline(pipeline1) .setContainerBlockID(new ContainerBlockID(CONTAINER_ID, LOCAL_ID)).build(); @@ -403,9 +410,8 @@ public void testAllocateBlockKeepsPerPipelineOrderWhenSortSkipped() throws Excep KeyManager mockKeyManager = mock(KeyManager.class); when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); - // Skip the sort: return the input list instance unchanged. - when(mockKeyManager.sortDatanodesForWrite(any(), any(), any())) - .thenAnswer(inv -> inv.getArgument(0)); + // Skip the sort: null signals that no sort happened. + when(mockKeyManager.sortDatanodesForWrite(any(), any(), any())).thenReturn(null); when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); @@ -416,11 +422,10 @@ public void testAllocateBlockKeepsPerPipelineOrderWhenSortSkipped() throws Excep UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(), ozoneManager); assertEquals(2, locations.size()); - // Each pipeline keeps its own order; the skipped-sort result is not shared. - assertEquals(nodes1, locations.get(0).getPipeline().getNodesInOrder()); - assertEquals(nodes2, locations.get(1).getPipeline().getNodesInOrder()); - // Sorted per pipeline, since the unsorted result is not cached. - verify(mockKeyManager, times(2)).sortDatanodesForWrite(any(), eq("1.2.3.4"), any()); + // Each pipeline keeps its own existing nodesInOrder; the skipped sort is + // neither shared nor replaced by getNodes(). + assertEquals(inOrder1, locations.get(0).getPipeline().getNodesInOrder()); + assertEquals(inOrder2, locations.get(1).getPipeline().getNodesInOrder()); } @Test From 3c438173be7363ce6b4868af9a4c1802ef60c5e0 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Sat, 12 Sep 2026 20:37:35 +0800 Subject: [PATCH 2/6] HDDS-15941. Resolve the write client once before the block loop and fall back to SCM --- .../hadoop/ozone/TestOMSortDatanodes.java | 130 ++++++++++++++++-- .../apache/hadoop/ozone/om/KeyManager.java | 30 ++-- .../hadoop/ozone/om/KeyManagerImpl.java | 32 +++-- .../ozone/om/request/key/OMKeyRequest.java | 14 +- .../key/TestOMAllocateBlockRequest.java | 130 ++++++++++++++++-- 5 files changed, 295 insertions(+), 41 deletions(-) diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java index ca983dc86992..9f02df65349b 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java @@ -22,6 +22,7 @@ import static org.apache.hadoop.hdds.scm.net.NetConstants.ROOT_LEVEL; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.mockito.Mockito.mock; @@ -30,12 +31,14 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.hadoop.hdds.HddsConfigKeys; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.scm.HddsTestUtils; import org.apache.hadoop.hdds.scm.ha.SCMContext; import org.apache.hadoop.hdds.scm.ha.SCMHAManagerStub; +import org.apache.hadoop.hdds.scm.net.Node; import org.apache.hadoop.hdds.scm.node.NodeManager; import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocol; import org.apache.hadoop.hdds.scm.server.SCMConfigurator; @@ -77,7 +80,7 @@ public static void setup() throws Exception { config = new OzoneConfiguration(); config.set(HddsConfigKeys.OZONE_METADATA_DIRS, dir.toString()); config.set(NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY, - StaticMapping.class.getName()); + CountingStaticMapping.class.getName()); config.set(OzoneConfigKeys.OZONE_NETWORK_TOPOLOGY_AWARE_READ_KEY, "true"); List datanodes = new ArrayList<>(NODE_COUNT); List nodeMapping = new ArrayList<>(NODE_COUNT); @@ -189,14 +192,16 @@ public void sortDatanodesForWriteSortsRpcDeserializedPipeline() { for (DatanodeDetails dn : nodeManager.getAllNodes()) { // The client address is normally an IP, but the sort must resolve a client // by either IP or hostname, so cover both. + Node ipClient = keyManager.resolveClientForWrite(dn.getIpAddress(), om.getClusterMap()); List byIp = - keyManager.sortDatanodesForWrite(rpcNodes, dn.getIpAddress(), om.getClusterMap()); + keyManager.sortDatanodesForWrite(rpcNodes, dn.getIpAddress(), ipClient, om.getClusterMap()); assertEquals(dn, byIp.get(0), "Source node should be sorted first for writes (IP client)"); assertRackOrder(dn.getNetworkLocation(), byIp); + Node hostClient = keyManager.resolveClientForWrite(dn.getHostName(), om.getClusterMap()); List byHostname = - keyManager.sortDatanodesForWrite(rpcNodes, dn.getHostName(), om.getClusterMap()); + keyManager.sortDatanodesForWrite(rpcNodes, dn.getHostName(), hostClient, om.getClusterMap()); assertEquals(dn, byHostname.get(0), "Source node should be sorted first for writes (hostname client)"); assertRackOrder(dn.getNetworkLocation(), byHostname); @@ -208,18 +213,112 @@ public void sortDatanodesForWriteReturnsNullForStaleTopology() { List nodes = new ArrayList<>(); nodes.add(randomDatanodeDetails()); nodes.addAll(nodeManager.getAllNodes()); + Node client = keyManager.resolveClientForWrite("edge0", om.getClusterMap()); + assertNotNull(client); - assertNull(keyManager.sortDatanodesForWrite(nodes, "edge0", om.getClusterMap()), + assertNull(keyManager.sortDatanodesForWrite(nodes, "edge0", client, om.getClusterMap()), "Sort must be skipped when a node is missing from the OM topology"); } @Test - public void sortDatanodesForWriteReturnsNullWhenClientUnresolved() { - List nodes = nodeManager.getAllNodes(); - // A client that resolves to no known rack must NOT trigger a shuffle. - String unresolved = nodes.get(0).getIpAddress() + "X"; - assertNull(keyManager.sortDatanodesForWrite(nodes, unresolved, om.getClusterMap()), - "Sort must be skipped when the client is unresolved"); + public void resolveClientForWriteReturnsNullWhenUnresolved() { + // A client that maps to no rack in the OM topology cannot be placed, so OM + // must report null and let the caller fall back to SCM. + String unresolved = nodeManager.getAllNodes().get(0).getIpAddress() + "X"; + assertNull(keyManager.resolveClientForWrite(unresolved, om.getClusterMap())); + } + + @Test + public void resolveClientForWriteAttachesNonDatanodeClientToItsRack() { + for (Map.Entry entry : EDGE_NODES.entrySet()) { + Node client = keyManager.resolveClientForWrite(entry.getKey(), om.getClusterMap()); + assertNotNull(client, "Edge client should resolve via the DNS-to-switch mapping"); + assertEquals(entry.getValue(), client.getNetworkLocation()); + } + } + + @Test + public void sortDatanodesForWritePrefersPipelineDatanodeAsClient() { + // The pre-resolved client is rack-level and equidistant to every datanode + // in its rack, so with two same-rack datanodes in the pipeline only the + // datanode identity can put the client's own datanode first. To make the + // test deterministic (no equidistant shuffle), pass a client resolved for + // the OTHER rack: without the pipeline match, an other-rack datanode would + // sort first; with it, the client datanode must. + List all = nodeManager.getAllNodes(); + DatanodeDetails clientDn = all.get(0); + DatanodeDetails sameRack = null; + List otherRack = new ArrayList<>(); + for (DatanodeDetails dn : all) { + if (dn.equals(clientDn)) { + continue; + } + if (dn.getNetworkLocation().equals(clientDn.getNetworkLocation())) { + if (sameRack == null) { + sameRack = dn; + } + } else if (otherRack.size() < 2) { + otherRack.add(dn); + } + } + assertNotNull(sameRack); + assertEquals(2, otherRack.size()); + String otherRackEdge = "/rack0".equals(clientDn.getNetworkLocation()) ? "edge1" : "edge0"; + Node otherRackClient = keyManager.resolveClientForWrite(otherRackEdge, om.getClusterMap()); + assertNotNull(otherRackClient); + assertNotEquals(clientDn.getNetworkLocation(), otherRackClient.getNetworkLocation()); + + for (String address : new String[] {clientDn.getIpAddress(), clientDn.getHostName()}) { + // Pipeline without the client datanode: the passed client decides, so an + // other-rack datanode sorts first. + List without = new ArrayList<>(); + without.add(rpcCopy(sameRack)); + without.add(rpcCopy(otherRack.get(0))); + without.add(rpcCopy(otherRack.get(1))); + List sortedWithout = + keyManager.sortDatanodesForWrite(without, address, otherRackClient, om.getClusterMap()); + assertNotNull(sortedWithout); + assertEquals(otherRackClient.getNetworkLocation(), sortedWithout.get(0).getNetworkLocation()); + + // Pipeline containing the client datanode: its identity wins over the + // passed other-rack client, so it sorts first (distance zero). + List with = new ArrayList<>(); + with.add(rpcCopy(otherRack.get(0))); + with.add(rpcCopy(sameRack)); + with.add(rpcCopy(clientDn)); + List sortedWith = + keyManager.sortDatanodesForWrite(with, address, otherRackClient, om.getClusterMap()); + assertNotNull(sortedWith); + assertEquals(clientDn, sortedWith.get(0)); + } + } + + @Test + public void resolveClientForWriteIsTheOnlyMappingLookup() { + // Resolve an edge client once, then sort several different datanode sets: + // the DNS-to-switch mapping must be consulted exactly once, by + // resolveClientForWrite, never by sortDatanodesForWrite. + List all = nodeManager.getAllNodes(); + CountingStaticMapping.RESOLVED_NAMES.set(0); + Node client = keyManager.resolveClientForWrite("edge0", om.getClusterMap()); + assertNotNull(client); + assertEquals(1, CountingStaticMapping.RESOLVED_NAMES.get()); + + for (int start = 0; start + 3 <= all.size(); start += 3) { + List pipeline = new ArrayList<>(); + for (DatanodeDetails dn : all.subList(start, start + 3)) { + pipeline.add(rpcCopy(dn)); + } + assertNotNull(keyManager.sortDatanodesForWrite(pipeline, "edge0", client, om.getClusterMap())); + } + assertEquals(1, CountingStaticMapping.RESOLVED_NAMES.get(), + "sortDatanodesForWrite must not call the DNS-to-switch mapping"); + } + + // Simulate a pipeline node as OM receives it from SCM over RPC: same identity, + // no topology linkage. + private static DatanodeDetails rpcCopy(DatanodeDetails dn) { + return DatanodeDetails.getFromProtoBuf(dn.getProtoBufMessage()); } private String nodeAddress(DatanodeDetails dn) { @@ -228,4 +327,15 @@ private String nodeAddress(DatanodeDetails dn) { HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME_DEFAULT); return useHostname ? dn.getHostName() : dn.getIpAddress(); } + + /** StaticMapping that counts how many names were resolved. */ + public static class CountingStaticMapping extends StaticMapping { + static final AtomicInteger RESOLVED_NAMES = new AtomicInteger(); + + @Override + public List resolve(List names) { + RESOLVED_NAMES.addAndGet(names.size()); + return super.resolve(names); + } + } } diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java index e1b6bee142ca..ab95f57d8315 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java @@ -25,6 +25,7 @@ import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.scm.net.NetworkTopology; +import org.apache.hadoop.hdds.scm.net.Node; import org.apache.hadoop.hdds.utils.BackgroundService; import org.apache.hadoop.hdds.utils.db.Table; import org.apache.hadoop.hdds.utils.db.TableIterator; @@ -401,21 +402,34 @@ DeleteKeysResult getPendingDeletionSubFiles(long volumeId, long bucketId, OmKeyI */ KeyLifecycleService getKeyLifecycleService(); + /** + * Resolve the streaming-write client to a node in OM's cached cluster map + * through the DNS-to-switch mapping. Called once per allocateBlock request, + * before the block loop, so the mapping is consulted at most once. + * + * @param clientMachine client address (IP or hostname), must not be empty + * @param clusterMap OM's cached cluster map + * @return a node attached to the client's rack, or null when the client + * cannot be placed in the topology; the caller then leaves the sort to SCM + */ + Node resolveClientForWrite(String clientMachine, NetworkTopology clusterMap); + /** * Sort the datanodes of a write pipeline by network-topology distance to the - * client, using OM's locally cached cluster map. Unlike the read-path sort, - * the original order is preserved when the client cannot be resolved, because - * the first node is used as the streaming-write primary. + * client, using OM's locally cached cluster map. When the client is one of the + * pipeline datanodes, that datanode is used as the client so it sorts first; + * otherwise {@code client} from {@link #resolveClientForWrite} is used. * * @param nodes the pipeline nodes to sort - * @param clientMachine client address (IP or hostname) + * @param clientMachine client address (IP or hostname), must not be empty + * @param client the node returned by {@link #resolveClientForWrite}, must not be null * @param clusterMap OM's cached cluster map used to resolve topology distance - * @return nodes sorted nearest-first, or null when the sort is skipped - * (client unresolved, or a pipeline node missing from the cluster map); - * callers must leave the pipeline order unchanged in that case + * @return nodes sorted nearest-first, or null when the sort is skipped because + * a pipeline node is missing from the cluster map; callers must leave the + * pipeline order unchanged in that case */ List sortDatanodesForWrite( - List nodes, String clientMachine, NetworkTopology clusterMap); + List nodes, String clientMachine, Node client, NetworkTopology clusterMap); /** * @return true if OM should sort the streaming-write pipeline locally diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java index 443a0d571157..63522bde9b5a 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java @@ -2312,21 +2312,25 @@ public List sortDatanodes(List sortDatanodesForWrite( - List nodes, String clientMachine, NetworkTopology clusterMap) { - Preconditions.checkArgument(!StringUtils.isEmpty(clientMachine), - "clientMachine is empty"); + List nodes, String clientMachine, Node client, NetworkTopology clusterMap) { + Preconditions.checkArgument(!StringUtils.isEmpty(clientMachine), "clientMachine is empty"); + Objects.requireNonNull(client, "client is null"); Objects.requireNonNull(clusterMap, "clusterMap is null"); return captureLatencyNs( metrics.getAllocateBlockSortDatanodesLatencyNs(), () -> { - final Node client = getClientNode(clientMachine, nodes, clusterMap); - if (client == null) { - // Skip the sort for writes: the first node is the write primary, so - // do not shuffle when the client cannot be resolved. - return null; - } - return sortByClusterMapDistance(clusterMap, client, nodes); + // A pipeline datanode matching the client is at distance zero from + // itself, so it sorts first; the rack-level client cannot do that. + final Node pipelineClient = findClientInPipeline(clientMachine, nodes, clusterMap); + return sortByClusterMapDistance(clusterMap, pipelineClient != null ? pipelineClient : client, nodes); }); } @@ -2369,6 +2373,12 @@ private List sortByClusterMapDistance( private Node getClientNode(String clientMachine, List nodes, NetworkTopology clusterMap) { + final Node pipelineClient = findClientInPipeline(clientMachine, nodes, clusterMap); + return pipelineClient != null ? pipelineClient : getOtherNode(clientMachine, clusterMap); + } + + private Node findClientInPipeline(String clientMachine, + List nodes, NetworkTopology clusterMap) { for (DatanodeDetails node : nodes) { // Match by either IP or hostname, like SCM's getNodesByAddress. clientMachine // may be a hostname on the read path; the streaming-write remoteAddress is @@ -2381,7 +2391,7 @@ private Node getClientNode(String clientMachine, return resolved != null ? resolved : node; } } - return getOtherNode(clientMachine, clusterMap); + return null; } private Node getOtherNode(String clientMachine, diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java index a0ce7d4b882a..8d8c28a636b6 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java @@ -64,6 +64,7 @@ import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList; import org.apache.hadoop.hdds.scm.exceptions.SCMException; import org.apache.hadoop.hdds.scm.net.NetworkTopology; +import org.apache.hadoop.hdds.scm.net.Node; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier; import org.apache.hadoop.hdds.utils.db.cache.CacheKey; @@ -210,18 +211,23 @@ protected List allocateBlock( final NetworkTopology clusterMap = shouldSortDatanodes && keyManager.isSortDatanodesForWriteEnabled() ? ozoneManager.getClusterMapAllowNull() : null; + // Resolve the client once per request, before the block loop, so the + // DNS-to-switch mapping is consulted at most once. + final Node omClient = clusterMap != null && !remoteAddress.isEmpty() + ? keyManager.resolveClientForWrite(remoteAddress, clusterMap) : null; if (!shouldSortDatanodes) { scmClientMachine = ""; omClientMachine = ""; sortedByNodes = null; - } else if (clusterMap != null && !remoteAddress.isEmpty()) { + } else if (omClient != null) { // Sort in OM: SCM skips sorting (empty machine), OM sorts by remoteAddress. scmClientMachine = ""; omClientMachine = remoteAddress; sortedByNodes = new HashMap<>(); } else { - // Sort in SCM (or keep order when remoteAddress is empty, since SCM skips - // sorting for an empty client machine). + // Sort in SCM when OM has no topology or cannot resolve the client (or + // keep order when remoteAddress is empty, since SCM skips sorting for an + // empty client machine). scmClientMachine = remoteAddress; omClientMachine = ""; sortedByNodes = null; @@ -250,7 +256,7 @@ protected List allocateBlock( .map(DatanodeDetails::getUuidString).collect(Collectors.toSet()); List sorted = sortedByNodes.get(uuidSet); if (sorted == null) { - sorted = keyManager.sortDatanodesForWrite(nodes, omClientMachine, clusterMap); + sorted = keyManager.sortDatanodesForWrite(nodes, omClientMachine, omClient, clusterMap); // A skipped sort returns null and is not cached: that pipeline keeps // its own order, which must not be reused for another pipeline with // the same node set. diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java index 784728ab3701..4b9b30e873c1 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java @@ -50,6 +50,7 @@ import org.apache.hadoop.hdds.scm.container.common.helpers.AllocatedBlock; import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList; import org.apache.hadoop.hdds.scm.net.NetworkTopology; +import org.apache.hadoop.hdds.scm.net.Node; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.hdds.scm.pipeline.PipelineID; import org.apache.hadoop.ipc_.Server; @@ -273,7 +274,8 @@ public void testAllocateBlockSendsClientMachineToScmWhenFlagOff() throws Excepti verify(scmBlockLocationProtocol).allocateBlock(anyLong(), anyInt(), any(), any(), any(), clientMachine.capture()); assertEquals("1.2.3.4", clientMachine.getValue()); - verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any()); + verify(mockKeyManager, never()).resolveClientForWrite(anyString(), any()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any(), any()); } @Test @@ -282,7 +284,8 @@ public void testAllocateBlockDoesNotSendClientMachineToScm() throws Exception { // clientMachine even when the client requests sorted datanodes. KeyManager mockKeyManager = mock(KeyManager.class); when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); - when(mockKeyManager.sortDatanodesForWrite(any(), any(), any())) + when(mockKeyManager.resolveClientForWrite(anyString(), any())).thenReturn(mock(Node.class)); + when(mockKeyManager.sortDatanodesForWrite(any(), any(), any(), any())) .thenAnswer(inv -> inv.getArgument(0)); when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); @@ -311,7 +314,111 @@ public void testAllocateBlockFallsBackToScmWhenTopologyUnavailable() throws Exce verify(scmBlockLocationProtocol).allocateBlock(anyLong(), anyInt(), any(), any(), any(), clientMachine.capture()); assertEquals("1.2.3.4", clientMachine.getValue()); - verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any()); + verify(mockKeyManager, never()).resolveClientForWrite(anyString(), any()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any(), any()); + } + + @Test + public void testAllocateBlockFallsBackToScmWhenClientUnresolved() throws Exception { + // Flag on and topology available, but OM cannot place the client in its + // topology: OM must not sort, and SCM receives the client address so it can + // try with its own node manager and mapping. + List nodes = Arrays.asList( + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()); + List scmOrder = new ArrayList<>(nodes); + Collections.reverse(scmOrder); + Pipeline pipeline = Pipeline.newBuilder() + .setState(Pipeline.PipelineState.OPEN) + .setId(PipelineID.randomId()) + .setReplicationConfig( + StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE)) + .setNodes(nodes) + .setNodesInOrder(scmOrder) + .build(); + AllocatedBlock block = new AllocatedBlock.Builder().setPipeline(pipeline) + .setContainerBlockID(new ContainerBlockID(CONTAINER_ID, LOCAL_ID)).build(); + ArgumentCaptor clientMachine = ArgumentCaptor.forClass(String.class); + when(scmBlockLocationProtocol.allocateBlock(anyLong(), anyInt(), any(), + anyString(), any(ExcludeList.class), clientMachine.capture())) + .thenReturn(Collections.singletonList(block)); + + KeyManager mockKeyManager = mock(KeyManager.class); + when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); + when(mockKeyManager.resolveClientForWrite(anyString(), any())).thenReturn(null); + when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); + when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); + + OMAllocateBlockRequest request = + getOmAllocateBlockRequest(createAllocateBlockRequest()); + List locations = request.allocateBlock(replicationConfig, + new ExcludeList(), scmBlockSize, true, + UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(), ozoneManager); + + assertEquals("1.2.3.4", clientMachine.getValue()); + verify(mockKeyManager, times(1)).resolveClientForWrite(eq("1.2.3.4"), any()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any(), any()); + assertEquals(1, locations.size()); + // SCM's order is kept as-is. + assertEquals(scmOrder, locations.get(0).getPipeline().getNodesInOrder()); + } + + @Test + public void testAllocateBlockResolvesClientOnceAcrossPipelines() throws Exception { + // Two blocks on two different pipelines (different datanode sets): the + // client is resolved once for the request and reused for both sorts. + List nodes1 = Arrays.asList( + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()); + List nodes2 = Arrays.asList( + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()); + Pipeline pipeline1 = Pipeline.newBuilder() + .setState(Pipeline.PipelineState.OPEN) + .setId(PipelineID.randomId()) + .setReplicationConfig( + StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE)) + .setNodes(nodes1) + .build(); + Pipeline pipeline2 = Pipeline.newBuilder() + .setState(Pipeline.PipelineState.OPEN) + .setId(PipelineID.randomId()) + .setReplicationConfig( + StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE)) + .setNodes(nodes2) + .build(); + AllocatedBlock block1 = new AllocatedBlock.Builder().setPipeline(pipeline1) + .setContainerBlockID(new ContainerBlockID(CONTAINER_ID, LOCAL_ID)).build(); + AllocatedBlock block2 = new AllocatedBlock.Builder().setPipeline(pipeline2) + .setContainerBlockID(new ContainerBlockID(CONTAINER_ID + 1, LOCAL_ID + 1)).build(); + ArgumentCaptor clientMachine = ArgumentCaptor.forClass(String.class); + when(scmBlockLocationProtocol.allocateBlock(anyLong(), anyInt(), any(), + anyString(), any(ExcludeList.class), clientMachine.capture())) + .thenReturn(Arrays.asList(block1, block2)); + + Node client = mock(Node.class); + KeyManager mockKeyManager = mock(KeyManager.class); + when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); + when(mockKeyManager.resolveClientForWrite(eq("1.2.3.4"), any())).thenReturn(client); + when(mockKeyManager.sortDatanodesForWrite(any(), any(), any(), any())) + .thenAnswer(inv -> inv.getArgument(0)); + when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); + when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); + + OMAllocateBlockRequest request = + getOmAllocateBlockRequest(createAllocateBlockRequest()); + List locations = request.allocateBlock(replicationConfig, + new ExcludeList(), 2 * scmBlockSize, true, + UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(), ozoneManager); + + assertEquals(2, locations.size()); + // OM sorts, so SCM is told not to. + assertEquals("", clientMachine.getValue()); + verify(mockKeyManager, times(1)).resolveClientForWrite(eq("1.2.3.4"), any()); + verify(mockKeyManager, times(2)).sortDatanodesForWrite(any(), eq("1.2.3.4"), eq(client), any()); } @Test @@ -345,9 +452,11 @@ public void testAllocateBlockSortsSharedPipelineOnce() throws Exception { List sortedOrder = new ArrayList<>(nodes); Collections.reverse(sortedOrder); + Node client = mock(Node.class); KeyManager mockKeyManager = mock(KeyManager.class); when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); - when(mockKeyManager.sortDatanodesForWrite(any(), any(), any())) + when(mockKeyManager.resolveClientForWrite(eq("1.2.3.4"), any())).thenReturn(client); + when(mockKeyManager.sortDatanodesForWrite(any(), any(), any(), any())) .thenAnswer(inv -> sortedOrder); when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); @@ -360,7 +469,8 @@ public void testAllocateBlockSortsSharedPipelineOnce() throws Exception { UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(), ozoneManager); // Sorted once for the shared pipeline... - verify(mockKeyManager, times(1)).sortDatanodesForWrite(any(), eq("1.2.3.4"), any()); + verify(mockKeyManager, times(1)).resolveClientForWrite(eq("1.2.3.4"), any()); + verify(mockKeyManager, times(1)).sortDatanodesForWrite(any(), eq("1.2.3.4"), eq(client), any()); // ...and the sorted order is applied to every block's pipeline. assertEquals(2, locations.size()); for (OmKeyLocationInfo location : locations) { @@ -411,7 +521,8 @@ public void testAllocateBlockKeepsPerPipelineOrderWhenSortSkipped() throws Excep KeyManager mockKeyManager = mock(KeyManager.class); when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); // Skip the sort: null signals that no sort happened. - when(mockKeyManager.sortDatanodesForWrite(any(), any(), any())).thenReturn(null); + when(mockKeyManager.resolveClientForWrite(anyString(), any())).thenReturn(mock(Node.class)); + when(mockKeyManager.sortDatanodesForWrite(any(), any(), any(), any())).thenReturn(null); when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); @@ -463,7 +574,8 @@ public void testAllocateBlockKeepsOrderWhenRemoteAddressEmpty() throws Exception UserInfo.newBuilder().setRemoteAddress("").build(), ozoneManager); assertEquals("", clientMachine.getValue()); - verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any()); + verify(mockKeyManager, never()).resolveClientForWrite(anyString(), any()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any(), any()); assertEquals(1, locations.size()); // Assert the write order (nodesInOrder), which copyWithNodesInOrder would // have changed had OM sorted; it must stay as the original pipeline order. @@ -477,7 +589,9 @@ public void sortDatanodesForWriteRequiresClientMachine() { MockDatanodeDetails.randomDatanodeDetails(), MockDatanodeDetails.randomDatanodeDetails()); assertThrows(IllegalArgumentException.class, - () -> keyManager.sortDatanodesForWrite(nodes, "", mock(NetworkTopology.class))); + () -> keyManager.sortDatanodesForWrite(nodes, "", mock(Node.class), mock(NetworkTopology.class))); + assertThrows(IllegalArgumentException.class, + () -> keyManager.resolveClientForWrite("", mock(NetworkTopology.class))); } // Like createAllocateBlockRequest, but sets sortDatanodes so preExecute From 6e673a5ad7bd8cf926d8fdb002e1a989827a2003 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Sat, 12 Sep 2026 20:52:06 +0800 Subject: [PATCH 3/6] HDDS-15941. Fix stale read-path comment and time client resolution under the sort metric --- .../java/org/apache/hadoop/ozone/om/KeyManagerImpl.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java index 63522bde9b5a..bf67b9e2fcee 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java @@ -2286,8 +2286,8 @@ private void sortDatanodes(String clientMachine, List keyInfos) { // Cache only a freshly sorted order, not an input list returned // unchanged when no sort happens: that order is per-pipeline and must // not be reused for another pipeline with the same node set. The read - // sort always returns a new list, so this never skips caching here; it - // keeps the pattern identical to the write path. + // sort always returns a new list, so this never skips caching here; the + // write path uses a null contract for a skipped sort instead. if (sortedNodes != null && sortedNodes != nodes) { sortedPipelines.put(uuidSet, sortedNodes); } @@ -2316,7 +2316,8 @@ public List sortDatanodes(List getOtherNode(clientMachine, clusterMap)); } @Override From d1d0c614f8c3d39765f7cdc88da4d98d88aaf224 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Sat, 12 Sep 2026 21:13:04 +0800 Subject: [PATCH 4/6] HDDS-15941. Trim test comments and the resolveClientForWrite javadoc --- .../org/apache/hadoop/ozone/TestOMSortDatanodes.java | 11 ++--------- .../java/org/apache/hadoop/ozone/om/KeyManager.java | 3 +-- .../om/request/key/TestOMAllocateBlockRequest.java | 6 ++---- 3 files changed, 5 insertions(+), 15 deletions(-) diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java index 9f02df65349b..91e1dac361cd 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java @@ -239,12 +239,8 @@ public void resolveClientForWriteAttachesNonDatanodeClientToItsRack() { @Test public void sortDatanodesForWritePrefersPipelineDatanodeAsClient() { - // The pre-resolved client is rack-level and equidistant to every datanode - // in its rack, so with two same-rack datanodes in the pipeline only the - // datanode identity can put the client's own datanode first. To make the - // test deterministic (no equidistant shuffle), pass a client resolved for - // the OTHER rack: without the pipeline match, an other-rack datanode would - // sort first; with it, the client datanode must. + // Two same-rack datanodes in the pipeline: only the datanode identity can put the client's own first. + // Resolve the client in the other rack so preferring the matching pipeline datanode is deterministic. List all = nodeManager.getAllNodes(); DatanodeDetails clientDn = all.get(0); DatanodeDetails sameRack = null; @@ -295,9 +291,6 @@ public void sortDatanodesForWritePrefersPipelineDatanodeAsClient() { @Test public void resolveClientForWriteIsTheOnlyMappingLookup() { - // Resolve an edge client once, then sort several different datanode sets: - // the DNS-to-switch mapping must be consulted exactly once, by - // resolveClientForWrite, never by sortDatanodesForWrite. List all = nodeManager.getAllNodes(); CountingStaticMapping.RESOLVED_NAMES.set(0); Node client = keyManager.resolveClientForWrite("edge0", om.getClusterMap()); diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java index ab95f57d8315..48afa20fa7fc 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java @@ -404,8 +404,7 @@ DeleteKeysResult getPendingDeletionSubFiles(long volumeId, long bucketId, OmKeyI /** * Resolve the streaming-write client to a node in OM's cached cluster map - * through the DNS-to-switch mapping. Called once per allocateBlock request, - * before the block loop, so the mapping is consulted at most once. + * through the DNS-to-switch mapping. * * @param clientMachine client address (IP or hostname), must not be empty * @param clusterMap OM's cached cluster map diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java index 4b9b30e873c1..8169e1dcf309 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java @@ -350,8 +350,7 @@ public void testAllocateBlockFallsBackToScmWhenClientUnresolved() throws Excepti when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); - OMAllocateBlockRequest request = - getOmAllocateBlockRequest(createAllocateBlockRequest()); + OMAllocateBlockRequest request = getOmAllocateBlockRequest(createAllocateBlockRequest()); List locations = request.allocateBlock(replicationConfig, new ExcludeList(), scmBlockSize, true, UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(), ozoneManager); @@ -408,8 +407,7 @@ public void testAllocateBlockResolvesClientOnceAcrossPipelines() throws Exceptio when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); - OMAllocateBlockRequest request = - getOmAllocateBlockRequest(createAllocateBlockRequest()); + OMAllocateBlockRequest request = getOmAllocateBlockRequest(createAllocateBlockRequest()); List locations = request.allocateBlock(replicationConfig, new ExcludeList(), 2 * scmBlockSize, true, UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(), ozoneManager); From 0ca26f47d551b219253c1b7d5e3a58e8a44c3aba Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Sun, 13 Sep 2026 08:22:25 +0800 Subject: [PATCH 5/6] HDDS-15941. Clarify write sorting contracts and simplify test comments --- .../hadoop/ozone/TestOMSortDatanodes.java | 7 ++----- .../org/apache/hadoop/ozone/om/KeyManager.java | 17 ++++++----------- .../ozone/om/request/key/OMKeyRequest.java | 4 +--- .../request/key/TestOMAllocateBlockRequest.java | 12 ++---------- 4 files changed, 11 insertions(+), 29 deletions(-) diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java index 91e1dac361cd..11c5be13b143 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java @@ -239,7 +239,6 @@ public void resolveClientForWriteAttachesNonDatanodeClientToItsRack() { @Test public void sortDatanodesForWritePrefersPipelineDatanodeAsClient() { - // Two same-rack datanodes in the pipeline: only the datanode identity can put the client's own first. // Resolve the client in the other rack so preferring the matching pipeline datanode is deterministic. List all = nodeManager.getAllNodes(); DatanodeDetails clientDn = all.get(0); @@ -265,8 +264,7 @@ public void sortDatanodesForWritePrefersPipelineDatanodeAsClient() { assertNotEquals(clientDn.getNetworkLocation(), otherRackClient.getNetworkLocation()); for (String address : new String[] {clientDn.getIpAddress(), clientDn.getHostName()}) { - // Pipeline without the client datanode: the passed client decides, so an - // other-rack datanode sorts first. + // Without a matching datanode, sort relative to the supplied client. List without = new ArrayList<>(); without.add(rpcCopy(sameRack)); without.add(rpcCopy(otherRack.get(0))); @@ -276,8 +274,7 @@ public void sortDatanodesForWritePrefersPipelineDatanodeAsClient() { assertNotNull(sortedWithout); assertEquals(otherRackClient.getNetworkLocation(), sortedWithout.get(0).getNetworkLocation()); - // Pipeline containing the client datanode: its identity wins over the - // passed other-rack client, so it sorts first (distance zero). + // A matching datanode takes precedence over the supplied client and sorts first. List with = new ArrayList<>(); with.add(rpcCopy(otherRack.get(0))); with.add(rpcCopy(sameRack)); diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java index 48afa20fa7fc..79f74a7f9812 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java @@ -403,29 +403,24 @@ DeleteKeysResult getPendingDeletionSubFiles(long volumeId, long bucketId, OmKeyI KeyLifecycleService getKeyLifecycleService(); /** - * Resolve the streaming-write client to a node in OM's cached cluster map - * through the DNS-to-switch mapping. + * Resolve the streaming-write client's rack in OM's cached cluster map using the DNS-to-switch mapping. * * @param clientMachine client address (IP or hostname), must not be empty * @param clusterMap OM's cached cluster map - * @return a node attached to the client's rack, or null when the client - * cannot be placed in the topology; the caller then leaves the sort to SCM + * @return a node attached to the client's rack, or null if unresolved so the caller can let SCM sort */ Node resolveClientForWrite(String clientMachine, NetworkTopology clusterMap); /** - * Sort the datanodes of a write pipeline by network-topology distance to the - * client, using OM's locally cached cluster map. When the client is one of the - * pipeline datanodes, that datanode is used as the client so it sorts first; - * otherwise {@code client} from {@link #resolveClientForWrite} is used. + * Sort write pipeline datanodes nearest-first using OM's cached cluster map. A datanode matching + * {@code clientMachine} takes precedence over {@code client} and sorts first. * * @param nodes the pipeline nodes to sort * @param clientMachine client address (IP or hostname), must not be empty * @param client the node returned by {@link #resolveClientForWrite}, must not be null * @param clusterMap OM's cached cluster map used to resolve topology distance - * @return nodes sorted nearest-first, or null when the sort is skipped because - * a pipeline node is missing from the cluster map; callers must leave the - * pipeline order unchanged in that case + * @return nodes sorted nearest-first, or null if a pipeline node is missing from the cluster map; + * callers must preserve the pipeline's order in that case */ List sortDatanodesForWrite( List nodes, String clientMachine, Node client, NetworkTopology clusterMap); diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java index 8d8c28a636b6..ea51773a6800 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java @@ -257,9 +257,7 @@ protected List allocateBlock( List sorted = sortedByNodes.get(uuidSet); if (sorted == null) { sorted = keyManager.sortDatanodesForWrite(nodes, omClientMachine, omClient, clusterMap); - // A skipped sort returns null and is not cached: that pipeline keeps - // its own order, which must not be reused for another pipeline with - // the same node set. + // Cache only sorted results; a skipped sort must preserve each pipeline's own order. if (sorted != null) { sortedByNodes.put(uuidSet, sorted); } diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java index 8169e1dcf309..3e40c790ec5b 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java @@ -478,18 +478,13 @@ public void testAllocateBlockSortsSharedPipelineOnce() throws Exception { @Test public void testAllocateBlockKeepsPerPipelineOrderWhenSortSkipped() throws Exception { - // Two pipelines share the same datanode set but in a different order. When - // the sort is skipped (sortDatanodesForWrite returns null), each pipeline - // must keep its own order: null must not be cached under the node set and - // reused for the other pipeline. + // Pipelines sharing a datanode set must keep their own order when sorting is skipped. DatanodeDetails a = MockDatanodeDetails.randomDatanodeDetails(); DatanodeDetails b = MockDatanodeDetails.randomDatanodeDetails(); DatanodeDetails c = MockDatanodeDetails.randomDatanodeDetails(); List nodes1 = Arrays.asList(a, b, c); List nodes2 = Arrays.asList(c, b, a); - // nodesInOrder deliberately differs from getNodes() so the test catches an - // implementation that overwrites an existing order with getNodes() when the - // sort is skipped. + // Use a different nodesInOrder to catch a skipped sort overwriting it with getNodes(). List inOrder1 = Arrays.asList(b, c, a); List inOrder2 = Arrays.asList(a, c, b); Pipeline pipeline1 = Pipeline.newBuilder() @@ -518,7 +513,6 @@ public void testAllocateBlockKeepsPerPipelineOrderWhenSortSkipped() throws Excep KeyManager mockKeyManager = mock(KeyManager.class); when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); - // Skip the sort: null signals that no sort happened. when(mockKeyManager.resolveClientForWrite(anyString(), any())).thenReturn(mock(Node.class)); when(mockKeyManager.sortDatanodesForWrite(any(), any(), any(), any())).thenReturn(null); when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); @@ -531,8 +525,6 @@ public void testAllocateBlockKeepsPerPipelineOrderWhenSortSkipped() throws Excep UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(), ozoneManager); assertEquals(2, locations.size()); - // Each pipeline keeps its own existing nodesInOrder; the skipped sort is - // neither shared nor replaced by getNodes(). assertEquals(inOrder1, locations.get(0).getPipeline().getNodesInOrder()); assertEquals(inOrder2, locations.get(1).getPipeline().getNodesInOrder()); } From 6cecf01ec71bce1c8408de2c20fcd7fbdc6787ac Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Sun, 13 Sep 2026 08:46:19 +0800 Subject: [PATCH 6/6] HDDS-15941. Simplify write sort selection and clarify fallback behavior --- .../org/apache/hadoop/ozone/om/OmConfig.java | 3 +- .../hadoop/ozone/om/KeyManagerImpl.java | 5 --- .../hadoop/ozone/om/OMPerformanceMetrics.java | 2 +- .../ozone/om/request/key/OMKeyRequest.java | 33 +++++-------------- .../key/TestOMAllocateBlockRequest.java | 7 ++++ 5 files changed, 19 insertions(+), 31 deletions(-) diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OmConfig.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OmConfig.java index c981afd414d8..33e1b511ca04 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OmConfig.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OmConfig.java @@ -192,7 +192,8 @@ public class OmConfig extends ReconfigurableConfig { "datanode first) locally using its cached cluster topology, instead " + "of asking SCM to sort on every allocateBlock. Defaults to false so " + "SCM performs the sort. Enable this to offload the sort from SCM " + - "when multiple OM services share a single SCM service." + "when multiple OM services share a single SCM service. SCM still sorts if OM has no cached topology " + + "or cannot resolve the client. If OM's topology lacks a pipeline datanode, the pipeline order is preserved." ) private boolean sortDatanodesForWriteEnabled; diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java index bf67b9e2fcee..3547b9803788 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java @@ -2283,11 +2283,6 @@ private void sortDatanodes(String clientMachine, List keyInfos) { List sortedNodes = sortedPipelines.get(uuidSet); if (sortedNodes == null) { sortedNodes = sortDatanodes(nodes, clientMachine); - // Cache only a freshly sorted order, not an input list returned - // unchanged when no sort happens: that order is per-pipeline and must - // not be reused for another pipeline with the same node set. The read - // sort always returns a new list, so this never skips caching here; the - // write path uses a null contract for a skipped sort instead. if (sortedNodes != null && sortedNodes != nodes) { sortedPipelines.put(uuidSet, sortedNodes); } diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMPerformanceMetrics.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMPerformanceMetrics.java index da687ae193c4..68b4c5929eca 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMPerformanceMetrics.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMPerformanceMetrics.java @@ -138,7 +138,7 @@ public OMPerformanceMetrics() { getKeyInfoSortDatanodesLatencyNs = stat("GetKeyInfoSortDatanodesLatencyNs", "Sort datanodes latency in getKeyInfo"); allocateBlockSortDatanodesLatencyNs = stat("AllocateBlockSortDatanodesLatencyNs", - "Sort datanodes latency in allocateBlock (streaming write)"); + "Client lookup and datanode sort latency in allocateBlock (streaming write), recorded as separate samples"); getKeyInfoResolveBucketLatencyNs = stat("GetKeyInfoResolveBucketLatencyNs", "resolveBucketLink latency in getKeyInfo"); s3VolumeContextLatencyNs = stat("S3VolumeContextLatencyNs", diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java index ea51773a6800..4f89f982d69c 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java @@ -201,12 +201,6 @@ protected List allocateBlock( final int numBlocks = (int) Math.min(ozoneManager.getPreallocateBlocksMax(), (requestedSize - 1) / (scmBlockSize * dataGroupSize) + 1); - final String scmClientMachine; - final String omClientMachine; - // Sorted order cached by datanode set so blocks whose pipelines share the - // same datanodes are sorted once (mirrors the read path's caching). Keyed by - // the UUID set so it is order-insensitive and dedups across pipelines. - final Map, List> sortedByNodes; final String remoteAddress = userInfo.getRemoteAddress(); final NetworkTopology clusterMap = shouldSortDatanodes && keyManager.isSortDatanodesForWriteEnabled() @@ -215,23 +209,14 @@ protected List allocateBlock( // DNS-to-switch mapping is consulted at most once. final Node omClient = clusterMap != null && !remoteAddress.isEmpty() ? keyManager.resolveClientForWrite(remoteAddress, clusterMap) : null; - if (!shouldSortDatanodes) { - scmClientMachine = ""; - omClientMachine = ""; - sortedByNodes = null; - } else if (omClient != null) { - // Sort in OM: SCM skips sorting (empty machine), OM sorts by remoteAddress. - scmClientMachine = ""; - omClientMachine = remoteAddress; - sortedByNodes = new HashMap<>(); - } else { - // Sort in SCM when OM has no topology or cannot resolve the client (or - // keep order when remoteAddress is empty, since SCM skips sorting for an - // empty client machine). - scmClientMachine = remoteAddress; - omClientMachine = ""; - sortedByNodes = null; - } + // Let SCM sort when OM sorting is disabled or its topology/client lookup is unavailable. + // An empty client address tells SCM to skip sorting. + final String scmClientMachine = shouldSortDatanodes && omClient == null ? remoteAddress : ""; + // Sorted order cached by datanode set so blocks whose pipelines share the + // same datanodes are sorted once (mirrors the read path's caching). Keyed by + // the UUID set so it is order-insensitive and dedups across pipelines. + final Map, List> sortedByNodes = + omClient != null ? new HashMap<>() : null; List locationInfos = new ArrayList<>(numBlocks); String remoteUser = getRemoteUser().getShortUserName(); @@ -256,7 +241,7 @@ protected List allocateBlock( .map(DatanodeDetails::getUuidString).collect(Collectors.toSet()); List sorted = sortedByNodes.get(uuidSet); if (sorted == null) { - sorted = keyManager.sortDatanodesForWrite(nodes, omClientMachine, omClient, clusterMap); + sorted = keyManager.sortDatanodesForWrite(nodes, remoteAddress, omClient, clusterMap); // Cache only sorted results; a skipped sort must preserve each pipeline's own order. if (sorted != null) { sortedByNodes.put(uuidSet, sorted); diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java index 3e40c790ec5b..186a808b1c6b 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java @@ -584,6 +584,13 @@ public void sortDatanodesForWriteRequiresClientMachine() { () -> keyManager.resolveClientForWrite("", mock(NetworkTopology.class))); } + @Test + public void sortDatanodesForWriteRequiresClient() { + NullPointerException exception = assertThrows(NullPointerException.class, + () -> keyManager.sortDatanodesForWrite(Collections.emptyList(), "1.2.3.4", null, mock(NetworkTopology.class))); + assertEquals("client is null", exception.getMessage()); + } + // Like createAllocateBlockRequest, but sets sortDatanodes so preExecute // resolves the client address from the RPC context. private OMRequest createAllocateBlockRequestWithSort() {