HDDS-16614. Remove pair usage from SCM replication core and tests.

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ECUnderReplicationHandler.java

@@ -31,7 +31,6 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
-import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.hdds.client.ECReplicationConfig;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.conf.StorageUnit;
@@ -144,15 +143,11 @@ public int processAndSendCommands(
       }
     }
 
-    Map<Integer, Pair<ContainerReplica, NodeStatus>> sources =
-        filterSources(replicas, deletionInFlight);
-    List<DatanodeDetails> availableSourceNodes =
-        sources.values().stream().map(Pair::getLeft)
-            .map(ContainerReplica::getDatanodeDetails)
-            .filter(datanodeDetails ->
-                datanodeDetails.getPersistedOpState() ==
-                    HddsProtos.NodeOperationalState.IN_SERVICE)
-            .collect(Collectors.toList());
+    Map<Integer, SourceReplica> sources = filterSources(replicas, deletionInFlight);
+    List<DatanodeDetails> availableSourceNodes = sources.values().stream().map(SourceReplica::replica)
+        .map(ContainerReplica::getDatanodeDetails).filter(datanodeDetails ->
+             datanodeDetails.getPersistedOpState() == HddsProtos.NodeOperationalState.IN_SERVICE)
+        .collect(Collectors.toList());
 
     try {
       IOException firstException = null;
@@ -229,7 +224,7 @@ public int processAndSendCommands(
     return commandsSent;
   }
 
-  private Map<Integer, Pair<ContainerReplica, NodeStatus>> filterSources(
+  private Map<Integer, SourceReplica> filterSources(
       Set<ContainerReplica> replicas, List<DatanodeDetails> deletionInFlight) {
     return replicas.stream().filter(r -> r
             .getState() == State.CLOSED)
@@ -239,25 +234,22 @@ private Map<Integer, Pair<ContainerReplica, NodeStatus>> filterSources(
         .filter(r -> !deletionInFlight.contains(r.getDatanodeDetails()))
         .map(r -> {
           try {
-            return Pair.of(r,
-                replicationManager.getNodeStatus(r.getDatanodeDetails()));
+            return new SourceReplica(r, replicationManager.getNodeStatus(r.getDatanodeDetails()));
           } catch (NodeNotFoundException e) {
             throw new IllegalStateException("Unable to find NodeStatus for "
                 + r.getDatanodeDetails(), e);
           }
         })
-        .filter(pair -> pair.getRight().isHealthy())
+        .filter(source -> source.nodeStatus().isHealthy())
         // If there are multiple nodes online for a given index, we just
         // pick any IN_SERVICE one. At the moment, the input streams cannot
         // handle multiple replicas for the same index, so if we passed them
         // all through they would not get used anyway.
         // If neither of the nodes are in service, we just pass one through,
         // as it will be decommission or maintenance.
-        .collect(Collectors.toMap(
-            pair -> pair.getLeft().getReplicaIndex(),
-            pair -> pair,
-            (p1, p2) -> {
-              if (p1.getRight().getOperationalState() == IN_SERVICE) {
+        .collect(Collectors.toMap(source -> source.replica().getReplicaIndex(),
+            source -> source, (p1, p2) -> {
+              if (p1.nodeStatus().getOperationalState() == IN_SERVICE) {
                 return p1;
               } else {
                 return p2;
@@ -272,7 +264,7 @@ private Map<Integer, Pair<ContainerReplica, NodeStatus>> filterSources(
    */
   private int processMissingIndexes(
       ECContainerReplicaCount replicaCount, Map<Integer,
-      Pair<ContainerReplica, NodeStatus>> sources,
+      SourceReplica> sources,
       List<DatanodeDetails> availableSourceNodes,
       List<DatanodeDetails> excludedNodes,
       List<DatanodeDetails> usedNodes) throws IOException {
@@ -299,7 +291,7 @@ private int processMissingIndexes(
   private int processReconstruction(
       ECContainerReplicaCount replicaCount,
       List<Integer> missingIndexes,
-      Map<Integer, Pair<ContainerReplica, NodeStatus>> sources,
+      Map<Integer, SourceReplica> sources,
       List<DatanodeDetails> availableSourceNodes,
       List<DatanodeDetails> excludedNodes,
       List<DatanodeDetails> usedNodes) throws IOException {
@@ -375,13 +367,11 @@ private int processReconstruction(
 
         // If we have more than the required number of data blocks, we can
         // prefer IN_SERVICE nodes to avoid overloading decommissioning nodes.
-        List<Pair<ContainerReplica, NodeStatus>> sortedSources =
+        List<SourceReplica> sortedSources =
             sources.values().stream()
                 .sorted((p1, p2) -> {
-                  boolean p1InService =
-                      p1.getRight().getOperationalState() == IN_SERVICE;
-                  boolean p2InService =
-                      p2.getRight().getOperationalState() == IN_SERVICE;
+                  boolean p1InService = p1.nodeStatus().getOperationalState() == IN_SERVICE;
+                  boolean p2InService = p2.nodeStatus().getOperationalState() == IN_SERVICE;
                   if (p1InService && !p2InService) {
                     return -1;
                   }
@@ -393,24 +383,24 @@ private int processReconstruction(
                 .collect(Collectors.toList());
 
         int inServiceCount = 0;
-        for (Pair<ContainerReplica, NodeStatus> src : sortedSources) {
-          if (src.getRight().getOperationalState() == IN_SERVICE) {
+        for (SourceReplica src : sortedSources) {
+          if (src.nodeStatus().getOperationalState() == IN_SERVICE) {
             inServiceCount++;
           }
         }
 
-        for (Pair<ContainerReplica, NodeStatus> src : sortedSources) {
+        for (SourceReplica src : sortedSources) {
           // If we have enough in-service nodes to fulfill the reconstruction
           // requirements, we skip any out-of-service nodes.
           if (inServiceCount >= repConfig.getData() &&
-              src.getRight().getOperationalState() != IN_SERVICE) {
+              src.nodeStatus().getOperationalState() != IN_SERVICE) {
             continue;
           }
           sourceDatanodesWithIndex.add(
               new ReconstructECContainersCommand
                   .DatanodeDetailsAndReplicaIndex(
-                  src.getLeft().getDatanodeDetails(),
-                  src.getLeft().getReplicaIndex()));
+                  src.replica().getDatanodeDetails(),
+                  src.replica().getReplicaIndex()));
         }
 
         final ReconstructECContainersCommand reconstructionCommand =
@@ -450,12 +440,9 @@ private int processReconstruction(
     return commandsSent;
   }
 
-  private static boolean hasSufficientSourcesForDecommissionReconstruction(
-      Map<Integer, Pair<ContainerReplica, NodeStatus>> sources,
-      Set<Integer> decomIndexes,
-      int requiredDataSources) {
-    long nonDecommissionSourceCount = sources.keySet().stream()
-        .filter(index -> !decomIndexes.contains(index))
+  private static boolean hasSufficientSourcesForDecommissionReconstruction(Map<Integer, SourceReplica> sources,
+      Set<Integer> decomIndexes, int requiredDataSources) {
+    long nonDecommissionSourceCount = sources.keySet().stream().filter(index -> !decomIndexes.contains(index))
         .count();
     return nonDecommissionSourceCount >= requiredDataSources;
   }
@@ -478,7 +465,7 @@ private List<DatanodeDetails> getTargetDatanodes(
    */
   private int processDecommissioningIndexes(
       ECContainerReplicaCount replicaCount,
-      Map<Integer, Pair<ContainerReplica, NodeStatus>> sources,
+      Map<Integer, SourceReplica> sources,
       List<DatanodeDetails> availableSourceNodes,
       List<DatanodeDetails> excludedNodes, List<DatanodeDetails> usedNodes)
       throws IOException {
@@ -489,12 +476,10 @@ private int processDecommissioningIndexes(
     }
 
     if (replicationManager.getConfig().isEcDecommissionReconstructionEnabled()) {
-      ECReplicationConfig repConfig =
-          (ECReplicationConfig) container.getReplicationConfig();
+      ECReplicationConfig repConfig = (ECReplicationConfig) container.getReplicationConfig();
       for (Integer index : decomIndexes) {
-        Pair<ContainerReplica, NodeStatus> source = sources.get(index);
-        if (source == null || !replicationManager.isNodeHighlyLoaded(
-            source.getLeft().getDatanodeDetails())) {
+        SourceReplica source = sources.get(index);
+        if (source == null || !replicationManager.isNodeHighlyLoaded(source.replica().getDatanodeDetails())) {
           continue;
         }
         if (!hasSufficientSourcesForDecommissionReconstruction(
@@ -513,9 +498,8 @@ private int processDecommissioningIndexes(
           metrics.incrEcReconstructionGlobalLimitReachedTotal();
           break;
         }
-        LOG.info("Source node {} is highly loaded, switching to "
-                + "reconstruction for decommissioning container {}",
-            source.getLeft().getDatanodeDetails(), container.containerID());
+        LOG.info("Source node {} is highly loaded, switching to " + "reconstruction for decommissioning container {}",
+            source.replica().getDatanodeDetails(), container.containerID());
         try {
           int commandsSent = processReconstruction(replicaCount,
               new ArrayList<>(decomIndexes), sources, availableSourceNodes,
@@ -567,19 +551,18 @@ private int processDecommissioningIndexes(
     int commandsSent = 0;
     CommandTargetOverloadedException overloadedException = null;
     for (Integer decomIndex : decomIndexes) {
-      Pair<ContainerReplica, NodeStatus> source = sources.get(decomIndex);
+      SourceReplica source = sources.get(decomIndex);
       if (source == null) {
         LOG.warn("Cannot find source replica for decommissioning index " +
                 "{} in container {}", decomIndex, container.containerID());
         continue;
       }
-      ContainerReplica sourceReplica = source.getLeft();
+      ContainerReplica sourceReplica = source.replica();
       if (!iterator.hasNext()) {
         LOG.warn("Couldn't find enough targets. Available source"
             + " nodes: {}, the target nodes: {}, excluded nodes: {},"
             + " usedNodes: {}, and the decommission indexes: {}",
-            sources.values().stream()
-                .map(Pair::getLeft).collect(Collectors.toSet()),
+            sources.values().stream().map(SourceReplica::replica).collect(Collectors.toSet()),
             selectedDatanodes, excludedNodes, usedNodes, decomIndexes);
         break;
       }
@@ -623,7 +606,7 @@ private int processDecommissioningIndexes(
    */
   private int processMaintenanceOnlyIndexes(
       ECContainerReplicaCount replicaCount,
-      Map<Integer, Pair<ContainerReplica, NodeStatus>> sources,
+      Map<Integer, SourceReplica> sources,
       List<DatanodeDetails> excludedNodes, List<DatanodeDetails> usedNodes)
       throws IOException {
     Set<Integer> maintIndexes = replicaCount.maintenanceOnlyIndexes(true);
@@ -657,20 +640,19 @@ private int processMaintenanceOnlyIndexes(
       if (additionalMaintenanceCopiesNeeded <= 0) {
         break;
       }
-      Pair<ContainerReplica, NodeStatus> source = sources.get(maintIndex);
+      SourceReplica source = sources.get(maintIndex);
       if (source == null) {
         LOG.warn("Cannot find source replica for maintenance index " +
             "{} in container {}", maintIndex, container.containerID());
         continue;
       }
-      ContainerReplica sourceReplica = source.getLeft();
+      ContainerReplica sourceReplica = source.replica();
       if (!iterator.hasNext()) {
         LOG.warn("Couldn't find enough targets. Available source"
                 + " nodes: {}, target nodes: {}, excluded nodes: {},"
                 + " usedNodes: {} and"
                 + " maintenance indexes: {}",
-            sources.values().stream()
-                .map(Pair::getLeft).collect(Collectors.toSet()),
+            sources.values().stream().map(SourceReplica::replica).collect(Collectors.toSet()),
             targets, excludedNodes, usedNodes, maintIndexes);
         break;
       }
@@ -830,4 +812,10 @@ private void checkAndRemoveUnhealthyReplica(
     }
   }
 
+  /**
+   * A container replica paired with the current status of its host node,
+   * used as a candidate source for EC reconstruction or replication.
+   */
+  private record SourceReplica(ContainerReplica replica, NodeStatus nodeStatus) { }
+
 }

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java

@@ -42,7 +42,6 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
-import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.hdds.ComponentVersion;
 import org.apache.hadoop.hdds.HddsConfigKeys;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
@@ -575,8 +574,7 @@ public void sendThrottledReplicationCommand(ContainerInfo containerInfo,
       List<DatanodeDetails> sources, DatanodeDetails target, int replicaIndex)
       throws CommandTargetOverloadedException, NotLeaderException {
     long containerID = containerInfo.getContainerID();
-    List<Pair<Integer, DatanodeDetails>> sourceWithCmds =
-        getAvailableDatanodesForReplication(sources);
+    List<DatanodeCommandCount> sourceWithCmds = getAvailableDatanodesForReplication(sources);
     if (sourceWithCmds.isEmpty()) {
       metrics.incrReplicateContainerCmdsDeferredTotal();
       throw new CommandTargetOverloadedException("No sources with capacity " +
@@ -618,8 +616,7 @@ public void sendThrottledReconstructionCommand(ContainerInfo containerInfo,
         () -> {
           // Pick the least-loaded target datanode that can accept the command.
           List<DatanodeDetails> targets = command.getTargetDatanodes();
-          List<Pair<Integer, DatanodeDetails>> targetWithCmds =
-              getAvailableDatanodesForReplication(targets);
+          List<DatanodeCommandCount> targetWithCmds = getAvailableDatanodesForReplication(targets);
           if (targetWithCmds.isEmpty()) {
             metrics.incrECReconstructionCmdsDeferredTotal();
             throw new CommandTargetOverloadedException("No target with capacity " +
@@ -662,14 +659,14 @@ private DatanodeInfo getDatanodeInfo(DatanodeDetails dnDetails) {
   }
 
   private DatanodeDetails selectAndOptionallyExcludeDatanode(
-      int additionalCmdCount, List<Pair<Integer, DatanodeDetails>> datanodes) {
+      int additionalCmdCount, List<DatanodeCommandCount> datanodes) {
     if (datanodes.isEmpty()) {
       return null;
     }
     // Put the least loaded datanode first
-    datanodes.sort(Comparator.comparingInt(Pair::getLeft));
-    DatanodeDetails datanode = datanodes.get(0).getRight();
-    int currentCount = datanodes.get(0).getLeft();
+    datanodes.sort(Comparator.comparingInt(DatanodeCommandCount::commandCount));
+    DatanodeDetails datanode = datanodes.get(0).datanode();
+    int currentCount = datanodes.get(0).commandCount();
     if (currentCount + additionalCmdCount >= getReplicationLimit(datanode)) {
       addExcludedNode(datanode);
     }
@@ -686,10 +683,8 @@ private DatanodeDetails selectAndOptionallyExcludeDatanode(
    * @return List of datanodes with the current command count that are not over
    *         the limit.
    */
-  private List<Pair<Integer, DatanodeDetails>>
-      getAvailableDatanodesForReplication(List<DatanodeDetails> datanodes) {
-    List<Pair<Integer, DatanodeDetails>> datanodeWithCommandCount
-        = new ArrayList<>();
+  private List<DatanodeCommandCount> getAvailableDatanodesForReplication(List<DatanodeDetails> datanodes) {
+    List<DatanodeCommandCount> datanodeWithCommandCount = new ArrayList<>();
     for (DatanodeDetails dn : datanodes) {
       try {
         int totalCount = getQueuedReplicationCount(dn);
@@ -701,7 +696,7 @@ private DatanodeDetails selectAndOptionallyExcludeDatanode(
           addExcludedNode(dn);
           continue;
         }
-        datanodeWithCommandCount.add(Pair.of(totalCount, dn));
+        datanodeWithCommandCount.add(new DatanodeCommandCount(totalCount, dn));
       } catch (NodeNotFoundException e) {
         LOG.error("Node {} not found in NodeManager. Should not happen",
             dn, e);
@@ -1893,5 +1888,11 @@ private void releaseSlot() {
   private interface ReconstructionSendAction {
     void run() throws CommandTargetOverloadedException, NotLeaderException;
   }
+
+  /**
+   * A datanode paired with its current queued replication command count,
+   * used to pick the least loaded datanode for a new command.
+   */
+  private record DatanodeCommandCount(int commandCount, DatanodeDetails datanode) { }
 }
 

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandlerTests.java

@@ -44,7 +44,6 @@
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
-import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -55,6 +54,7 @@
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.ContainerReplica;
 import org.apache.hadoop.hdds.scm.container.replication.ReplicationManager.ReplicationManagerConfiguration;
+import org.apache.hadoop.hdds.scm.container.replication.ReplicationTestUtil.DatanodeCommand;
 import org.apache.hadoop.hdds.scm.net.NodeSchema;
 import org.apache.hadoop.hdds.scm.net.NodeSchemaManager;
 import org.apache.hadoop.hdds.scm.node.NodeStatus;
@@ -74,7 +74,7 @@ public abstract class MisReplicationHandlerTests {
   private ContainerInfo container;
   private OzoneConfiguration conf;
   private ReplicationManager replicationManager;
-  private Set<Pair<DatanodeDetails, SCMCommand<?>>> commandsSent;
+  private Set<DatanodeCommand> commandsSent;
   private final AtomicBoolean throwThrottledException =
       new AtomicBoolean(false);
   private ReplicationManagerMetrics metrics;
@@ -229,14 +229,12 @@ protected void testMisReplication(Set<ContainerReplica> availableReplicas,
           pendingOp, result, maintenanceCnt);
     } finally {
       assertEquals(expectedNumberOfCommands, commandsSent.size());
-      for (Pair<DatanodeDetails, SCMCommand<?>> pair : commandsSent) {
-        SCMCommand<?> command = pair.getValue();
+      for (DatanodeCommand datanodeCommand : commandsSent) {
+        SCMCommand<?> command = datanodeCommand.command();
         assertSame(replicateContainerCommand, command.getType());
-        ReplicateContainerCommand replicateContainerCommand =
-            (ReplicateContainerCommand) command;
-        assertEquals(replicateContainerCommand.getContainerID(),
-            container.getContainerID());
-        DatanodeDetails replicateSrcDn = pair.getKey();
+        ReplicateContainerCommand replicateContainerCommand = (ReplicateContainerCommand) command;
+        assertEquals(replicateContainerCommand.getContainerID(), container.getContainerID());
+        DatanodeDetails replicateSrcDn = datanodeCommand.datanode();
         DatanodeDetails target = replicateContainerCommand.getTargetDatanode();
         assertThat(sourceDns).contains(replicateSrcDn);
         assertThat(targetNodes).contains(target);