diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OmConfig.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OmConfig.java index c981afd414d8..33e1b511ca04 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OmConfig.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OmConfig.java @@ -192,7 +192,8 @@ public class OmConfig extends ReconfigurableConfig { "datanode first) locally using its cached cluster topology, instead " + "of asking SCM to sort on every allocateBlock. Defaults to false so " + "SCM performs the sort. Enable this to offload the sort from SCM " + - "when multiple OM services share a single SCM service." + "when multiple OM services share a single SCM service. SCM still sorts if OM has no cached topology " + + "or cannot resolve the client. If OM's topology lacks a pipeline datanode, the pipeline order is preserved." ) private boolean sortDatanodesForWriteEnabled; diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java index a206276ce8fd..11c5be13b143 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java @@ -22,7 +22,8 @@ import static org.apache.hadoop.hdds.scm.net.NetConstants.ROOT_LEVEL; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotEquals; -import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.mockito.Mockito.mock; import com.google.common.collect.ImmutableMap; @@ -30,12 +31,14 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.hadoop.hdds.HddsConfigKeys; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.scm.HddsTestUtils; import org.apache.hadoop.hdds.scm.ha.SCMContext; import org.apache.hadoop.hdds.scm.ha.SCMHAManagerStub; +import org.apache.hadoop.hdds.scm.net.Node; import org.apache.hadoop.hdds.scm.node.NodeManager; import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocol; import org.apache.hadoop.hdds.scm.server.SCMConfigurator; @@ -77,7 +80,7 @@ public static void setup() throws Exception { config = new OzoneConfiguration(); config.set(HddsConfigKeys.OZONE_METADATA_DIRS, dir.toString()); config.set(NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY, - StaticMapping.class.getName()); + CountingStaticMapping.class.getName()); config.set(OzoneConfigKeys.OZONE_NETWORK_TOPOLOGY_AWARE_READ_KEY, "true"); List datanodes = new ArrayList<>(NODE_COUNT); List nodeMapping = new ArrayList<>(NODE_COUNT); @@ -189,14 +192,16 @@ public void sortDatanodesForWriteSortsRpcDeserializedPipeline() { for (DatanodeDetails dn : nodeManager.getAllNodes()) { // The client address is normally an IP, but the sort must resolve a client // by either IP or hostname, so cover both. + Node ipClient = keyManager.resolveClientForWrite(dn.getIpAddress(), om.getClusterMap()); List byIp = - keyManager.sortDatanodesForWrite(rpcNodes, dn.getIpAddress(), om.getClusterMap()); + keyManager.sortDatanodesForWrite(rpcNodes, dn.getIpAddress(), ipClient, om.getClusterMap()); assertEquals(dn, byIp.get(0), "Source node should be sorted first for writes (IP client)"); assertRackOrder(dn.getNetworkLocation(), byIp); + Node hostClient = keyManager.resolveClientForWrite(dn.getHostName(), om.getClusterMap()); List byHostname = - keyManager.sortDatanodesForWrite(rpcNodes, dn.getHostName(), om.getClusterMap()); + keyManager.sortDatanodesForWrite(rpcNodes, dn.getHostName(), hostClient, om.getClusterMap()); assertEquals(dn, byHostname.get(0), "Source node should be sorted first for writes (hostname client)"); assertRackOrder(dn.getNetworkLocation(), byHostname); @@ -204,28 +209,106 @@ public void sortDatanodesForWriteSortsRpcDeserializedPipeline() { } @Test - public void sortDatanodesForWriteKeepsOrderForStaleTopology() { + public void sortDatanodesForWriteReturnsNullForStaleTopology() { List nodes = new ArrayList<>(); nodes.add(randomDatanodeDetails()); nodes.addAll(nodeManager.getAllNodes()); + Node client = keyManager.resolveClientForWrite("edge0", om.getClusterMap()); + assertNotNull(client); - List sorted = - keyManager.sortDatanodesForWrite(nodes, "edge0", om.getClusterMap()); + assertNull(keyManager.sortDatanodesForWrite(nodes, "edge0", client, om.getClusterMap()), + "Sort must be skipped when a node is missing from the OM topology"); + } - assertSame(nodes, sorted, - "Pipeline order should be preserved when a node is missing from the OM topology"); + @Test + public void resolveClientForWriteReturnsNullWhenUnresolved() { + // A client that maps to no rack in the OM topology cannot be placed, so OM + // must report null and let the caller fall back to SCM. + String unresolved = nodeManager.getAllNodes().get(0).getIpAddress() + "X"; + assertNull(keyManager.resolveClientForWrite(unresolved, om.getClusterMap())); } @Test - public void 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, om.getClusterMap()); - assertEquals(original, result, - "Write pipeline order must be preserved when client is unresolved"); + public void resolveClientForWriteAttachesNonDatanodeClientToItsRack() { + for (Map.Entry entry : EDGE_NODES.entrySet()) { + Node client = keyManager.resolveClientForWrite(entry.getKey(), om.getClusterMap()); + assertNotNull(client, "Edge client should resolve via the DNS-to-switch mapping"); + assertEquals(entry.getValue(), client.getNetworkLocation()); + } + } + + @Test + public void sortDatanodesForWritePrefersPipelineDatanodeAsClient() { + // Resolve the client in the other rack so preferring the matching pipeline datanode is deterministic. + List all = nodeManager.getAllNodes(); + DatanodeDetails clientDn = all.get(0); + DatanodeDetails sameRack = null; + List otherRack = new ArrayList<>(); + for (DatanodeDetails dn : all) { + if (dn.equals(clientDn)) { + continue; + } + if (dn.getNetworkLocation().equals(clientDn.getNetworkLocation())) { + if (sameRack == null) { + sameRack = dn; + } + } else if (otherRack.size() < 2) { + otherRack.add(dn); + } + } + assertNotNull(sameRack); + assertEquals(2, otherRack.size()); + String otherRackEdge = "/rack0".equals(clientDn.getNetworkLocation()) ? "edge1" : "edge0"; + Node otherRackClient = keyManager.resolveClientForWrite(otherRackEdge, om.getClusterMap()); + assertNotNull(otherRackClient); + assertNotEquals(clientDn.getNetworkLocation(), otherRackClient.getNetworkLocation()); + + for (String address : new String[] {clientDn.getIpAddress(), clientDn.getHostName()}) { + // Without a matching datanode, sort relative to the supplied client. + List without = new ArrayList<>(); + without.add(rpcCopy(sameRack)); + without.add(rpcCopy(otherRack.get(0))); + without.add(rpcCopy(otherRack.get(1))); + List sortedWithout = + keyManager.sortDatanodesForWrite(without, address, otherRackClient, om.getClusterMap()); + assertNotNull(sortedWithout); + assertEquals(otherRackClient.getNetworkLocation(), sortedWithout.get(0).getNetworkLocation()); + + // A matching datanode takes precedence over the supplied client and sorts first. + List with = new ArrayList<>(); + with.add(rpcCopy(otherRack.get(0))); + with.add(rpcCopy(sameRack)); + with.add(rpcCopy(clientDn)); + List sortedWith = + keyManager.sortDatanodesForWrite(with, address, otherRackClient, om.getClusterMap()); + assertNotNull(sortedWith); + assertEquals(clientDn, sortedWith.get(0)); + } + } + + @Test + public void resolveClientForWriteIsTheOnlyMappingLookup() { + List all = nodeManager.getAllNodes(); + CountingStaticMapping.RESOLVED_NAMES.set(0); + Node client = keyManager.resolveClientForWrite("edge0", om.getClusterMap()); + assertNotNull(client); + assertEquals(1, CountingStaticMapping.RESOLVED_NAMES.get()); + + for (int start = 0; start + 3 <= all.size(); start += 3) { + List pipeline = new ArrayList<>(); + for (DatanodeDetails dn : all.subList(start, start + 3)) { + pipeline.add(rpcCopy(dn)); + } + assertNotNull(keyManager.sortDatanodesForWrite(pipeline, "edge0", client, om.getClusterMap())); + } + assertEquals(1, CountingStaticMapping.RESOLVED_NAMES.get(), + "sortDatanodesForWrite must not call the DNS-to-switch mapping"); + } + + // Simulate a pipeline node as OM receives it from SCM over RPC: same identity, + // no topology linkage. + private static DatanodeDetails rpcCopy(DatanodeDetails dn) { + return DatanodeDetails.getFromProtoBuf(dn.getProtoBufMessage()); } private String nodeAddress(DatanodeDetails dn) { @@ -234,4 +317,15 @@ private String nodeAddress(DatanodeDetails dn) { HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME_DEFAULT); return useHostname ? dn.getHostName() : dn.getIpAddress(); } + + /** StaticMapping that counts how many names were resolved. */ + public static class CountingStaticMapping extends StaticMapping { + static final AtomicInteger RESOLVED_NAMES = new AtomicInteger(); + + @Override + public List resolve(List names) { + RESOLVED_NAMES.addAndGet(names.size()); + return super.resolve(names); + } + } } diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java index 2c959cb4c018..79f74a7f9812 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java @@ -25,6 +25,7 @@ import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.scm.net.NetworkTopology; +import org.apache.hadoop.hdds.scm.net.Node; import org.apache.hadoop.hdds.utils.BackgroundService; import org.apache.hadoop.hdds.utils.db.Table; import org.apache.hadoop.hdds.utils.db.TableIterator; @@ -402,20 +403,27 @@ DeleteKeysResult getPendingDeletionSubFiles(long volumeId, long bucketId, OmKeyI KeyLifecycleService getKeyLifecycleService(); /** - * 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. + * Resolve the streaming-write client's rack in OM's cached cluster map using the DNS-to-switch mapping. + * + * @param clientMachine client address (IP or hostname), must not be empty + * @param clusterMap OM's cached cluster map + * @return a node attached to the client's rack, or null if unresolved so the caller can let SCM sort + */ + Node resolveClientForWrite(String clientMachine, NetworkTopology clusterMap); + + /** + * Sort write pipeline datanodes nearest-first using OM's cached cluster map. A datanode matching + * {@code clientMachine} takes precedence over {@code client} and sorts first. * * @param nodes the pipeline nodes to sort - * @param clientMachine client address (IP or hostname) + * @param clientMachine client address (IP or hostname), must not be empty + * @param client the node returned by {@link #resolveClientForWrite}, must not be null * @param clusterMap OM's cached cluster map used to resolve topology distance - * @return nodes sorted nearest-first, or the original {@code nodes} list - * instance unchanged when sorting is skipped (client unresolved or stale - * topology); callers may use reference equality to detect a skipped sort + * @return nodes sorted nearest-first, or null if a pipeline node is missing from the cluster map; + * callers must preserve the pipeline's order in that case */ List sortDatanodesForWrite( - List nodes, String clientMachine, NetworkTopology clusterMap); + List nodes, String clientMachine, Node client, NetworkTopology clusterMap); /** * @return true if OM should sort the streaming-write pipeline locally diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java index d74614579e24..3547b9803788 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java @@ -2283,11 +2283,6 @@ private void sortDatanodes(String clientMachine, List keyInfos) { List sortedNodes = sortedPipelines.get(uuidSet); if (sortedNodes == null) { sortedNodes = sortDatanodes(nodes, clientMachine); - // Cache only a freshly sorted order, not an input list returned - // unchanged when no sort happens: that order is per-pipeline and must - // not be reused for another pipeline with the same node set. The read - // sort always returns a new list, so this never skips caching here; it - // keeps the pattern identical to the write path. if (sortedNodes != null && sortedNodes != nodes) { sortedPipelines.put(uuidSet, sortedNodes); } @@ -2312,21 +2307,26 @@ public List sortDatanodes(List getOtherNode(clientMachine, clusterMap)); + } + @Override public List sortDatanodesForWrite( - List nodes, String clientMachine, NetworkTopology clusterMap) { - Preconditions.checkArgument(!StringUtils.isEmpty(clientMachine), - "clientMachine is empty"); + List nodes, String clientMachine, Node client, NetworkTopology clusterMap) { + Preconditions.checkArgument(!StringUtils.isEmpty(clientMachine), "clientMachine is empty"); + Objects.requireNonNull(client, "client is null"); Objects.requireNonNull(clusterMap, "clusterMap is null"); return captureLatencyNs( metrics.getAllocateBlockSortDatanodesLatencyNs(), () -> { - final Node client = getClientNode(clientMachine, nodes, clusterMap); - if (client == null) { - // 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(clusterMap, client, nodes); + // A pipeline datanode matching the client is at distance zero from + // itself, so it sorts first; the rack-level client cannot do that. + final Node pipelineClient = findClientInPipeline(clientMachine, nodes, clusterMap); + return sortByClusterMapDistance(clusterMap, pipelineClient != null ? pipelineClient : client, nodes); }); } @@ -2342,6 +2342,8 @@ public boolean isSortDatanodesForWriteEnabled() { * {@link Integer#MAX_VALUE}) and the order comes out random. Look each node * up in OM's cluster map to get the topology-linked instance, sort those, * then map the order back to the original nodes. + * + * @return the sorted nodes, or null when a node is missing from the cluster map */ private List sortByClusterMapDistance( NetworkTopology clusterMap, Node client, @@ -2351,7 +2353,7 @@ private List sortByClusterMapDistance( for (DatanodeDetails node : nodes) { final Node resolved = clusterMap.getNode(node.getNetworkFullPath()); if (resolved == null) { - return nodes; + return null; } topologyNodes.add(resolved); nodeByPath.put(resolved.getNetworkFullPath(), node); @@ -2367,6 +2369,12 @@ private List sortByClusterMapDistance( private Node getClientNode(String clientMachine, List nodes, NetworkTopology clusterMap) { + final Node pipelineClient = findClientInPipeline(clientMachine, nodes, clusterMap); + return pipelineClient != null ? pipelineClient : getOtherNode(clientMachine, clusterMap); + } + + private Node findClientInPipeline(String clientMachine, + List nodes, NetworkTopology clusterMap) { for (DatanodeDetails node : nodes) { // Match by either IP or hostname, like SCM's getNodesByAddress. clientMachine // may be a hostname on the read path; the streaming-write remoteAddress is @@ -2379,7 +2387,7 @@ private Node getClientNode(String clientMachine, return resolved != null ? resolved : node; } } - return getOtherNode(clientMachine, clusterMap); + return null; } private Node getOtherNode(String clientMachine, diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMPerformanceMetrics.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMPerformanceMetrics.java index da687ae193c4..68b4c5929eca 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMPerformanceMetrics.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMPerformanceMetrics.java @@ -138,7 +138,7 @@ public OMPerformanceMetrics() { getKeyInfoSortDatanodesLatencyNs = stat("GetKeyInfoSortDatanodesLatencyNs", "Sort datanodes latency in getKeyInfo"); allocateBlockSortDatanodesLatencyNs = stat("AllocateBlockSortDatanodesLatencyNs", - "Sort datanodes latency in allocateBlock (streaming write)"); + "Client lookup and datanode sort latency in allocateBlock (streaming write), recorded as separate samples"); getKeyInfoResolveBucketLatencyNs = stat("GetKeyInfoResolveBucketLatencyNs", "resolveBucketLink latency in getKeyInfo"); s3VolumeContextLatencyNs = stat("S3VolumeContextLatencyNs", diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java index 3d478e461396..4f89f982d69c 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java @@ -64,6 +64,7 @@ import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList; import org.apache.hadoop.hdds.scm.exceptions.SCMException; import org.apache.hadoop.hdds.scm.net.NetworkTopology; +import org.apache.hadoop.hdds.scm.net.Node; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier; import org.apache.hadoop.hdds.utils.db.cache.CacheKey; @@ -200,32 +201,22 @@ protected List allocateBlock( final int numBlocks = (int) Math.min(ozoneManager.getPreallocateBlocksMax(), (requestedSize - 1) / (scmBlockSize * dataGroupSize) + 1); - final String scmClientMachine; - final String omClientMachine; - // Sorted order cached by datanode set so blocks whose pipelines share the - // same datanodes are sorted once (mirrors the read path's caching). Keyed by - // the UUID set so it is order-insensitive and dedups across pipelines. - final Map, List> sortedByNodes; final String remoteAddress = userInfo.getRemoteAddress(); final NetworkTopology clusterMap = shouldSortDatanodes && keyManager.isSortDatanodesForWriteEnabled() ? ozoneManager.getClusterMapAllowNull() : null; - if (!shouldSortDatanodes) { - scmClientMachine = ""; - omClientMachine = ""; - sortedByNodes = 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<>(); - } 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; - } + // Resolve the client once per request, before the block loop, so the + // DNS-to-switch mapping is consulted at most once. + final Node omClient = clusterMap != null && !remoteAddress.isEmpty() + ? keyManager.resolveClientForWrite(remoteAddress, clusterMap) : null; + // Let SCM sort when OM sorting is disabled or its topology/client lookup is unavailable. + // An empty client address tells SCM to skip sorting. + final String scmClientMachine = shouldSortDatanodes && omClient == null ? remoteAddress : ""; + // Sorted order cached by datanode set so blocks whose pipelines share the + // same datanodes are sorted once (mirrors the read path's caching). Keyed by + // the UUID set so it is order-insensitive and dedups across pipelines. + final Map, List> sortedByNodes = + omClient != null ? new HashMap<>() : null; List locationInfos = new ArrayList<>(numBlocks); String remoteUser = getRemoteUser().getShortUserName(); @@ -250,15 +241,13 @@ protected List allocateBlock( .map(DatanodeDetails::getUuidString).collect(Collectors.toSet()); 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) { + sorted = keyManager.sortDatanodesForWrite(nodes, remoteAddress, omClient, clusterMap); + // Cache only sorted results; a skipped sort must preserve each pipeline's own order. + if (sorted != null) { sortedByNodes.put(uuidSet, sorted); } } - if (!Objects.equals(sorted, pipeline.getNodesInOrder())) { + if (sorted != null && !Objects.equals(sorted, pipeline.getNodesInOrder())) { pipeline = pipeline.copyWithNodesInOrder(sorted); } } diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java index f389717907c4..186a808b1c6b 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java @@ -50,6 +50,7 @@ import org.apache.hadoop.hdds.scm.container.common.helpers.AllocatedBlock; import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList; import org.apache.hadoop.hdds.scm.net.NetworkTopology; +import org.apache.hadoop.hdds.scm.net.Node; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.hdds.scm.pipeline.PipelineID; import org.apache.hadoop.ipc_.Server; @@ -273,7 +274,8 @@ public void testAllocateBlockSendsClientMachineToScmWhenFlagOff() throws Excepti verify(scmBlockLocationProtocol).allocateBlock(anyLong(), anyInt(), any(), any(), any(), clientMachine.capture()); assertEquals("1.2.3.4", clientMachine.getValue()); - verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any()); + verify(mockKeyManager, never()).resolveClientForWrite(anyString(), any()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any(), any()); } @Test @@ -282,7 +284,8 @@ public void testAllocateBlockDoesNotSendClientMachineToScm() throws Exception { // clientMachine even when the client requests sorted datanodes. KeyManager mockKeyManager = mock(KeyManager.class); when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); - when(mockKeyManager.sortDatanodesForWrite(any(), any(), any())) + when(mockKeyManager.resolveClientForWrite(anyString(), any())).thenReturn(mock(Node.class)); + when(mockKeyManager.sortDatanodesForWrite(any(), any(), any(), any())) .thenAnswer(inv -> inv.getArgument(0)); when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); @@ -311,7 +314,109 @@ public void testAllocateBlockFallsBackToScmWhenTopologyUnavailable() throws Exce verify(scmBlockLocationProtocol).allocateBlock(anyLong(), anyInt(), any(), any(), any(), clientMachine.capture()); assertEquals("1.2.3.4", clientMachine.getValue()); - verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any()); + verify(mockKeyManager, never()).resolveClientForWrite(anyString(), any()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any(), any()); + } + + @Test + public void testAllocateBlockFallsBackToScmWhenClientUnresolved() throws Exception { + // Flag on and topology available, but OM cannot place the client in its + // topology: OM must not sort, and SCM receives the client address so it can + // try with its own node manager and mapping. + List nodes = Arrays.asList( + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()); + List scmOrder = new ArrayList<>(nodes); + Collections.reverse(scmOrder); + Pipeline pipeline = Pipeline.newBuilder() + .setState(Pipeline.PipelineState.OPEN) + .setId(PipelineID.randomId()) + .setReplicationConfig( + StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE)) + .setNodes(nodes) + .setNodesInOrder(scmOrder) + .build(); + AllocatedBlock block = new AllocatedBlock.Builder().setPipeline(pipeline) + .setContainerBlockID(new ContainerBlockID(CONTAINER_ID, LOCAL_ID)).build(); + ArgumentCaptor clientMachine = ArgumentCaptor.forClass(String.class); + when(scmBlockLocationProtocol.allocateBlock(anyLong(), anyInt(), any(), + anyString(), any(ExcludeList.class), clientMachine.capture())) + .thenReturn(Collections.singletonList(block)); + + KeyManager mockKeyManager = mock(KeyManager.class); + when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); + when(mockKeyManager.resolveClientForWrite(anyString(), any())).thenReturn(null); + when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); + when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); + + OMAllocateBlockRequest request = getOmAllocateBlockRequest(createAllocateBlockRequest()); + List locations = request.allocateBlock(replicationConfig, + new ExcludeList(), scmBlockSize, true, + UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(), ozoneManager); + + assertEquals("1.2.3.4", clientMachine.getValue()); + verify(mockKeyManager, times(1)).resolveClientForWrite(eq("1.2.3.4"), any()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any(), any()); + assertEquals(1, locations.size()); + // SCM's order is kept as-is. + assertEquals(scmOrder, locations.get(0).getPipeline().getNodesInOrder()); + } + + @Test + public void testAllocateBlockResolvesClientOnceAcrossPipelines() throws Exception { + // Two blocks on two different pipelines (different datanode sets): the + // client is resolved once for the request and reused for both sorts. + List nodes1 = Arrays.asList( + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()); + List nodes2 = Arrays.asList( + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()); + Pipeline pipeline1 = Pipeline.newBuilder() + .setState(Pipeline.PipelineState.OPEN) + .setId(PipelineID.randomId()) + .setReplicationConfig( + StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE)) + .setNodes(nodes1) + .build(); + Pipeline pipeline2 = Pipeline.newBuilder() + .setState(Pipeline.PipelineState.OPEN) + .setId(PipelineID.randomId()) + .setReplicationConfig( + StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE)) + .setNodes(nodes2) + .build(); + AllocatedBlock block1 = new AllocatedBlock.Builder().setPipeline(pipeline1) + .setContainerBlockID(new ContainerBlockID(CONTAINER_ID, LOCAL_ID)).build(); + AllocatedBlock block2 = new AllocatedBlock.Builder().setPipeline(pipeline2) + .setContainerBlockID(new ContainerBlockID(CONTAINER_ID + 1, LOCAL_ID + 1)).build(); + ArgumentCaptor clientMachine = ArgumentCaptor.forClass(String.class); + when(scmBlockLocationProtocol.allocateBlock(anyLong(), anyInt(), any(), + anyString(), any(ExcludeList.class), clientMachine.capture())) + .thenReturn(Arrays.asList(block1, block2)); + + Node client = mock(Node.class); + KeyManager mockKeyManager = mock(KeyManager.class); + when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); + when(mockKeyManager.resolveClientForWrite(eq("1.2.3.4"), any())).thenReturn(client); + when(mockKeyManager.sortDatanodesForWrite(any(), any(), any(), any())) + .thenAnswer(inv -> inv.getArgument(0)); + when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); + when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); + + OMAllocateBlockRequest request = getOmAllocateBlockRequest(createAllocateBlockRequest()); + List locations = request.allocateBlock(replicationConfig, + new ExcludeList(), 2 * scmBlockSize, true, + UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(), ozoneManager); + + assertEquals(2, locations.size()); + // OM sorts, so SCM is told not to. + assertEquals("", clientMachine.getValue()); + verify(mockKeyManager, times(1)).resolveClientForWrite(eq("1.2.3.4"), any()); + verify(mockKeyManager, times(2)).sortDatanodesForWrite(any(), eq("1.2.3.4"), eq(client), any()); } @Test @@ -345,9 +450,11 @@ public void testAllocateBlockSortsSharedPipelineOnce() throws Exception { List sortedOrder = new ArrayList<>(nodes); Collections.reverse(sortedOrder); + Node client = mock(Node.class); KeyManager mockKeyManager = mock(KeyManager.class); when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); - when(mockKeyManager.sortDatanodesForWrite(any(), any(), any())) + when(mockKeyManager.resolveClientForWrite(eq("1.2.3.4"), any())).thenReturn(client); + when(mockKeyManager.sortDatanodesForWrite(any(), any(), any(), any())) .thenAnswer(inv -> sortedOrder); when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); @@ -360,7 +467,8 @@ public void testAllocateBlockSortsSharedPipelineOnce() throws Exception { UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(), ozoneManager); // Sorted once for the shared pipeline... - verify(mockKeyManager, times(1)).sortDatanodesForWrite(any(), eq("1.2.3.4"), any()); + verify(mockKeyManager, times(1)).resolveClientForWrite(eq("1.2.3.4"), any()); + verify(mockKeyManager, times(1)).sortDatanodesForWrite(any(), eq("1.2.3.4"), eq(client), any()); // ...and the sorted order is applied to every block's pipeline. assertEquals(2, locations.size()); for (OmKeyLocationInfo location : locations) { @@ -370,21 +478,22 @@ 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. + // Pipelines sharing a datanode set must keep their own order when sorting is skipped. DatanodeDetails a = MockDatanodeDetails.randomDatanodeDetails(); DatanodeDetails b = MockDatanodeDetails.randomDatanodeDetails(); DatanodeDetails c = MockDatanodeDetails.randomDatanodeDetails(); List nodes1 = Arrays.asList(a, b, c); List nodes2 = Arrays.asList(c, b, a); + // Use a different nodesInOrder to catch a skipped sort overwriting it with getNodes(). + List inOrder1 = Arrays.asList(b, c, a); + List inOrder2 = Arrays.asList(a, c, b); Pipeline pipeline1 = Pipeline.newBuilder() .setState(Pipeline.PipelineState.OPEN) .setId(PipelineID.randomId()) .setReplicationConfig( StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE)) .setNodes(nodes1) + .setNodesInOrder(inOrder1) .build(); Pipeline pipeline2 = Pipeline.newBuilder() .setState(Pipeline.PipelineState.OPEN) @@ -392,6 +501,7 @@ public void testAllocateBlockKeepsPerPipelineOrderWhenSortSkipped() throws Excep .setReplicationConfig( StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE)) .setNodes(nodes2) + .setNodesInOrder(inOrder2) .build(); AllocatedBlock block1 = new AllocatedBlock.Builder().setPipeline(pipeline1) .setContainerBlockID(new ContainerBlockID(CONTAINER_ID, LOCAL_ID)).build(); @@ -403,9 +513,8 @@ public void testAllocateBlockKeepsPerPipelineOrderWhenSortSkipped() throws Excep KeyManager mockKeyManager = mock(KeyManager.class); when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); - // Skip the sort: return the input list instance unchanged. - when(mockKeyManager.sortDatanodesForWrite(any(), any(), any())) - .thenAnswer(inv -> inv.getArgument(0)); + when(mockKeyManager.resolveClientForWrite(anyString(), any())).thenReturn(mock(Node.class)); + when(mockKeyManager.sortDatanodesForWrite(any(), any(), any(), any())).thenReturn(null); when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); @@ -416,11 +525,8 @@ public void testAllocateBlockKeepsPerPipelineOrderWhenSortSkipped() throws Excep UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(), ozoneManager); assertEquals(2, locations.size()); - // Each pipeline keeps its own order; the skipped-sort result is not shared. - assertEquals(nodes1, locations.get(0).getPipeline().getNodesInOrder()); - assertEquals(nodes2, locations.get(1).getPipeline().getNodesInOrder()); - // Sorted per pipeline, since the unsorted result is not cached. - verify(mockKeyManager, times(2)).sortDatanodesForWrite(any(), eq("1.2.3.4"), any()); + assertEquals(inOrder1, locations.get(0).getPipeline().getNodesInOrder()); + assertEquals(inOrder2, locations.get(1).getPipeline().getNodesInOrder()); } @Test @@ -458,7 +564,8 @@ public void testAllocateBlockKeepsOrderWhenRemoteAddressEmpty() throws Exception UserInfo.newBuilder().setRemoteAddress("").build(), ozoneManager); assertEquals("", clientMachine.getValue()); - verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any()); + verify(mockKeyManager, never()).resolveClientForWrite(anyString(), any()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any(), any()); assertEquals(1, locations.size()); // Assert the write order (nodesInOrder), which copyWithNodesInOrder would // have changed had OM sorted; it must stay as the original pipeline order. @@ -472,7 +579,16 @@ public void sortDatanodesForWriteRequiresClientMachine() { MockDatanodeDetails.randomDatanodeDetails(), MockDatanodeDetails.randomDatanodeDetails()); assertThrows(IllegalArgumentException.class, - () -> keyManager.sortDatanodesForWrite(nodes, "", mock(NetworkTopology.class))); + () -> keyManager.sortDatanodesForWrite(nodes, "", mock(Node.class), mock(NetworkTopology.class))); + assertThrows(IllegalArgumentException.class, + () -> keyManager.resolveClientForWrite("", mock(NetworkTopology.class))); + } + + @Test + public void sortDatanodesForWriteRequiresClient() { + NullPointerException exception = assertThrows(NullPointerException.class, + () -> keyManager.sortDatanodesForWrite(Collections.emptyList(), "1.2.3.4", null, mock(NetworkTopology.class))); + assertEquals("client is null", exception.getMessage()); } // Like createAllocateBlockRequest, but sets sortDatanodes so preExecute