HDDS-15941. Resolve client before block loop and return null from sortDatanodesForWrite

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;
 

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java

@@ -22,20 +22,23 @@
 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;
 import java.io.File;
 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<DatanodeDetails> datanodes = new ArrayList<>(NODE_COUNT);
     List<String> nodeMapping = new ArrayList<>(NODE_COUNT);
@@ -189,43 +192,123 @@ 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<? extends DatanodeDetails> 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<? extends DatanodeDetails> 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);
     }
   }
 
   @Test
-  public void sortDatanodesForWriteKeepsOrderForStaleTopology() {
+  public void sortDatanodesForWriteReturnsNullForStaleTopology() {
     List<DatanodeDetails> nodes = new ArrayList<>();
     nodes.add(randomDatanodeDetails());
     nodes.addAll(nodeManager.getAllNodes());
+    Node client = keyManager.resolveClientForWrite("edge0", om.getClusterMap());
+    assertNotNull(client);
 
-    List<? extends DatanodeDetails> 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<? extends DatanodeDetails> nodes = nodeManager.getAllNodes();
-    List<DatanodeDetails> 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<? extends DatanodeDetails> 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<String, String> 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<? extends DatanodeDetails> all = nodeManager.getAllNodes();
+    DatanodeDetails clientDn = all.get(0);
+    DatanodeDetails sameRack = null;
+    List<DatanodeDetails> 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<DatanodeDetails> without = new ArrayList<>();
+      without.add(rpcCopy(sameRack));
+      without.add(rpcCopy(otherRack.get(0)));
+      without.add(rpcCopy(otherRack.get(1)));
+      List<? extends DatanodeDetails> 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<DatanodeDetails> with = new ArrayList<>();
+      with.add(rpcCopy(otherRack.get(0)));
+      with.add(rpcCopy(sameRack));
+      with.add(rpcCopy(clientDn));
+      List<? extends DatanodeDetails> sortedWith =
+          keyManager.sortDatanodesForWrite(with, address, otherRackClient, om.getClusterMap());
+      assertNotNull(sortedWith);
+      assertEquals(clientDn, sortedWith.get(0));
+    }
+  }
+
+  @Test
+  public void resolveClientForWriteIsTheOnlyMappingLookup() {
+    List<? extends DatanodeDetails> 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<DatanodeDetails> 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<String> resolve(List<String> names) {
+      RESOLVED_NAMES.addAndGet(names.size());
+      return super.resolve(names);
+    }
+  }
 }

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<? extends DatanodeDetails> sortDatanodesForWrite(
-      List<? extends DatanodeDetails> nodes, String clientMachine, NetworkTopology clusterMap);
+      List<? extends DatanodeDetails> nodes, String clientMachine, Node client, NetworkTopology clusterMap);
 
   /**
    * @return true if OM should sort the streaming-write pipeline locally

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java

@@ -2283,11 +2283,6 @@ private void sortDatanodes(String clientMachine, List<OmKeyInfo> keyInfos) {
             List<? extends DatanodeDetails> 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<? extends DatanodeDetails> sortDatanodes(List<? extends DatanodeDeta
     return clusterMap.sortByDistanceCost(client, nodes, nodes.size());
   }
 
+  @Override
+  public Node resolveClientForWrite(String clientMachine, NetworkTopology clusterMap) {
+    Preconditions.checkArgument(!StringUtils.isEmpty(clientMachine), "clientMachine is empty");
+    Objects.requireNonNull(clusterMap, "clusterMap is null");
+    return captureLatencyNs(metrics.getAllocateBlockSortDatanodesLatencyNs(),
+        () -> getOtherNode(clientMachine, clusterMap));
+  }
+
   @Override
   public List<? extends DatanodeDetails> sortDatanodesForWrite(
-      List<? extends DatanodeDetails> nodes, String clientMachine, NetworkTopology clusterMap) {
-    Preconditions.checkArgument(!StringUtils.isEmpty(clientMachine),
-        "clientMachine is empty");
+      List<? extends DatanodeDetails> 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<? extends DatanodeDetails> sortByClusterMapDistance(
       NetworkTopology clusterMap, Node client,
@@ -2351,7 +2353,7 @@ private List<? extends DatanodeDetails> 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<? extends DatanodeDetails> sortByClusterMapDistance(
 
   private Node getClientNode(String clientMachine,
       List<? extends DatanodeDetails> nodes, NetworkTopology clusterMap) {
+    final Node pipelineClient = findClientInPipeline(clientMachine, nodes, clusterMap);
+    return pipelineClient != null ? pipelineClient : getOtherNode(clientMachine, clusterMap);
+  }
+
+  private Node findClientInPipeline(String clientMachine,
+      List<? extends DatanodeDetails> 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,

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",