From a767f7ba0d994e078a6a23cd4321ff05c6a79df5 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Mon, 29 Jun 2026 21:54:17 +0800 Subject: [PATCH 01/29] HDDS-15059. Add OM-side sortDatanodesForWrite with latency metric --- .../hadoop/ozone/TestOMSortDatanodes.java | 23 +++++++++++++++++++ .../apache/hadoop/ozone/om/KeyManager.java | 14 +++++++++++ .../hadoop/ozone/om/KeyManagerImpl.java | 16 +++++++++++++ .../hadoop/ozone/om/OMPerformanceMetrics.java | 7 ++++++ 4 files changed, 60 insertions(+) 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 cfce524537fc..dcdd176a0f2d 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 @@ -176,6 +176,29 @@ private static void assertRackOrder(String rack, List } } + @Test + public void sortDatanodesForWritePicksNearestFirst() { + for (DatanodeDetails dn : nodeManager.getAllNodes()) { + List sorted = + keyManager.sortDatanodesForWrite(nodeManager.getAllNodes(), nodeAddress(dn)); + assertEquals(dn, sorted.get(0), + "Source node should be sorted first for writes"); + assertRackOrder(dn.getNetworkLocation(), sorted); + } + } + + @Test + public void sortDatanodesForWriteKeepsOrderWhenClientUnresolved() { + 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); + assertEquals(original, result, + "Write pipeline order must be preserved when client is unresolved"); + } + private String nodeAddress(DatanodeDetails dn) { boolean useHostname = config.getBoolean( HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME, 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 fdf4172c71b2..0a561a277725 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 @@ -23,6 +23,7 @@ import java.util.List; import java.util.Map; import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.utils.BackgroundService; import org.apache.hadoop.hdds.utils.db.Table; import org.apache.hadoop.hdds.utils.db.TableIterator; @@ -363,4 +364,17 @@ DeleteKeysResult getPendingDeletionSubFiles(long volumeId, long bucketId, OmKeyI * @return BackgroundService */ CompactionService getCompactionService(); + + /** + * 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. + * + * @param nodes the pipeline nodes to sort + * @param clientMachine client address (IP or hostname) + * @return nodes sorted nearest-first, or the input order if client is unresolved + */ + List sortDatanodesForWrite( + List nodes, String clientMachine); } 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 08b6d6abbf18..2f01ca2d97e0 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 @@ -2209,6 +2209,22 @@ public List sortDatanodes(List sortDatanodesForWrite( + List nodes, String clientMachine) { + return captureLatencyNs( + metrics.getAllocateBlockSortDatanodesLatencyNs(), () -> { + final Node client = getClientNode(clientMachine, nodes); + 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; + } + return ozoneManager.getClusterMap() + .sortByDistanceCost(client, nodes, nodes.size()); + }); + } + private Node getClientNode(String clientMachine, List nodes) { List matchingNodes = new ArrayList<>(); 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 e142e4cfa715..b38f42b3e03b 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 @@ -67,6 +67,9 @@ public class OMPerformanceMetrics { @Metric(about = "Sort datanodes latency in getKeyInfo") private MutableRate getKeyInfoSortDatanodesLatencyNs; + @Metric(about = "Sort datanodes latency in allocateBlock (streaming write)") + private MutableRate allocateBlockSortDatanodesLatencyNs; + @Metric(about = "resolveBucketLink latency in getKeyInfo") private MutableRate getKeyInfoResolveBucketLatencyNs; @@ -246,6 +249,10 @@ MutableRate getGetKeyInfoSortDatanodesLatencyNs() { return getKeyInfoSortDatanodesLatencyNs; } + public MutableRate getAllocateBlockSortDatanodesLatencyNs() { + return allocateBlockSortDatanodesLatencyNs; + } + public void setForceContainerCacheRefresh(boolean value) { forceContainerCacheRefresh.add(value ? 1L : 0L); } From 35ad85cbbf9591341376e3dbb3c4dcc98a95ab18 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Mon, 29 Jun 2026 22:05:59 +0800 Subject: [PATCH 02/29] HDDS-15059. Sort write pipeline on OM instead of SCM during allocateBlock --- .../om/request/file/OMFileCreateRequest.java | 2 +- .../request/key/OMAllocateBlockRequest.java | 2 +- .../om/request/key/OMKeyCreateRequest.java | 2 +- .../ozone/om/request/key/OMKeyRequest.java | 18 ++++++-- .../key/TestOMAllocateBlockRequest.java | 44 +++++++++++++++++++ 5 files changed, 62 insertions(+), 6 deletions(-) diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java index 9788cfbafe17..2fde651759d3 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java @@ -138,7 +138,7 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException { ozoneManager.getOMServiceId(), ozoneManager.getMetrics(), keyArgs.getSortDatanodes(), - userInfo); + userInfo, ozoneManager.getKeyManager()); KeyArgs.Builder newKeyArgs = keyArgs.toBuilder() .setModificationTime(Time.now()).setType(type).setFactor(factor) diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequest.java index d7cd3bea5ecb..1dff8fadc334 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequest.java @@ -118,7 +118,7 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException { ozoneManager.getPreallocateBlocksMax(), ozoneManager.isGrpcBlockTokenEnabled(), ozoneManager.getOMServiceId(), ozoneManager.getMetrics(), - keyArgs.getSortDatanodes(), userInfo); + keyArgs.getSortDatanodes(), userInfo, ozoneManager.getKeyManager()); // Set modification time and normalize key if required. KeyArgs.Builder newKeyArgs = diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java index d7b14455369f..00de78b98dfc 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java @@ -170,7 +170,7 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException { ozoneManager.getOMServiceId(), ozoneManager.getMetrics(), keyArgs.getSortDatanodes(), - userInfo)); + userInfo, ozoneManager.getKeyManager())); effectiveDataSize = requestedSize; } 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 7967007263e7..18ffca85241b 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 @@ -55,10 +55,12 @@ import org.apache.hadoop.hdds.client.ContainerBlockID; import org.apache.hadoop.hdds.client.ECReplicationConfig; import org.apache.hadoop.hdds.client.ReplicationConfig; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; 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.exceptions.SCMException; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.hdds.security.token.OzoneBlockTokenSecretManager; import org.apache.hadoop.hdds.utils.db.cache.CacheKey; import org.apache.hadoop.hdds.utils.db.cache.CacheValue; @@ -66,6 +68,7 @@ import org.apache.hadoop.ozone.OmUtils; import org.apache.hadoop.ozone.OzoneAcl; import org.apache.hadoop.ozone.OzoneConsts; +import org.apache.hadoop.ozone.om.KeyManager; import org.apache.hadoop.ozone.om.OMMetadataManager; import org.apache.hadoop.ozone.om.OMMetrics; import org.apache.hadoop.ozone.om.OmConfig; @@ -190,7 +193,7 @@ protected List< OmKeyLocationInfo > allocateBlock(ScmClient scmClient, ReplicationConfig replicationConfig, ExcludeList excludeList, long requestedSize, long scmBlockSize, int preallocateBlocksMax, boolean grpcBlockTokenEnabled, String serviceID, OMMetrics omMetrics, - boolean shouldSortDatanodes, UserInfo userInfo) + boolean shouldSortDatanodes, UserInfo userInfo, KeyManager keyManager) throws IOException { int dataGroupSize = replicationConfig instanceof ECReplicationConfig ? ((ECReplicationConfig) replicationConfig).getData() : 1; @@ -208,7 +211,7 @@ protected List< OmKeyLocationInfo > allocateBlock(ScmClient scmClient, try { allocatedBlocks = scmClient.getBlockClient() .allocateBlock(scmBlockSize, numBlocks, replicationConfig, serviceID, - excludeList, clientMachine); + excludeList, ""); } catch (SCMException ex) { omMetrics.incNumBlockAllocateCallFails(); if (ex.getResult() @@ -220,11 +223,20 @@ protected List< OmKeyLocationInfo > allocateBlock(ScmClient scmClient, } for (AllocatedBlock allocatedBlock : allocatedBlocks) { BlockID blockID = new BlockID(allocatedBlock.getBlockID()); + Pipeline pipeline = allocatedBlock.getPipeline(); + if (shouldSortDatanodes) { + final List nodes = pipeline.getNodes(); + final List sorted = + keyManager.sortDatanodesForWrite(nodes, clientMachine); + if (!Objects.equals(sorted, pipeline.getNodesInOrder())) { + pipeline = pipeline.copyWithNodesInOrder(sorted); + } + } OmKeyLocationInfo.Builder builder = new OmKeyLocationInfo.Builder() .setBlockID(blockID) .setLength(scmBlockSize) .setOffset(0) - .setPipeline(allocatedBlock.getPipeline()); + .setPipeline(pipeline); if (grpcBlockTokenEnabled) { builder.setToken(secretManager.generateToken(remoteUser, blockID, EnumSet.of(READ, WRITE), scmBlockSize)); 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 1318e5b0645f..f91d15a03dc5 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 @@ -21,12 +21,19 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import jakarta.annotation.Nonnull; import java.util.List; import java.util.UUID; import org.apache.hadoop.hdds.client.RatisReplicationConfig; import org.apache.hadoop.ozone.OzoneConsts; +import org.apache.hadoop.ozone.om.KeyManager; import org.apache.hadoop.ozone.om.helpers.BucketLayout; import org.apache.hadoop.ozone.om.helpers.OmKeyInfo; import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo; @@ -36,7 +43,9 @@ import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.AllocateBlockRequest; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyArgs; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.UserInfo; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; /** * Tests OMAllocateBlockRequest class. @@ -225,6 +234,41 @@ protected OMRequest doPreExecute(OMRequest originalOMRequest) return modifiedOmRequest; } + @Test + public void testAllocateBlockDoesNotSendClientMachineToScm() throws Exception { + // OM now sorts the write pipeline locally, so SCM must receive an empty + // clientMachine even when the client requests sorted datanodes. + KeyManager mockKeyManager = mock(KeyManager.class); + when(mockKeyManager.sortDatanodesForWrite(any(), any())) + .thenAnswer(inv -> inv.getArgument(0)); + when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); + + OMAllocateBlockRequest request = + getOmAllocateBlockRequest(createAllocateBlockRequestWithSort("1.2.3.4")); + request.preExecute(ozoneManager); + + ArgumentCaptor clientMachine = ArgumentCaptor.forClass(String.class); + verify(scmBlockLocationProtocol).allocateBlock(anyLong(), anyInt(), any(), + any(), any(), clientMachine.capture()); + assertEquals("", clientMachine.getValue()); + } + + private OMRequest createAllocateBlockRequestWithSort(String clientAddress) { + KeyArgs keyArgs = KeyArgs.newBuilder() + .setVolumeName(volumeName).setBucketName(bucketName).setKeyName(keyName) + .setFactor(((RatisReplicationConfig) replicationConfig).getReplicationFactor()) + .setType(replicationConfig.getReplicationType()) + .setSortDatanodes(true) + .build(); + AllocateBlockRequest allocateBlockRequest = AllocateBlockRequest.newBuilder() + .setClientID(clientID).setKeyArgs(keyArgs).build(); + return OMRequest.newBuilder() + .setCmdType(OzoneManagerProtocolProtos.Type.AllocateBlock) + .setClientId(UUID.randomUUID().toString()) + .setUserInfo(UserInfo.newBuilder().setRemoteAddress(clientAddress).build()) + .setAllocateBlockRequest(allocateBlockRequest).build(); + } + protected OMRequest createAllocateBlockRequest() { KeyArgs keyArgs = KeyArgs.newBuilder() From c3ea84c1adca15d563ab5f29af06d34f79998e3c Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Mon, 29 Jun 2026 22:15:06 +0800 Subject: [PATCH 03/29] HDDS-15059. Address review: metric getter visibility and test helper comment --- .../java/org/apache/hadoop/ozone/om/OMPerformanceMetrics.java | 2 +- .../hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) 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 b38f42b3e03b..3f965ba025e6 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 @@ -249,7 +249,7 @@ MutableRate getGetKeyInfoSortDatanodesLatencyNs() { return getKeyInfoSortDatanodesLatencyNs; } - public MutableRate getAllocateBlockSortDatanodesLatencyNs() { + MutableRate getAllocateBlockSortDatanodesLatencyNs() { return allocateBlockSortDatanodesLatencyNs; } 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 f91d15a03dc5..e29c150c9670 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 @@ -253,6 +253,7 @@ public void testAllocateBlockDoesNotSendClientMachineToScm() throws Exception { assertEquals("", clientMachine.getValue()); } + // Like createAllocateBlockRequest, but sets sortDatanodes and a UserInfo remote address for OM-side sort. private OMRequest createAllocateBlockRequestWithSort(String clientAddress) { KeyArgs keyArgs = KeyArgs.newBuilder() .setVolumeName(volumeName).setBucketName(bucketName).setKeyName(keyName) From fa5a4665ad1660a59c694000321fe594fac8dc32 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Mon, 29 Jun 2026 23:21:09 +0800 Subject: [PATCH 04/29] HDDS-15059. Guard empty client and cache per-pipeline sort in allocateBlock --- .../hadoop/ozone/TestOMSortDatanodes.java | 11 +++++++++ .../hadoop/ozone/om/KeyManagerImpl.java | 5 ++++ .../ozone/om/request/key/OMKeyRequest.java | 10 ++++++-- .../key/TestOMAllocateBlockRequest.java | 24 +++++++++++++++++++ 4 files changed, 48 insertions(+), 2 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 dcdd176a0f2d..2c616c298a8e 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 @@ -187,6 +187,17 @@ public void sortDatanodesForWritePicksNearestFirst() { } } + @Test + public void sortDatanodesForWriteIgnoresEmptyClient() { + List nodes = nodeManager.getAllNodes(); + List original = new ArrayList<>(nodes); + // An empty client address must never reorder the write pipeline. + List result = + keyManager.sortDatanodesForWrite(nodes, ""); + assertEquals(original, result, + "Empty client must leave the write pipeline order unchanged"); + } + @Test public void sortDatanodesForWriteKeepsOrderWhenClientUnresolved() { List nodes = nodeManager.getAllNodes(); 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 2f01ca2d97e0..fe4d1f1afa45 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 @@ -2212,6 +2212,11 @@ public List sortDatanodes(List sortDatanodesForWrite( List nodes, String clientMachine) { + if (StringUtils.isEmpty(clientMachine)) { + // No client address: keep the pipeline order (the first node is the write + // primary). Mirrors SCMBlockProtocolServer#getClientNode's empty guard. + return nodes; + } return captureLatencyNs( metrics.getAllocateBlockSortDatanodesLatencyNs(), () -> { final Node client = getClientNode(clientMachine, nodes); 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 18ffca85241b..6c6f71a86133 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 @@ -221,13 +221,19 @@ protected List< OmKeyLocationInfo > allocateBlock(ScmClient scmClient, } throw ex; } + // Cache the sorted order per pipeline so blocks sharing a pipeline are + // sorted once (mirrors the read path's per-pipeline caching). + final Map, List> sortedByNodes = + shouldSortDatanodes ? new HashMap<>() : null; + final String clientAddress = clientMachine; for (AllocatedBlock allocatedBlock : allocatedBlocks) { BlockID blockID = new BlockID(allocatedBlock.getBlockID()); Pipeline pipeline = allocatedBlock.getPipeline(); if (shouldSortDatanodes) { final List nodes = pipeline.getNodes(); - final List sorted = - keyManager.sortDatanodesForWrite(nodes, clientMachine); + final List sorted = sortedByNodes + .computeIfAbsent(nodes, + n -> keyManager.sortDatanodesForWrite(n, clientAddress)); if (!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 e29c150c9670..f4626599954b 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 @@ -24,7 +24,9 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -32,6 +34,7 @@ import java.util.List; import java.util.UUID; import org.apache.hadoop.hdds.client.RatisReplicationConfig; +import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList; import org.apache.hadoop.ozone.OzoneConsts; import org.apache.hadoop.ozone.om.KeyManager; import org.apache.hadoop.ozone.om.helpers.BucketLayout; @@ -253,6 +256,27 @@ public void testAllocateBlockDoesNotSendClientMachineToScm() throws Exception { assertEquals("", clientMachine.getValue()); } + @Test + public void testAllocateBlockSortsSharedPipelineOnce() throws Exception { + // Multiple preallocated blocks that share one pipeline must be sorted once. + KeyManager mockKeyManager = mock(KeyManager.class); + when(mockKeyManager.sortDatanodesForWrite(any(), any())) + .thenAnswer(inv -> inv.getArgument(0)); + when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); + + OMAllocateBlockRequest request = + getOmAllocateBlockRequest(createAllocateBlockRequest()); + // requestedSize spans two scmBlockSize blocks; the SCM stub returns both on + // the same pipeline, so the OM-side sort must run only once. + request.allocateBlock(scmClient, ozoneBlockTokenSecretManager, + replicationConfig, new ExcludeList(), 2 * scmBlockSize, scmBlockSize, + 2, false, "svc", omMetrics, true, + UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(), + mockKeyManager); + + verify(mockKeyManager, times(1)).sortDatanodesForWrite(any(), eq("1.2.3.4")); + } + // Like createAllocateBlockRequest, but sets sortDatanodes and a UserInfo remote address for OM-side sort. private OMRequest createAllocateBlockRequestWithSort(String clientAddress) { KeyArgs keyArgs = KeyArgs.newBuilder() From 5b19b72de76e7213655df4de38afe69284366d9f Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Tue, 30 Jun 2026 21:30:22 +0800 Subject: [PATCH 05/29] HDDS-15059. Strengthen allocateBlock sort tests for reorder and empty client --- .../hadoop/ozone/TestOMSortDatanodes.java | 11 --- .../key/TestOMAllocateBlockRequest.java | 73 +++++++++++++++++-- 2 files changed, 66 insertions(+), 18 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 2c616c298a8e..dcdd176a0f2d 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 @@ -187,17 +187,6 @@ public void sortDatanodesForWritePicksNearestFirst() { } } - @Test - public void sortDatanodesForWriteIgnoresEmptyClient() { - List nodes = nodeManager.getAllNodes(); - List original = new ArrayList<>(nodes); - // An empty client address must never reorder the write pipeline. - List result = - keyManager.sortDatanodesForWrite(nodes, ""); - assertEquals(original, result, - "Empty client must leave the write pipeline order unchanged"); - } - @Test public void sortDatanodesForWriteKeepsOrderWhenClientUnresolved() { List nodes = nodeManager.getAllNodes(); 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 f4626599954b..6ae960c59273 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 @@ -24,6 +24,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; @@ -31,10 +32,21 @@ import static org.mockito.Mockito.when; import jakarta.annotation.Nonnull; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.UUID; +import org.apache.hadoop.hdds.client.ContainerBlockID; import org.apache.hadoop.hdds.client.RatisReplicationConfig; +import org.apache.hadoop.hdds.client.StandaloneReplicationConfig; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor; +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.pipeline.Pipeline; +import org.apache.hadoop.hdds.scm.pipeline.PipelineID; import org.apache.hadoop.ozone.OzoneConsts; import org.apache.hadoop.ozone.om.KeyManager; import org.apache.hadoop.ozone.om.helpers.BucketLayout; @@ -258,23 +270,70 @@ public void testAllocateBlockDoesNotSendClientMachineToScm() throws Exception { @Test public void testAllocateBlockSortsSharedPipelineOnce() throws Exception { - // Multiple preallocated blocks that share one pipeline must be sorted once. + // Two blocks on the same 3-node pipeline must be sorted once, and the + // sorted order must land in every block's pipeline. + List nodes = Arrays.asList( + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()); + Pipeline pipeline = Pipeline.newBuilder() + .setState(Pipeline.PipelineState.OPEN) + .setId(PipelineID.randomId()) + .setReplicationConfig( + StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE)) + .setNodes(nodes) + .build(); + AllocatedBlock.Builder blockBuilder = + new AllocatedBlock.Builder().setPipeline(pipeline); + when(scmBlockLocationProtocol.allocateBlock(anyLong(), anyInt(), any(), + anyString(), any(ExcludeList.class), anyString())).thenAnswer(inv -> { + int num = inv.getArgument(1); + List blocks = new ArrayList<>(num); + for (int i = 0; i < num; i++) { + blockBuilder.setContainerBlockID( + new ContainerBlockID(CONTAINER_ID + i, LOCAL_ID + i)); + blocks.add(blockBuilder.build()); + } + return blocks; + }); + + List sortedOrder = new ArrayList<>(nodes); + Collections.reverse(sortedOrder); KeyManager mockKeyManager = mock(KeyManager.class); when(mockKeyManager.sortDatanodesForWrite(any(), any())) - .thenAnswer(inv -> inv.getArgument(0)); + .thenAnswer(inv -> sortedOrder); when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); OMAllocateBlockRequest request = getOmAllocateBlockRequest(createAllocateBlockRequest()); - // requestedSize spans two scmBlockSize blocks; the SCM stub returns both on - // the same pipeline, so the OM-side sort must run only once. - request.allocateBlock(scmClient, ozoneBlockTokenSecretManager, - replicationConfig, new ExcludeList(), 2 * scmBlockSize, scmBlockSize, - 2, false, "svc", omMetrics, true, + // requestedSize spans two scmBlockSize blocks on the same pipeline. + List locations = request.allocateBlock(scmClient, + ozoneBlockTokenSecretManager, replicationConfig, new ExcludeList(), + 2 * scmBlockSize, scmBlockSize, 2, false, "svc", omMetrics, true, UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(), mockKeyManager); + // Sorted once for the shared pipeline... verify(mockKeyManager, times(1)).sortDatanodesForWrite(any(), eq("1.2.3.4")); + // ...and the sorted order is applied to every block's pipeline. + assertEquals(2, locations.size()); + for (OmKeyLocationInfo location : locations) { + assertEquals(sortedOrder, location.getPipeline().getNodesInOrder()); + } + } + + @Test + public void sortDatanodesForWriteSkipsEmptyClient() { + // An empty client address must short-circuit and return the input order + // unchanged; without the guard the timed sort path runs against the + // fixture's mock metric and throws, so this returning cleanly is the test. + List nodes = Arrays.asList( + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()); + List result = + keyManager.sortDatanodesForWrite(nodes, ""); + assertEquals(nodes, result); } // Like createAllocateBlockRequest, but sets sortDatanodes and a UserInfo remote address for OM-side sort. From e29a89998ceb98d3cd2b6a54c50a1e4c4cce0ac9 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Tue, 30 Jun 2026 21:44:27 +0800 Subject: [PATCH 06/29] HDDS-15059. Match write-sort client by both hostname and IP --- .../hadoop/ozone/TestOMSortDatanodes.java | 19 +++++++++++++++++++ .../hadoop/ozone/om/KeyManagerImpl.java | 15 +++++++-------- 2 files changed, 26 insertions(+), 8 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 dcdd176a0f2d..009c622166ab 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 @@ -187,6 +187,25 @@ public void sortDatanodesForWritePicksNearestFirst() { } } + @Test + public void getClientNodeMatchesByIpAndHostnameInHostnameMode() { + // The client address is always an IP, so a client must match its datanode by + // IP even when use.datanode.hostname is enabled. + config.setBoolean(HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME, true); + try { + List nodes = nodeManager.getAllNodes(); + for (DatanodeDetails dn : nodes) { + assertEquals(dn, keyManager.getClientNode(dn.getIpAddress(), nodes), + "Client IP must resolve to its own datanode in hostname mode"); + assertEquals(dn, keyManager.getClientNode(dn.getHostName(), nodes), + "Client hostname must resolve to its own datanode"); + } + } finally { + config.setBoolean(HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME, + HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME_DEFAULT); + } + } + @Test public void sortDatanodesForWriteKeepsOrderWhenClientUnresolved() { List nodes = nodeManager.getAllNodes(); 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 fe4d1f1afa45..2a0beae25c19 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 @@ -115,7 +115,6 @@ import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension; import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension.EncryptedKeyVersion; import org.apache.hadoop.fs.FileEncryptionInfo; -import org.apache.hadoop.hdds.HddsConfigKeys; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.client.ReplicationConfig; import org.apache.hadoop.hdds.conf.OzoneConfiguration; @@ -2230,15 +2229,15 @@ public List sortDatanodesForWrite( }); } - private Node getClientNode(String clientMachine, - List nodes) { + @VisibleForTesting + public Node getClientNode(String clientMachine, + List nodes) { List matchingNodes = new ArrayList<>(); - boolean useHostname = ozoneManager.getConfiguration().getBoolean( - HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME, - HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME_DEFAULT); for (DatanodeDetails node : nodes) { - if ((useHostname ? node.getHostName() : node.getIpAddress()).equals( - clientMachine)) { + // Match by either IP or hostname, like SCM's getNodesByAddress; the client + // address is always an IP even when use.datanode.hostname is enabled. + if (clientMachine.equals(node.getIpAddress()) + || clientMachine.equals(node.getHostName())) { matchingNodes.add(node); } } From 91bed089b6e9ce03fccf78570b1f3ff27594ffac Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Tue, 30 Jun 2026 22:18:52 +0800 Subject: [PATCH 07/29] HDDS-15059. Resolve write-sort pipeline nodes to OM cluster map --- .../hadoop/ozone/TestOMSortDatanodes.java | 32 +++++++++++++ .../hadoop/ozone/om/KeyManagerImpl.java | 46 ++++++++++++++++++- 2 files changed, 76 insertions(+), 2 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 009c622166ab..a92e10b7e03e 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.assertSame; import static org.mockito.Mockito.mock; import com.google.common.collect.ImmutableMap; @@ -187,6 +188,37 @@ public void sortDatanodesForWritePicksNearestFirst() { } } + @Test + public void sortDatanodesForWriteSortsRpcDeserializedPipeline() { + // Pipeline nodes arrive from SCM over RPC as deserialized DatanodeDetails + // with no topology linkage; the write sort must resolve them to OM's cluster + // map, otherwise every node is equidistant (MAX) and the order is random. + List rpcNodes = new ArrayList<>(); + for (DatanodeDetails dn : nodeManager.getAllNodes()) { + rpcNodes.add(DatanodeDetails.getFromProtoBuf(dn.getProtoBufMessage())); + } + for (DatanodeDetails dn : nodeManager.getAllNodes()) { + List sorted = + keyManager.sortDatanodesForWrite(rpcNodes, dn.getIpAddress()); + assertEquals(dn, sorted.get(0), + "Source node should be sorted first for writes"); + assertRackOrder(dn.getNetworkLocation(), sorted); + } + } + + @Test + public void sortDatanodesForWriteKeepsOrderForStaleTopology() { + List nodes = new ArrayList<>(); + nodes.add(randomDatanodeDetails()); + nodes.addAll(nodeManager.getAllNodes()); + + List sorted = + keyManager.sortDatanodesForWrite(nodes, "edge0"); + + assertSame(nodes, sorted, + "Pipeline order should be preserved when a node is missing from the OM topology"); + } + @Test public void getClientNodeMatchesByIpAndHostnameInHostnameMode() { // The client address is always an IP, so a client must match its datanode by 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 2a0beae25c19..ade1fb9e7845 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 @@ -123,6 +123,7 @@ import org.apache.hadoop.hdds.scm.ScmConfigKeys; import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline; import org.apache.hadoop.hdds.scm.net.InnerNode; +import org.apache.hadoop.hdds.scm.net.NetworkTopology; import org.apache.hadoop.hdds.scm.net.Node; import org.apache.hadoop.hdds.scm.net.NodeImpl; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; @@ -2224,11 +2225,52 @@ public List sortDatanodesForWrite( // primary, so do not shuffle when the client cannot be resolved. return nodes; } - return ozoneManager.getClusterMap() - .sortByDistanceCost(client, nodes, nodes.size()); + return sortByClusterMapDistance(client, nodes); }); } + /** + * Sort a pipeline's nodes by topology distance to the client. The nodes come + * from SCM over RPC, so they are deserialized {@link DatanodeDetails} with no + * parent/level and would be treated as outside the topology (distance + * {@link Integer#MAX_VALUE}) and shuffled. Resolve each node (and a co-located + * client) to its canonical instance in OM's cluster map before sorting, then + * map the sorted order back to the original pipeline nodes. + */ + private List sortByClusterMapDistance( + Node client, List nodes) { + // Read the cluster map once: the topology poller may swap it, and resolving + // and sorting against the same instance keeps the parent pointers consistent. + final NetworkTopology clusterMap = ozoneManager.getClusterMap(); + final Node reader = toClusterMapNode(clusterMap, client); + final List topologyNodes = new ArrayList<>(nodes.size()); + final Map nodeByPath = new HashMap<>(); + for (DatanodeDetails node : nodes) { + final Node resolved = clusterMap.getNode(node.getNetworkFullPath()); + if (resolved == null) { + return nodes; + } + topologyNodes.add(resolved); + nodeByPath.put(resolved.getNetworkFullPath(), node); + } + final List sorted = + clusterMap.sortByDistanceCost(reader, topologyNodes, topologyNodes.size()); + final List result = new ArrayList<>(sorted.size()); + for (Node node : sorted) { + result.add(nodeByPath.get(node.getNetworkFullPath())); + } + return result; + } + + /** + * Resolve a node to its canonical, topology-linked instance in the given + * cluster map, or return the input node if it is not in the map. + */ + private Node toClusterMapNode(NetworkTopology clusterMap, Node node) { + final Node resolved = clusterMap.getNode(node.getNetworkFullPath()); + return resolved != null ? resolved : node; + } + @VisibleForTesting public Node getClientNode(String clientMachine, List nodes) { From 59f9c26c8d1cf1a81357a750b78b5ae1024f75de Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Tue, 30 Jun 2026 23:13:15 +0800 Subject: [PATCH 08/29] HDDS-15059. Consolidate write-sort datanode tests --- .../hadoop/ozone/TestOMSortDatanodes.java | 46 +++++-------------- 1 file changed, 12 insertions(+), 34 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 a92e10b7e03e..5732f50328a8 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 @@ -177,17 +177,6 @@ private static void assertRackOrder(String rack, List } } - @Test - public void sortDatanodesForWritePicksNearestFirst() { - for (DatanodeDetails dn : nodeManager.getAllNodes()) { - List sorted = - keyManager.sortDatanodesForWrite(nodeManager.getAllNodes(), nodeAddress(dn)); - assertEquals(dn, sorted.get(0), - "Source node should be sorted first for writes"); - assertRackOrder(dn.getNetworkLocation(), sorted); - } - } - @Test public void sortDatanodesForWriteSortsRpcDeserializedPipeline() { // Pipeline nodes arrive from SCM over RPC as deserialized DatanodeDetails @@ -198,11 +187,19 @@ public void sortDatanodesForWriteSortsRpcDeserializedPipeline() { rpcNodes.add(DatanodeDetails.getFromProtoBuf(dn.getProtoBufMessage())); } for (DatanodeDetails dn : nodeManager.getAllNodes()) { - List sorted = + // The client address is normally an IP, but the sort must resolve a client + // by either IP or hostname, so cover both. + List byIp = keyManager.sortDatanodesForWrite(rpcNodes, dn.getIpAddress()); - assertEquals(dn, sorted.get(0), - "Source node should be sorted first for writes"); - assertRackOrder(dn.getNetworkLocation(), sorted); + assertEquals(dn, byIp.get(0), + "Source node should be sorted first for writes (IP client)"); + assertRackOrder(dn.getNetworkLocation(), byIp); + + List byHostname = + keyManager.sortDatanodesForWrite(rpcNodes, dn.getHostName()); + assertEquals(dn, byHostname.get(0), + "Source node should be sorted first for writes (hostname client)"); + assertRackOrder(dn.getNetworkLocation(), byHostname); } } @@ -219,25 +216,6 @@ public void sortDatanodesForWriteKeepsOrderForStaleTopology() { "Pipeline order should be preserved when a node is missing from the OM topology"); } - @Test - public void getClientNodeMatchesByIpAndHostnameInHostnameMode() { - // The client address is always an IP, so a client must match its datanode by - // IP even when use.datanode.hostname is enabled. - config.setBoolean(HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME, true); - try { - List nodes = nodeManager.getAllNodes(); - for (DatanodeDetails dn : nodes) { - assertEquals(dn, keyManager.getClientNode(dn.getIpAddress(), nodes), - "Client IP must resolve to its own datanode in hostname mode"); - assertEquals(dn, keyManager.getClientNode(dn.getHostName(), nodes), - "Client hostname must resolve to its own datanode"); - } - } finally { - config.setBoolean(HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME, - HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME_DEFAULT); - } - } - @Test public void sortDatanodesForWriteKeepsOrderWhenClientUnresolved() { List nodes = nodeManager.getAllNodes(); From 13f1e1a79d51cb4b434477f66a3de88a6bfbc444 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Tue, 30 Jun 2026 23:13:15 +0800 Subject: [PATCH 09/29] HDDS-15059. Read cluster map once per sort to avoid topology swap race --- .../hadoop/ozone/om/KeyManagerImpl.java | 33 +++++++++---------- 1 file changed, 15 insertions(+), 18 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 ade1fb9e7845..8cdb80967ab9 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 @@ -2204,9 +2204,9 @@ private void sortDatanodes(String clientMachine, List keyInfos) { @VisibleForTesting public List sortDatanodes(List nodes, String clientMachine) { - final Node client = getClientNode(clientMachine, nodes); - return ozoneManager.getClusterMap() - .sortByDistanceCost(client, nodes, nodes.size()); + final NetworkTopology clusterMap = ozoneManager.getClusterMap(); + final Node client = getClientNode(clientMachine, nodes, clusterMap); + return clusterMap.sortByDistanceCost(client, nodes, nodes.size()); } @Override @@ -2219,13 +2219,14 @@ public List sortDatanodesForWrite( } return captureLatencyNs( metrics.getAllocateBlockSortDatanodesLatencyNs(), () -> { - final Node client = getClientNode(clientMachine, nodes); + final NetworkTopology clusterMap = ozoneManager.getClusterMap(); + 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; } - return sortByClusterMapDistance(client, nodes); + return sortByClusterMapDistance(clusterMap, client, nodes); }); } @@ -2238,10 +2239,8 @@ public List sortDatanodesForWrite( * map the sorted order back to the original pipeline nodes. */ private List sortByClusterMapDistance( - Node client, List nodes) { - // Read the cluster map once: the topology poller may swap it, and resolving - // and sorting against the same instance keeps the parent pointers consistent. - final NetworkTopology clusterMap = ozoneManager.getClusterMap(); + NetworkTopology clusterMap, Node client, + List nodes) { final Node reader = toClusterMapNode(clusterMap, client); final List topologyNodes = new ArrayList<>(nodes.size()); final Map nodeByPath = new HashMap<>(); @@ -2271,27 +2270,25 @@ private Node toClusterMapNode(NetworkTopology clusterMap, Node node) { return resolved != null ? resolved : node; } - @VisibleForTesting - public Node getClientNode(String clientMachine, - List nodes) { - List matchingNodes = new ArrayList<>(); + private Node getClientNode(String clientMachine, + List nodes, NetworkTopology clusterMap) { for (DatanodeDetails node : nodes) { // Match by either IP or hostname, like SCM's getNodesByAddress; the client // address is always an IP even when use.datanode.hostname is enabled. if (clientMachine.equals(node.getIpAddress()) || clientMachine.equals(node.getHostName())) { - matchingNodes.add(node); + return node; } } - return !matchingNodes.isEmpty() ? matchingNodes.get(0) : - getOtherNode(clientMachine); + return getOtherNode(clientMachine, clusterMap); } - private Node getOtherNode(String clientMachine) { + private Node getOtherNode(String clientMachine, + NetworkTopology clusterMap) { try { String clientLocation = resolveNodeLocation(clientMachine); if (clientLocation != null) { - Node rack = ozoneManager.getClusterMap().getNode(clientLocation); + Node rack = clusterMap.getNode(clientLocation); if (rack instanceof InnerNode) { return new NodeImpl(clientMachine, clientLocation, (InnerNode) rack, rack.getLevel() + 1, From 5040d00e80dd9da74923432acf4cc05288fd302c Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Tue, 30 Jun 2026 23:15:18 +0800 Subject: [PATCH 10/29] HDDS-15059. Simplify write-sort client handling --- .../apache/hadoop/ozone/om/request/key/OMKeyRequest.java | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) 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 6c6f71a86133..f8cb96703f79 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 @@ -200,10 +200,8 @@ protected List< OmKeyLocationInfo > allocateBlock(ScmClient scmClient, int numBlocks = (int) Math.min(preallocateBlocksMax, (requestedSize - 1) / (scmBlockSize * dataGroupSize) + 1); - String clientMachine = ""; - if (shouldSortDatanodes) { - clientMachine = userInfo.getRemoteAddress(); - } + final String clientMachine = + shouldSortDatanodes ? userInfo.getRemoteAddress() : ""; List locationInfos = new ArrayList<>(numBlocks); String remoteUser = getRemoteUser().getShortUserName(); @@ -225,7 +223,6 @@ protected List< OmKeyLocationInfo > allocateBlock(ScmClient scmClient, // sorted once (mirrors the read path's per-pipeline caching). final Map, List> sortedByNodes = shouldSortDatanodes ? new HashMap<>() : null; - final String clientAddress = clientMachine; for (AllocatedBlock allocatedBlock : allocatedBlocks) { BlockID blockID = new BlockID(allocatedBlock.getBlockID()); Pipeline pipeline = allocatedBlock.getPipeline(); @@ -233,7 +230,7 @@ protected List< OmKeyLocationInfo > allocateBlock(ScmClient scmClient, final List nodes = pipeline.getNodes(); final List sorted = sortedByNodes .computeIfAbsent(nodes, - n -> keyManager.sortDatanodesForWrite(n, clientAddress)); + n -> keyManager.sortDatanodesForWrite(n, clientMachine)); if (!Objects.equals(sorted, pipeline.getNodesInOrder())) { pipeline = pipeline.copyWithNodesInOrder(sorted); } From ec36f7dd7321717f4a62b51da5bddf4e060e24b3 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Mon, 6 Jul 2026 21:35:51 +0800 Subject: [PATCH 11/29] HDDS-15059. Make OM-side write sortDatanodes configurable (default off) --- .../org/apache/hadoop/ozone/om/OmConfig.java | 16 ++++++++++++++ .../apache/hadoop/ozone/om/KeyManager.java | 7 ++++++ .../hadoop/ozone/om/KeyManagerImpl.java | 5 +++++ .../ozone/om/request/key/OMKeyRequest.java | 17 +++++++++----- .../key/TestOMAllocateBlockRequest.java | 22 +++++++++++++++++++ 5 files changed, 61 insertions(+), 6 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 b8a60f9bcdb1..c981afd414d8 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 @@ -184,6 +184,18 @@ public class OmConfig extends ReconfigurableConfig { ) private long followerReadLocalLeaseTimeMs; + @Config(key = "ozone.om.block.write.sort.datanodes.enabled", + defaultValue = "false", + type = ConfigType.BOOLEAN, + tags = {ConfigTag.OM, ConfigTag.PERFORMANCE}, + description = "If true, OM sorts the streaming-write pipeline (nearest " + + "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." + ) + private boolean sortDatanodesForWriteEnabled; + public long getRatisBasedFinalizationTimeout() { return ratisBasedFinalizationTimeout; } @@ -224,6 +236,10 @@ public void setAllowLeaderSkipLinearizableRead(boolean newValue) { allowLeaderSkipLinearizableRead = newValue; } + public boolean isSortDatanodesForWriteEnabled() { + return sortDatanodesForWriteEnabled; + } + public boolean isFollowerReadLocalLeaseEnabled() { return followerReadLocalLeaseEnabled; } 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 0a561a277725..f4061cc8b17a 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 @@ -377,4 +377,11 @@ DeleteKeysResult getPendingDeletionSubFiles(long volumeId, long bucketId, OmKeyI */ List sortDatanodesForWrite( List nodes, String clientMachine); + + /** + * @return true if OM should sort the streaming-write pipeline locally + * ({@code ozone.om.block.write.sort.datanodes.enabled}); false to leave + * the sort to SCM. + */ + boolean isSortDatanodesForWriteEnabled(); } 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 8cdb80967ab9..a0430cc8b221 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 @@ -2230,6 +2230,11 @@ public List sortDatanodesForWrite( }); } + @Override + public boolean isSortDatanodesForWriteEnabled() { + return ozoneManager.getConfig().isSortDatanodesForWriteEnabled(); + } + /** * Sort a pipeline's nodes by topology distance to the client. The nodes come * from SCM over RPC, so they are deserialized {@link DatanodeDetails} with no 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 f8cb96703f79..56e742539f3e 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 @@ -200,8 +200,13 @@ protected List< OmKeyLocationInfo > allocateBlock(ScmClient scmClient, int numBlocks = (int) Math.min(preallocateBlocksMax, (requestedSize - 1) / (scmBlockSize * dataGroupSize) + 1); - final String clientMachine = - shouldSortDatanodes ? userInfo.getRemoteAddress() : ""; + final boolean sortOnOm = + shouldSortDatanodes && keyManager.isSortDatanodesForWriteEnabled(); + // When SCM sorts, it still needs the real client address; the OM-sort path + // sends an empty address so SCM skips its sort. + final String scmClientMachine = + (shouldSortDatanodes && !sortOnOm) ? userInfo.getRemoteAddress() : ""; + final String omClientMachine = sortOnOm ? userInfo.getRemoteAddress() : ""; List locationInfos = new ArrayList<>(numBlocks); String remoteUser = getRemoteUser().getShortUserName(); @@ -209,7 +214,7 @@ protected List< OmKeyLocationInfo > allocateBlock(ScmClient scmClient, try { allocatedBlocks = scmClient.getBlockClient() .allocateBlock(scmBlockSize, numBlocks, replicationConfig, serviceID, - excludeList, ""); + excludeList, scmClientMachine); } catch (SCMException ex) { omMetrics.incNumBlockAllocateCallFails(); if (ex.getResult() @@ -222,15 +227,15 @@ protected List< OmKeyLocationInfo > allocateBlock(ScmClient scmClient, // Cache the sorted order per pipeline so blocks sharing a pipeline are // sorted once (mirrors the read path's per-pipeline caching). final Map, List> sortedByNodes = - shouldSortDatanodes ? new HashMap<>() : null; + sortOnOm ? new HashMap<>() : null; for (AllocatedBlock allocatedBlock : allocatedBlocks) { BlockID blockID = new BlockID(allocatedBlock.getBlockID()); Pipeline pipeline = allocatedBlock.getPipeline(); - if (shouldSortDatanodes) { + if (sortOnOm) { final List nodes = pipeline.getNodes(); final List sorted = sortedByNodes .computeIfAbsent(nodes, - n -> keyManager.sortDatanodesForWrite(n, clientMachine)); + n -> keyManager.sortDatanodesForWrite(n, omClientMachine)); if (!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 6ae960c59273..a4c5e550f8f4 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 @@ -27,6 +27,7 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -249,11 +250,31 @@ protected OMRequest doPreExecute(OMRequest originalOMRequest) return modifiedOmRequest; } + @Test + public void testAllocateBlockSendsClientMachineToScmWhenFlagOff() throws Exception { + // Flag off (default): OM must NOT sort; SCM receives the real client address + // so it performs the sort. + KeyManager mockKeyManager = mock(KeyManager.class); + when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(false); + when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); + + OMAllocateBlockRequest request = + getOmAllocateBlockRequest(createAllocateBlockRequestWithSort("1.2.3.4")); + request.preExecute(ozoneManager); + + ArgumentCaptor clientMachine = ArgumentCaptor.forClass(String.class); + verify(scmBlockLocationProtocol).allocateBlock(anyLong(), anyInt(), any(), + any(), any(), clientMachine.capture()); + assertEquals("1.2.3.4", clientMachine.getValue()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString()); + } + @Test public void testAllocateBlockDoesNotSendClientMachineToScm() throws Exception { // OM now sorts the write pipeline locally, so SCM must receive an empty // clientMachine even when the client requests sorted datanodes. KeyManager mockKeyManager = mock(KeyManager.class); + when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); when(mockKeyManager.sortDatanodesForWrite(any(), any())) .thenAnswer(inv -> inv.getArgument(0)); when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); @@ -300,6 +321,7 @@ public void testAllocateBlockSortsSharedPipelineOnce() throws Exception { List sortedOrder = new ArrayList<>(nodes); Collections.reverse(sortedOrder); KeyManager mockKeyManager = mock(KeyManager.class); + when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); when(mockKeyManager.sortDatanodesForWrite(any(), any())) .thenAnswer(inv -> sortedOrder); when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); From 7d3bc3ee61623b3fccb392c15dadab51cf74e5cd Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Wed, 8 Jul 2026 21:30:32 +0800 Subject: [PATCH 12/29] HDDS-15059. Key read-sort cache by pipeline nodes, dropping getUuidString --- .../java/org/apache/hadoop/ozone/om/KeyManagerImpl.java | 9 +++------ 1 file changed, 3 insertions(+), 6 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 a0430cc8b221..07613375fe4b 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 @@ -2165,7 +2165,7 @@ private void sortDatanodes(String clientMachine, OmKeyInfo keyInfo) { private void sortDatanodes(String clientMachine, List keyInfos) { if (keyInfos != null && clientMachine != null) { - final Map, List> sortedPipelines = new HashMap<>(); + final Map, List> sortedPipelines = new HashMap<>(); for (OmKeyInfo keyInfo : keyInfos) { OmKeyLocationInfoGroup key = keyInfo.getLatestVersionLocations(); if (key == null) { @@ -2180,14 +2180,11 @@ private void sortDatanodes(String clientMachine, List keyInfos) { continue; } - final Set uuidSet = nodes.stream().map(DatanodeDetails::getUuidString) - .collect(Collectors.toSet()); - - List sortedNodes = sortedPipelines.get(uuidSet); + List sortedNodes = sortedPipelines.get(nodes); if (sortedNodes == null) { sortedNodes = sortDatanodes(nodes, clientMachine); if (sortedNodes != null) { - sortedPipelines.put(uuidSet, sortedNodes); + sortedPipelines.put(nodes, sortedNodes); } } else if (LOG.isDebugEnabled()) { LOG.debug("Found sorted datanodes for pipeline {} and client {} " From 5d67fdd42a840db974350a2d460bc0e1917d8e95 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Wed, 8 Jul 2026 21:30:41 +0800 Subject: [PATCH 13/29] HDDS-15059. Clarify write-sort cache comment (keyed by pipeline nodes) --- .../org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 56e742539f3e..c2a50b7e3108 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 @@ -224,8 +224,8 @@ protected List< OmKeyLocationInfo > allocateBlock(ScmClient scmClient, } throw ex; } - // Cache the sorted order per pipeline so blocks sharing a pipeline are - // sorted once (mirrors the read path's per-pipeline caching). + // Cache the sorted order by pipeline nodes so blocks whose pipelines have + // the same datanodes are sorted once (mirrors the read path's caching). final Map, List> sortedByNodes = sortOnOm ? new HashMap<>() : null; for (AllocatedBlock allocatedBlock : allocatedBlocks) { From d310622491f8e204fb00f70f2b0a6658126cf3d2 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Wed, 8 Jul 2026 21:30:58 +0800 Subject: [PATCH 14/29] HDDS-15059. Reword write-sort javadoc to drop unclear terms --- .../org/apache/hadoop/ozone/om/KeyManagerImpl.java | 12 ++++++------ 1 file changed, 6 insertions(+), 6 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 07613375fe4b..b2cbbe06fff4 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 @@ -2235,10 +2235,10 @@ public boolean isSortDatanodesForWriteEnabled() { /** * Sort a pipeline's nodes by topology distance to the client. The nodes come * from SCM over RPC, so they are deserialized {@link DatanodeDetails} with no - * parent/level and would be treated as outside the topology (distance - * {@link Integer#MAX_VALUE}) and shuffled. Resolve each node (and a co-located - * client) to its canonical instance in OM's cluster map before sorting, then - * map the sorted order back to the original pipeline nodes. + * parent/level: the topology treats them as unknown (distance + * {@link Integer#MAX_VALUE}) and the order comes out random. Look each node + * (and a co-located client) up in OM's cluster map to get the topology-linked + * instance, sort those, then map the order back to the original nodes. */ private List sortByClusterMapDistance( NetworkTopology clusterMap, Node client, @@ -2264,8 +2264,8 @@ private List sortByClusterMapDistance( } /** - * Resolve a node to its canonical, topology-linked instance in the given - * cluster map, or return the input node if it is not in the map. + * Look a node up in the cluster map and return the topology-linked instance + * found there, or the input node if the map has no entry for it. */ private Node toClusterMapNode(NetworkTopology clusterMap, Node node) { final Node resolved = clusterMap.getNode(node.getNetworkFullPath()); From 356023d1c3105eeba37757600cfd00f7422dd49b Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Wed, 8 Jul 2026 21:39:29 +0800 Subject: [PATCH 15/29] HDDS-15059. Resolve co-located write client in getClientNode --- .../hadoop/ozone/om/KeyManagerImpl.java | 21 +++++++------------ 1 file changed, 7 insertions(+), 14 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 b2cbbe06fff4..3fad3d08fdc2 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 @@ -2237,13 +2237,12 @@ public boolean isSortDatanodesForWriteEnabled() { * from SCM over RPC, so they are deserialized {@link DatanodeDetails} with no * parent/level: the topology treats them as unknown (distance * {@link Integer#MAX_VALUE}) and the order comes out random. Look each node - * (and a co-located client) up in OM's cluster map to get the topology-linked - * instance, sort those, then map the order back to the original nodes. + * up in OM's cluster map to get the topology-linked instance, sort those, + * then map the order back to the original nodes. */ private List sortByClusterMapDistance( NetworkTopology clusterMap, Node client, List nodes) { - final Node reader = toClusterMapNode(clusterMap, client); final List topologyNodes = new ArrayList<>(nodes.size()); final Map nodeByPath = new HashMap<>(); for (DatanodeDetails node : nodes) { @@ -2255,7 +2254,7 @@ private List sortByClusterMapDistance( nodeByPath.put(resolved.getNetworkFullPath(), node); } final List sorted = - clusterMap.sortByDistanceCost(reader, topologyNodes, topologyNodes.size()); + clusterMap.sortByDistanceCost(client, topologyNodes, topologyNodes.size()); final List result = new ArrayList<>(sorted.size()); for (Node node : sorted) { result.add(nodeByPath.get(node.getNetworkFullPath())); @@ -2263,15 +2262,6 @@ private List sortByClusterMapDistance( return result; } - /** - * Look a node up in the cluster map and return the topology-linked instance - * found there, or the input node if the map has no entry for it. - */ - private Node toClusterMapNode(NetworkTopology clusterMap, Node node) { - final Node resolved = clusterMap.getNode(node.getNetworkFullPath()); - return resolved != null ? resolved : node; - } - private Node getClientNode(String clientMachine, List nodes, NetworkTopology clusterMap) { for (DatanodeDetails node : nodes) { @@ -2279,7 +2269,10 @@ private Node getClientNode(String clientMachine, // address is always an IP even when use.datanode.hostname is enabled. if (clientMachine.equals(node.getIpAddress()) || clientMachine.equals(node.getHostName())) { - return node; + // The pipeline nodes are RPC-deserialized and not linked into OM's + // cluster map; prefer the map's instance so distance can be computed. + final Node resolved = clusterMap.getNode(node.getNetworkFullPath()); + return resolved != null ? resolved : node; } } return getOtherNode(clientMachine, clusterMap); From 16086e2b30d983938df8b632fb3a78cfa82a1642 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Thu, 9 Jul 2026 20:18:08 +0800 Subject: [PATCH 16/29] HDDS-15059. Split write datanode sort cases --- .../ozone/om/request/key/OMKeyRequest.java | 27 ++++++++++++------- 1 file changed, 17 insertions(+), 10 deletions(-) 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 c2a50b7e3108..9c132e606fcc 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 @@ -200,13 +200,22 @@ protected List< OmKeyLocationInfo > allocateBlock(ScmClient scmClient, int numBlocks = (int) Math.min(preallocateBlocksMax, (requestedSize - 1) / (scmBlockSize * dataGroupSize) + 1); - final boolean sortOnOm = - shouldSortDatanodes && keyManager.isSortDatanodesForWriteEnabled(); - // When SCM sorts, it still needs the real client address; the OM-sort path - // sends an empty address so SCM skips its sort. - final String scmClientMachine = - (shouldSortDatanodes && !sortOnOm) ? userInfo.getRemoteAddress() : ""; - final String omClientMachine = sortOnOm ? userInfo.getRemoteAddress() : ""; + final String scmClientMachine; + final String omClientMachine; + final Map, List> sortedByNodes; + if (!shouldSortDatanodes) { + scmClientMachine = ""; + omClientMachine = ""; + sortedByNodes = null; + } else if (keyManager.isSortDatanodesForWriteEnabled()) { + scmClientMachine = ""; + omClientMachine = userInfo.getRemoteAddress(); + sortedByNodes = new HashMap<>(); + } else { + scmClientMachine = userInfo.getRemoteAddress(); + omClientMachine = ""; + sortedByNodes = null; + } List locationInfos = new ArrayList<>(numBlocks); String remoteUser = getRemoteUser().getShortUserName(); @@ -226,12 +235,10 @@ protected List< OmKeyLocationInfo > allocateBlock(ScmClient scmClient, } // Cache the sorted order by pipeline nodes so blocks whose pipelines have // the same datanodes are sorted once (mirrors the read path's caching). - final Map, List> sortedByNodes = - sortOnOm ? new HashMap<>() : null; for (AllocatedBlock allocatedBlock : allocatedBlocks) { BlockID blockID = new BlockID(allocatedBlock.getBlockID()); Pipeline pipeline = allocatedBlock.getPipeline(); - if (sortOnOm) { + if (sortedByNodes != null) { final List nodes = pipeline.getNodes(); final List sorted = sortedByNodes .computeIfAbsent(nodes, From f48037da9345c80977c0d28f20b69d1b34e77e48 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Thu, 9 Jul 2026 20:18:17 +0800 Subject: [PATCH 17/29] HDDS-15059. Require client machine for write sorting --- .../org/apache/hadoop/ozone/om/KeyManagerImpl.java | 6 +----- .../om/request/key/TestOMAllocateBlockRequest.java | 11 ++++------- 2 files changed, 5 insertions(+), 12 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 3fad3d08fdc2..b3ab8bbef1e0 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 @@ -2209,11 +2209,7 @@ public List sortDatanodes(List sortDatanodesForWrite( List nodes, String clientMachine) { - if (StringUtils.isEmpty(clientMachine)) { - // No client address: keep the pipeline order (the first node is the write - // primary). Mirrors SCMBlockProtocolServer#getClientNode's empty guard. - return nodes; - } + Preconditions.checkArgument(!StringUtils.isEmpty(clientMachine), clientMachine); return captureLatencyNs( metrics.getAllocateBlockSortDatanodesLatencyNs(), () -> { final NetworkTopology clusterMap = ozoneManager.getClusterMap(); 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 a4c5e550f8f4..cdc451277ece 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 @@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; @@ -345,17 +346,13 @@ ozoneBlockTokenSecretManager, replicationConfig, new ExcludeList(), } @Test - public void sortDatanodesForWriteSkipsEmptyClient() { - // An empty client address must short-circuit and return the input order - // unchanged; without the guard the timed sort path runs against the - // fixture's mock metric and throws, so this returning cleanly is the test. + public void sortDatanodesForWriteRequiresClientMachine() { List nodes = Arrays.asList( MockDatanodeDetails.randomDatanodeDetails(), MockDatanodeDetails.randomDatanodeDetails(), MockDatanodeDetails.randomDatanodeDetails()); - List result = - keyManager.sortDatanodesForWrite(nodes, ""); - assertEquals(nodes, result); + assertThrows(IllegalArgumentException.class, + () -> keyManager.sortDatanodesForWrite(nodes, "")); } // Like createAllocateBlockRequest, but sets sortDatanodes and a UserInfo remote address for OM-side sort. From 914bf1b873dc2e2cf1dc255dfead6b280bff342f Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Thu, 9 Jul 2026 20:26:50 +0800 Subject: [PATCH 18/29] HDDS-15059. Keep write order without OM topology cache --- .../apache/hadoop/ozone/om/KeyManagerImpl.java | 9 ++++++++- .../request/key/TestOMAllocateBlockRequest.java | 15 +++++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) 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 b3ab8bbef1e0..ec043a2fe20a 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 @@ -2210,9 +2210,16 @@ public List sortDatanodes(List sortDatanodesForWrite( List nodes, String clientMachine) { Preconditions.checkArgument(!StringUtils.isEmpty(clientMachine), clientMachine); + final NetworkTopology clusterMap; + try { + clusterMap = ozoneManager.getClusterMap(); + } catch (NullPointerException ex) { + LOG.warn("Could not sort write pipeline for client {} because OM's " + + "network topology cache is unavailable.", clientMachine); + return nodes; + } return captureLatencyNs( metrics.getAllocateBlockSortDatanodesLatencyNs(), () -> { - final NetworkTopology clusterMap = ozoneManager.getClusterMap(); final Node client = getClientNode(clientMachine, nodes, clusterMap); if (client == null) { // Preserve pipeline order for writes: the first node is the write 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 cdc451277ece..482ae7be54e2 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 @@ -355,6 +355,21 @@ public void sortDatanodesForWriteRequiresClientMachine() { () -> keyManager.sortDatanodesForWrite(nodes, "")); } + @Test + public void sortDatanodesForWriteKeepsOrderWhenTopologyUnavailable() { + when(ozoneManager.getClusterMap()).thenThrow( + new NullPointerException("ScmBlockLocationClient must have been initialized already.")); + List nodes = Arrays.asList( + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()); + + List result = + keyManager.sortDatanodesForWrite(nodes, "1.2.3.4"); + + assertSame(nodes, result); + } + // Like createAllocateBlockRequest, but sets sortDatanodes and a UserInfo remote address for OM-side sort. private OMRequest createAllocateBlockRequestWithSort(String clientAddress) { KeyArgs keyArgs = KeyArgs.newBuilder() From 8f420e9dca8b5a11a6096116604da8702fd4078e Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Fri, 10 Jul 2026 20:29:19 +0800 Subject: [PATCH 19/29] HDDS-15059. Fix import order in TestOMAllocateBlockRequest --- .../hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 96bfff048e54..b6c29ef9a71f 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 @@ -19,8 +19,8 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; From 0d47a4b7685868c0bce58bc4300859f649d526e8 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Fri, 10 Jul 2026 20:35:08 +0800 Subject: [PATCH 20/29] HDDS-15059. Fall back to SCM when OM topology is unavailable --- .../hdds/scm/client/ScmTopologyClient.java | 4 +-- .../hadoop/ozone/om/KeyManagerImpl.java | 9 +---- .../apache/hadoop/ozone/om/OzoneManager.java | 6 ++++ .../ozone/om/request/key/OMKeyRequest.java | 6 +++- .../key/TestOMAllocateBlockRequest.java | 35 +++++++++++-------- 5 files changed, 33 insertions(+), 27 deletions(-) diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/client/ScmTopologyClient.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/client/ScmTopologyClient.java index d595bd6e0958..7abd2c9c0161 100644 --- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/client/ScmTopologyClient.java +++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/client/ScmTopologyClient.java @@ -17,7 +17,6 @@ package org.apache.hadoop.hdds.scm.client; -import static java.util.Objects.requireNonNull; import static org.apache.hadoop.hdds.scm.net.NetConstants.ROOT; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_OM_NETWORK_TOPOLOGY_REFRESH_DURATION; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_OM_NETWORK_TOPOLOGY_REFRESH_DURATION_DEFAULT; @@ -60,8 +59,7 @@ public ScmTopologyClient( } public NetworkTopology getClusterMap() { - return requireNonNull(cache.get(), - "ScmBlockLocationClient must have been initialized already."); + return cache.get(); } public void start(ConfigurationSource conf) throws IOException { 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 ec043a2fe20a..03f2182b7483 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 @@ -2210,14 +2210,7 @@ public List sortDatanodes(List sortDatanodesForWrite( List nodes, String clientMachine) { Preconditions.checkArgument(!StringUtils.isEmpty(clientMachine), clientMachine); - final NetworkTopology clusterMap; - try { - clusterMap = ozoneManager.getClusterMap(); - } catch (NullPointerException ex) { - LOG.warn("Could not sort write pipeline for client {} because OM's " - + "network topology cache is unavailable.", clientMachine); - return nodes; - } + final NetworkTopology clusterMap = ozoneManager.getClusterMap(); return captureLatencyNs( metrics.getAllocateBlockSortDatanodesLatencyNs(), () -> { final Node client = getClientNode(clientMachine, nodes, clusterMap); diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java index 1910d92e9690..da123ec32883 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java @@ -18,6 +18,7 @@ package org.apache.hadoop.ozone.om; import static java.nio.charset.StandardCharsets.UTF_8; +import static java.util.Objects.requireNonNull; import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_DEFAULT; import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_KEY; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_BLOCK_TOKEN_ENABLED; @@ -1342,6 +1343,11 @@ public void setScmTopologyClient( } public NetworkTopology getClusterMap() { + return requireNonNull(getClusterMapAllowNull(), + "ScmBlockLocationClient must have been initialized already."); + } + + public NetworkTopology getClusterMapAllowNull() { return scmTopologyClient.getClusterMap(); } 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 4f60a0a51df9..d93c78c1bf34 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 @@ -62,6 +62,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.exceptions.SCMException; +import org.apache.hadoop.hdds.scm.net.NetworkTopology; 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; @@ -201,11 +202,14 @@ protected List allocateBlock( final String scmClientMachine; final String omClientMachine; final Map, List> sortedByNodes; + final NetworkTopology clusterMap = shouldSortDatanodes + && keyManager.isSortDatanodesForWriteEnabled() + ? ozoneManager.getClusterMapAllowNull() : null; if (!shouldSortDatanodes) { scmClientMachine = ""; omClientMachine = ""; sortedByNodes = null; - } else if (keyManager.isSortDatanodesForWriteEnabled()) { + } else if (clusterMap != null) { scmClientMachine = ""; omClientMachine = userInfo.getRemoteAddress(); sortedByNodes = new HashMap<>(); 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 b6c29ef9a71f..76d9e9ed78bc 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 @@ -47,6 +47,7 @@ import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor; 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.pipeline.Pipeline; import org.apache.hadoop.hdds.scm.pipeline.PipelineID; import org.apache.hadoop.ozone.OzoneConsts; @@ -279,6 +280,7 @@ public void testAllocateBlockDoesNotSendClientMachineToScm() throws Exception { when(mockKeyManager.sortDatanodesForWrite(any(), any())) .thenAnswer(inv -> inv.getArgument(0)); when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); + when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); OMAllocateBlockRequest request = getOmAllocateBlockRequest(createAllocateBlockRequestWithSort("1.2.3.4")); @@ -290,6 +292,23 @@ public void testAllocateBlockDoesNotSendClientMachineToScm() throws Exception { assertEquals("", clientMachine.getValue()); } + @Test + public void testAllocateBlockFallsBackToScmWhenTopologyUnavailable() throws Exception { + KeyManager mockKeyManager = mock(KeyManager.class); + when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); + when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); + + OMAllocateBlockRequest request = + getOmAllocateBlockRequest(createAllocateBlockRequestWithSort("1.2.3.4")); + request.preExecute(ozoneManager); + + ArgumentCaptor clientMachine = ArgumentCaptor.forClass(String.class); + verify(scmBlockLocationProtocol).allocateBlock(anyLong(), anyInt(), any(), + any(), any(), clientMachine.capture()); + assertEquals("1.2.3.4", clientMachine.getValue()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString()); + } + @Test public void testAllocateBlockSortsSharedPipelineOnce() throws Exception { // Two blocks on the same 3-node pipeline must be sorted once, and the @@ -326,6 +345,7 @@ public void testAllocateBlockSortsSharedPipelineOnce() throws Exception { when(mockKeyManager.sortDatanodesForWrite(any(), any())) .thenAnswer(inv -> sortedOrder); when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); + when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); OMAllocateBlockRequest request = getOmAllocateBlockRequest(createAllocateBlockRequest()); @@ -353,21 +373,6 @@ public void sortDatanodesForWriteRequiresClientMachine() { () -> keyManager.sortDatanodesForWrite(nodes, "")); } - @Test - public void sortDatanodesForWriteKeepsOrderWhenTopologyUnavailable() { - when(ozoneManager.getClusterMap()).thenThrow( - new NullPointerException("ScmBlockLocationClient must have been initialized already.")); - List nodes = Arrays.asList( - MockDatanodeDetails.randomDatanodeDetails(), - MockDatanodeDetails.randomDatanodeDetails(), - MockDatanodeDetails.randomDatanodeDetails()); - - List result = - keyManager.sortDatanodesForWrite(nodes, "1.2.3.4"); - - assertSame(nodes, result); - } - // Like createAllocateBlockRequest, but sets sortDatanodes and a UserInfo remote address for OM-side sort. private OMRequest createAllocateBlockRequestWithSort(String clientAddress) { KeyArgs keyArgs = KeyArgs.newBuilder() From 356f00785b64f85261aa3d69345a4f1c7fc806fc Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Fri, 10 Jul 2026 20:55:11 +0800 Subject: [PATCH 21/29] HDDS-15059. Keep pipeline order when client remote address is empty --- .../hadoop/ozone/om/KeyManagerImpl.java | 3 +- .../apache/hadoop/ozone/om/OzoneManager.java | 2 +- .../ozone/om/request/key/OMKeyRequest.java | 10 +++-- .../key/TestOMAllocateBlockRequest.java | 40 +++++++++++++++++++ 4 files changed, 50 insertions(+), 5 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 03f2182b7483..65a48883174c 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 @@ -2209,7 +2209,8 @@ public List sortDatanodes(List sortDatanodesForWrite( List nodes, String clientMachine) { - Preconditions.checkArgument(!StringUtils.isEmpty(clientMachine), clientMachine); + Preconditions.checkArgument(!StringUtils.isEmpty(clientMachine), + "clientMachine is empty"); final NetworkTopology clusterMap = ozoneManager.getClusterMap(); return captureLatencyNs( metrics.getAllocateBlockSortDatanodesLatencyNs(), () -> { diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java index da123ec32883..13198fbcbc30 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java @@ -1344,7 +1344,7 @@ public void setScmTopologyClient( public NetworkTopology getClusterMap() { return requireNonNull(getClusterMapAllowNull(), - "ScmBlockLocationClient must have been initialized already."); + "OM topology cache has not been initialized yet."); } public NetworkTopology getClusterMapAllowNull() { 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 d93c78c1bf34..515d5e274be8 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 @@ -202,6 +202,7 @@ protected List allocateBlock( final String scmClientMachine; final String omClientMachine; final Map, List> sortedByNodes; + final String remoteAddress = userInfo.getRemoteAddress(); final NetworkTopology clusterMap = shouldSortDatanodes && keyManager.isSortDatanodesForWriteEnabled() ? ozoneManager.getClusterMapAllowNull() : null; @@ -209,12 +210,15 @@ protected List allocateBlock( scmClientMachine = ""; omClientMachine = ""; sortedByNodes = null; - } else if (clusterMap != null) { + } else if (clusterMap != null && !remoteAddress.isEmpty()) { + // Sort in OM: SCM skips sorting (empty machine), OM sorts by remoteAddress. scmClientMachine = ""; - omClientMachine = userInfo.getRemoteAddress(); + omClientMachine = remoteAddress; sortedByNodes = new HashMap<>(); } else { - scmClientMachine = userInfo.getRemoteAddress(); + // Sort in SCM (or keep order when remoteAddress is empty, since SCM skips + // sorting for an empty client machine). + scmClientMachine = remoteAddress; omClientMachine = ""; sortedByNodes = null; } 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 76d9e9ed78bc..a4c1739bb1b0 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 @@ -363,6 +363,46 @@ public void testAllocateBlockSortsSharedPipelineOnce() throws Exception { } } + @Test + public void testAllocateBlockKeepsOrderWhenRemoteAddressEmpty() throws Exception { + // Sort enabled and topology available, but the client has no remote address: + // OM must not sort, SCM receives an empty clientMachine, and the pipeline + // order is preserved. + List nodes = Arrays.asList( + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()); + Pipeline pipeline = Pipeline.newBuilder() + .setState(Pipeline.PipelineState.OPEN) + .setId(PipelineID.randomId()) + .setReplicationConfig( + StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE)) + .setNodes(nodes) + .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(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("").build(), ozoneManager); + + assertEquals("", clientMachine.getValue()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString()); + assertEquals(1, locations.size()); + assertEquals(nodes, locations.get(0).getPipeline().getNodes()); + } + @Test public void sortDatanodesForWriteRequiresClientMachine() { List nodes = Arrays.asList( From 0229ec399049aef38fd9830ac23e41ec23922dd6 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Fri, 10 Jul 2026 21:09:29 +0800 Subject: [PATCH 22/29] HDDS-15059. Key allocateBlock write-sort cache by pipeline id --- .../ozone/om/request/key/OMKeyRequest.java | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) 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 515d5e274be8..fb4872244bcc 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.exceptions.SCMException; import org.apache.hadoop.hdds.scm.net.NetworkTopology; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; +import org.apache.hadoop.hdds.scm.pipeline.PipelineID; import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier; import org.apache.hadoop.hdds.utils.db.cache.CacheKey; import org.apache.hadoop.hdds.utils.db.cache.CacheValue; @@ -201,7 +202,10 @@ protected List allocateBlock( final String scmClientMachine; final String omClientMachine; - final Map, List> sortedByNodes; + // Sorted order cached per pipeline so blocks sharing a pipeline are sorted + // once (mirrors the read path's caching). Keyed by pipeline id so it is + // robust to node-list ordering or copying. + final Map> sortedByPipeline; final String remoteAddress = userInfo.getRemoteAddress(); final NetworkTopology clusterMap = shouldSortDatanodes && keyManager.isSortDatanodesForWriteEnabled() @@ -209,18 +213,18 @@ protected List allocateBlock( if (!shouldSortDatanodes) { scmClientMachine = ""; omClientMachine = ""; - sortedByNodes = null; + sortedByPipeline = null; } else if (clusterMap != null && !remoteAddress.isEmpty()) { // Sort in OM: SCM skips sorting (empty machine), OM sorts by remoteAddress. scmClientMachine = ""; omClientMachine = remoteAddress; - sortedByNodes = new HashMap<>(); + sortedByPipeline = new HashMap<>(); } else { // Sort in SCM (or keep order when remoteAddress is empty, since SCM skips // sorting for an empty client machine). scmClientMachine = remoteAddress; omClientMachine = ""; - sortedByNodes = null; + sortedByPipeline = null; } List locationInfos = new ArrayList<>(numBlocks); @@ -237,16 +241,14 @@ protected List allocateBlock( } throw ex; } - // Cache the sorted order by pipeline nodes so blocks whose pipelines have - // the same datanodes are sorted once (mirrors the read path's caching). for (AllocatedBlock allocatedBlock : allocatedBlocks) { BlockID blockID = new BlockID(allocatedBlock.getBlockID()); Pipeline pipeline = allocatedBlock.getPipeline(); - if (sortedByNodes != null) { + if (sortedByPipeline != null) { final List nodes = pipeline.getNodes(); - final List sorted = sortedByNodes - .computeIfAbsent(nodes, - n -> keyManager.sortDatanodesForWrite(n, omClientMachine)); + final List sorted = sortedByPipeline + .computeIfAbsent(pipeline.getId(), + id -> keyManager.sortDatanodesForWrite(nodes, omClientMachine)); if (!Objects.equals(sorted, pipeline.getNodesInOrder())) { pipeline = pipeline.copyWithNodesInOrder(sorted); } From 3b9e45643b09d266cd8aa1baef3f581d34ac8d9c Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Fri, 10 Jul 2026 21:14:10 +0800 Subject: [PATCH 23/29] HDDS-15059. Key read-path sort cache by pipeline id for consistency --- .../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 65a48883174c..4343748b16e8 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 @@ -127,6 +127,7 @@ import org.apache.hadoop.hdds.scm.net.Node; import org.apache.hadoop.hdds.scm.net.NodeImpl; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; +import org.apache.hadoop.hdds.scm.pipeline.PipelineID; import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo; import org.apache.hadoop.hdds.security.token.OzoneBlockTokenSecretManager; import org.apache.hadoop.hdds.utils.BackgroundService; @@ -2165,7 +2166,7 @@ private void sortDatanodes(String clientMachine, OmKeyInfo keyInfo) { private void sortDatanodes(String clientMachine, List keyInfos) { if (keyInfos != null && clientMachine != null) { - final Map, List> sortedPipelines = new HashMap<>(); + final Map> sortedPipelines = new HashMap<>(); for (OmKeyInfo keyInfo : keyInfos) { OmKeyLocationInfoGroup key = keyInfo.getLatestVersionLocations(); if (key == null) { @@ -2180,11 +2181,11 @@ private void sortDatanodes(String clientMachine, List keyInfos) { continue; } - List sortedNodes = sortedPipelines.get(nodes); + List sortedNodes = sortedPipelines.get(pipeline.getId()); if (sortedNodes == null) { sortedNodes = sortDatanodes(nodes, clientMachine); if (sortedNodes != null) { - sortedPipelines.put(nodes, sortedNodes); + sortedPipelines.put(pipeline.getId(), sortedNodes); } } else if (LOG.isDebugEnabled()) { LOG.debug("Found sorted datanodes for pipeline {} and client {} " From 49f4b0286a85cc35d96c5cbccbecc84811ed6e3e Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Fri, 10 Jul 2026 21:26:02 +0800 Subject: [PATCH 24/29] HDDS-15059. Assert write order via getNodesInOrder and clarify client match comment --- .../main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java | 5 +++-- .../ozone/om/request/key/TestOMAllocateBlockRequest.java | 4 +++- 2 files changed, 6 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 4343748b16e8..052e44038ab5 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 @@ -2263,8 +2263,9 @@ private List sortByClusterMapDistance( private Node getClientNode(String clientMachine, List nodes, NetworkTopology clusterMap) { for (DatanodeDetails node : nodes) { - // Match by either IP or hostname, like SCM's getNodesByAddress; the client - // address is always an IP even when use.datanode.hostname is enabled. + // Match by either IP or hostname, like SCM's getNodesByAddress. clientMachine + // may be a hostname on the read path; the streaming-write remoteAddress is + // typically an IP. Matching both covers use.datanode.hostname either way. if (clientMachine.equals(node.getIpAddress()) || clientMachine.equals(node.getHostName())) { // The pipeline nodes are RPC-deserialized and not linked into OM's 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 a4c1739bb1b0..7264fb90d31b 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 @@ -400,7 +400,9 @@ public void testAllocateBlockKeepsOrderWhenRemoteAddressEmpty() throws Exception assertEquals("", clientMachine.getValue()); verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString()); assertEquals(1, locations.size()); - assertEquals(nodes, locations.get(0).getPipeline().getNodes()); + // Assert the write order (nodesInOrder), which copyWithNodesInOrder would + // have changed had OM sorted; it must stay as the original pipeline order. + assertEquals(nodes, locations.get(0).getPipeline().getNodesInOrder()); } @Test From 59b6b3b0ba4e8317e074de6e16a318df99522e49 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Sat, 11 Jul 2026 14:44:43 +0800 Subject: [PATCH 25/29] HDDS-15059. Pass OM cluster map into sortDatanodesForWrite --- .../apache/hadoop/ozone/TestOMSortDatanodes.java | 8 ++++---- .../org/apache/hadoop/ozone/om/KeyManager.java | 4 +++- .../org/apache/hadoop/ozone/om/KeyManagerImpl.java | 3 +-- .../hadoop/ozone/om/request/key/OMKeyRequest.java | 2 +- .../om/request/key/TestOMAllocateBlockRequest.java | 14 +++++++------- 5 files changed, 16 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 5732f50328a8..a206276ce8fd 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 @@ -190,13 +190,13 @@ public void sortDatanodesForWriteSortsRpcDeserializedPipeline() { // The client address is normally an IP, but the sort must resolve a client // by either IP or hostname, so cover both. List byIp = - keyManager.sortDatanodesForWrite(rpcNodes, dn.getIpAddress()); + keyManager.sortDatanodesForWrite(rpcNodes, dn.getIpAddress(), om.getClusterMap()); assertEquals(dn, byIp.get(0), "Source node should be sorted first for writes (IP client)"); assertRackOrder(dn.getNetworkLocation(), byIp); List byHostname = - keyManager.sortDatanodesForWrite(rpcNodes, dn.getHostName()); + keyManager.sortDatanodesForWrite(rpcNodes, dn.getHostName(), om.getClusterMap()); assertEquals(dn, byHostname.get(0), "Source node should be sorted first for writes (hostname client)"); assertRackOrder(dn.getNetworkLocation(), byHostname); @@ -210,7 +210,7 @@ public void sortDatanodesForWriteKeepsOrderForStaleTopology() { nodes.addAll(nodeManager.getAllNodes()); List sorted = - keyManager.sortDatanodesForWrite(nodes, "edge0"); + keyManager.sortDatanodesForWrite(nodes, "edge0", om.getClusterMap()); assertSame(nodes, sorted, "Pipeline order should be preserved when a node is missing from the OM topology"); @@ -223,7 +223,7 @@ public void sortDatanodesForWriteKeepsOrderWhenClientUnresolved() { // 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); + keyManager.sortDatanodesForWrite(nodes, unresolved, om.getClusterMap()); assertEquals(original, result, "Write pipeline order must be preserved when client is unresolved"); } 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 f4061cc8b17a..96c259398d02 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 @@ -24,6 +24,7 @@ import java.util.Map; 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.utils.BackgroundService; import org.apache.hadoop.hdds.utils.db.Table; import org.apache.hadoop.hdds.utils.db.TableIterator; @@ -373,10 +374,11 @@ 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 input order if client is unresolved */ List sortDatanodesForWrite( - List nodes, String clientMachine); + List nodes, String clientMachine, 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 052e44038ab5..3b7c3e0cf4f4 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 @@ -2209,10 +2209,9 @@ public List sortDatanodes(List sortDatanodesForWrite( - List nodes, String clientMachine) { + List nodes, String clientMachine, NetworkTopology clusterMap) { Preconditions.checkArgument(!StringUtils.isEmpty(clientMachine), "clientMachine is empty"); - final NetworkTopology clusterMap = ozoneManager.getClusterMap(); return captureLatencyNs( metrics.getAllocateBlockSortDatanodesLatencyNs(), () -> { final Node client = getClientNode(clientMachine, nodes, 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 fb4872244bcc..32cb2d96a3b8 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 @@ -248,7 +248,7 @@ protected List allocateBlock( final List nodes = pipeline.getNodes(); final List sorted = sortedByPipeline .computeIfAbsent(pipeline.getId(), - id -> keyManager.sortDatanodesForWrite(nodes, omClientMachine)); + id -> keyManager.sortDatanodesForWrite(nodes, omClientMachine, clusterMap)); if (!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 7264fb90d31b..3089098c76d2 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 @@ -268,7 +268,7 @@ 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()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any()); } @Test @@ -277,7 +277,7 @@ 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())) + when(mockKeyManager.sortDatanodesForWrite(any(), any(), any())) .thenAnswer(inv -> inv.getArgument(0)); when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); @@ -306,7 +306,7 @@ 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()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any()); } @Test @@ -342,7 +342,7 @@ public void testAllocateBlockSortsSharedPipelineOnce() throws Exception { Collections.reverse(sortedOrder); KeyManager mockKeyManager = mock(KeyManager.class); when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); - when(mockKeyManager.sortDatanodesForWrite(any(), any())) + when(mockKeyManager.sortDatanodesForWrite(any(), any(), any())) .thenAnswer(inv -> sortedOrder); when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); @@ -355,7 +355,7 @@ 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")); + verify(mockKeyManager, times(1)).sortDatanodesForWrite(any(), eq("1.2.3.4"), any()); // ...and the sorted order is applied to every block's pipeline. assertEquals(2, locations.size()); for (OmKeyLocationInfo location : locations) { @@ -398,7 +398,7 @@ public void testAllocateBlockKeepsOrderWhenRemoteAddressEmpty() throws Exception UserInfo.newBuilder().setRemoteAddress("").build(), ozoneManager); assertEquals("", clientMachine.getValue()); - verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), 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. @@ -412,7 +412,7 @@ public void sortDatanodesForWriteRequiresClientMachine() { MockDatanodeDetails.randomDatanodeDetails(), MockDatanodeDetails.randomDatanodeDetails()); assertThrows(IllegalArgumentException.class, - () -> keyManager.sortDatanodesForWrite(nodes, "")); + () -> keyManager.sortDatanodesForWrite(nodes, "", mock(NetworkTopology.class))); } // Like createAllocateBlockRequest, but sets sortDatanodes and a UserInfo remote address for OM-side sort. From 7307ae201520166a2abe8594aa7b3cb61f95c7ea Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Sat, 11 Jul 2026 14:45:57 +0800 Subject: [PATCH 26/29] HDDS-15059. Key write-sort cache by datanode UUID set, matching read path --- .../hadoop/ozone/om/KeyManagerImpl.java | 10 +++++--- .../ozone/om/request/key/OMKeyRequest.java | 25 ++++++++++--------- 2 files changed, 19 insertions(+), 16 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 3b7c3e0cf4f4..6c0fc0a5f000 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 @@ -127,7 +127,6 @@ import org.apache.hadoop.hdds.scm.net.Node; import org.apache.hadoop.hdds.scm.net.NodeImpl; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; -import org.apache.hadoop.hdds.scm.pipeline.PipelineID; import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo; import org.apache.hadoop.hdds.security.token.OzoneBlockTokenSecretManager; import org.apache.hadoop.hdds.utils.BackgroundService; @@ -2166,7 +2165,7 @@ private void sortDatanodes(String clientMachine, OmKeyInfo keyInfo) { private void sortDatanodes(String clientMachine, List keyInfos) { if (keyInfos != null && clientMachine != null) { - final Map> sortedPipelines = new HashMap<>(); + final Map, List> sortedPipelines = new HashMap<>(); for (OmKeyInfo keyInfo : keyInfos) { OmKeyLocationInfoGroup key = keyInfo.getLatestVersionLocations(); if (key == null) { @@ -2181,11 +2180,14 @@ private void sortDatanodes(String clientMachine, List keyInfos) { continue; } - List sortedNodes = sortedPipelines.get(pipeline.getId()); + final Set uuidSet = nodes.stream().map(DatanodeDetails::getUuidString) + .collect(Collectors.toSet()); + + List sortedNodes = sortedPipelines.get(uuidSet); if (sortedNodes == null) { sortedNodes = sortDatanodes(nodes, clientMachine); if (sortedNodes != null) { - sortedPipelines.put(pipeline.getId(), sortedNodes); + sortedPipelines.put(uuidSet, sortedNodes); } } else if (LOG.isDebugEnabled()) { LOG.debug("Found sorted datanodes for pipeline {} and client {} " 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 32cb2d96a3b8..47409071a017 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,7 +64,6 @@ import org.apache.hadoop.hdds.scm.exceptions.SCMException; import org.apache.hadoop.hdds.scm.net.NetworkTopology; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; -import org.apache.hadoop.hdds.scm.pipeline.PipelineID; import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier; import org.apache.hadoop.hdds.utils.db.cache.CacheKey; import org.apache.hadoop.hdds.utils.db.cache.CacheValue; @@ -202,10 +201,10 @@ protected List allocateBlock( final String scmClientMachine; final String omClientMachine; - // Sorted order cached per pipeline so blocks sharing a pipeline are sorted - // once (mirrors the read path's caching). Keyed by pipeline id so it is - // robust to node-list ordering or copying. - final Map> sortedByPipeline; + // 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() @@ -213,18 +212,18 @@ protected List allocateBlock( if (!shouldSortDatanodes) { scmClientMachine = ""; omClientMachine = ""; - sortedByPipeline = null; + sortedByNodes = null; } else if (clusterMap != null && !remoteAddress.isEmpty()) { // Sort in OM: SCM skips sorting (empty machine), OM sorts by remoteAddress. scmClientMachine = ""; omClientMachine = remoteAddress; - sortedByPipeline = new HashMap<>(); + sortedByNodes = new HashMap<>(); } else { // Sort in SCM (or keep order when remoteAddress is empty, since SCM skips // sorting for an empty client machine). scmClientMachine = remoteAddress; omClientMachine = ""; - sortedByPipeline = null; + sortedByNodes = null; } List locationInfos = new ArrayList<>(numBlocks); @@ -244,11 +243,13 @@ protected List allocateBlock( for (AllocatedBlock allocatedBlock : allocatedBlocks) { BlockID blockID = new BlockID(allocatedBlock.getBlockID()); Pipeline pipeline = allocatedBlock.getPipeline(); - if (sortedByPipeline != null) { + if (sortedByNodes != null) { final List nodes = pipeline.getNodes(); - final List sorted = sortedByPipeline - .computeIfAbsent(pipeline.getId(), - id -> keyManager.sortDatanodesForWrite(nodes, omClientMachine, clusterMap)); + final Set uuidSet = nodes.stream() + .map(DatanodeDetails::getUuidString).collect(Collectors.toSet()); + final List sorted = sortedByNodes + .computeIfAbsent(uuidSet, + k -> keyManager.sortDatanodesForWrite(nodes, omClientMachine, clusterMap)); if (!Objects.equals(sorted, pipeline.getNodesInOrder())) { pipeline = pipeline.copyWithNodesInOrder(sorted); } From de7d86c59634be82f3625a85f644fc2495f0d82f Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Sat, 11 Jul 2026 15:30:13 +0800 Subject: [PATCH 27/29] HDDS-15059. Cache datanode sort only when an actual sort happens --- .../hadoop/ozone/om/KeyManagerImpl.java | 7 ++- .../ozone/om/request/key/OMKeyRequest.java | 13 ++++- .../key/TestOMAllocateBlockRequest.java | 55 +++++++++++++++++++ 3 files changed, 71 insertions(+), 4 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 6c0fc0a5f000..1f272cbf5f3d 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 @@ -2186,7 +2186,12 @@ private void sortDatanodes(String clientMachine, List keyInfos) { List sortedNodes = sortedPipelines.get(uuidSet); if (sortedNodes == null) { sortedNodes = sortDatanodes(nodes, clientMachine); - if (sortedNodes != null) { + // 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. + if (sortedNodes != null && sortedNodes != nodes) { sortedPipelines.put(uuidSet, sortedNodes); } } else if (LOG.isDebugEnabled()) { 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 47409071a017..9f28a7692ca5 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 @@ -247,9 +247,16 @@ protected List allocateBlock( final List nodes = pipeline.getNodes(); final Set uuidSet = nodes.stream() .map(DatanodeDetails::getUuidString).collect(Collectors.toSet()); - final List sorted = sortedByNodes - .computeIfAbsent(uuidSet, - k -> keyManager.sortDatanodesForWrite(nodes, omClientMachine, clusterMap)); + 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) { + sortedByNodes.put(uuidSet, sorted); + } + } if (!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 3089098c76d2..0c2a593cfe8a 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 @@ -363,6 +363,61 @@ 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. + 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); + 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(); + when(scmBlockLocationProtocol.allocateBlock(anyLong(), anyInt(), any(), + anyString(), any(ExcludeList.class), anyString())) + .thenReturn(Arrays.asList(block1, block2)); + + 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)); + 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()); + // 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()); + } + @Test public void testAllocateBlockKeepsOrderWhenRemoteAddressEmpty() throws Exception { // Sort enabled and topology available, but the client has no remote address: From 93b0da98bf33c5f264128890e61d9f8407f27958 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Sat, 11 Jul 2026 15:44:15 +0800 Subject: [PATCH 28/29] HDDS-15059. Null-check clusterMap and document instance-return contract --- .../src/main/java/org/apache/hadoop/ozone/om/KeyManager.java | 4 +++- .../main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java | 1 + 2 files changed, 4 insertions(+), 1 deletion(-) 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 96c259398d02..320536dab2e4 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 @@ -375,7 +375,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 input order if client is unresolved + * @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 */ 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 1f272cbf5f3d..658f78f68d62 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 @@ -2219,6 +2219,7 @@ public List sortDatanodesForWrite( List nodes, String clientMachine, NetworkTopology clusterMap) { Preconditions.checkArgument(!StringUtils.isEmpty(clientMachine), "clientMachine is empty"); + Preconditions.checkNotNull(clusterMap, "clusterMap is null"); return captureLatencyNs( metrics.getAllocateBlockSortDatanodesLatencyNs(), () -> { final Node client = getClientNode(clientMachine, nodes, clusterMap); From fc381eaed5f4aced0bd27b2679ce98e70fa08546 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Sun, 12 Jul 2026 11:11:57 +0800 Subject: [PATCH 29/29] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../ozone/om/request/key/TestOMAllocateBlockRequest.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 0c2a593cfe8a..a584fd7b1b25 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 @@ -470,7 +470,8 @@ public void sortDatanodesForWriteRequiresClientMachine() { () -> keyManager.sortDatanodesForWrite(nodes, "", mock(NetworkTopology.class))); } - // Like createAllocateBlockRequest, but sets sortDatanodes and a UserInfo remote address for OM-side sort. + // Like createAllocateBlockRequest, but sets sortDatanodes and a UserInfo remote + // address for OM-side sort. private OMRequest createAllocateBlockRequestWithSort(String clientAddress) { KeyArgs keyArgs = KeyArgs.newBuilder() .setVolumeName(volumeName).setBucketName(bucketName).setKeyName(keyName)