HDDS-7252. Polled source Datanodes are wrongly not re-considered for balancing in Container Balancer

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerSelectionCriteria.java

@@ -54,6 +54,7 @@ public class ContainerBalancerSelectionCriteria {
   private ContainerManager containerManager;
   private Set<ContainerID> selectedContainers;
   private Set<ContainerID> excludeContainers;
+  private Map<DatanodeDetails, Set<ContainerID>> excludeContainersDueToFailure;
   private FindSourceStrategy findSourceStrategy;
   private Map<DatanodeDetails, NavigableSet<ContainerID>> setMap;
 
@@ -68,6 +69,7 @@ public ContainerBalancerSelectionCriteria(
     this.replicationManager = replicationManager;
     this.containerManager = containerManager;
     selectedContainers = new HashSet<>();
+    excludeContainersDueToFailure = new HashMap<>();
     excludeContainers = balancerConfiguration.getExcludeContainers();
     this.findSourceStrategy = findSourceStrategy;
     this.setMap = new HashMap<>();
@@ -174,7 +176,12 @@ public boolean shouldBeExcluded(ContainerID containerID,
           "candidate container. Excluding it.", containerID);
       return true;
     }
+    Set<ContainerID> excludeContainersForSource = excludeContainersDueToFailure.get(node);
+    if (excludeContainersForSource == null) {
+      excludeContainersForSource = Collections.emptySet();
+    }
     return excludeContainers.contains(containerID) || selectedContainers.contains(containerID) ||
+        excludeContainersForSource.contains(containerID) ||
         !isContainerClosed(container, node) || isECContainerAndLegacyRMEnabled(container) ||
         isContainerReplicatingOrDeleting(containerID) ||
         !findSourceStrategy.canSizeLeaveSource(node, container.getUsedBytes())
@@ -242,6 +249,15 @@ public void setSelectedContainers(
     this.selectedContainers = selectedContainers;
   }
 
+  public void addToExcludeDueToFailContainers(DatanodeDetails node, ContainerID container) {
+    Set<ContainerID> excludeContainersForSource = excludeContainersDueToFailure.get(node);
+    if (excludeContainersForSource == null) {
+      excludeContainersForSource = new HashSet<>();
+    }
+    excludeContainersForSource.add(container);
+    excludeContainersDueToFailure.put(node, excludeContainersForSource);
+  }
+
 
   private NavigableSet<ContainerID> getCandidateContainers(DatanodeDetails node) {
     NavigableSet<ContainerID> newSet =
@@ -251,6 +267,10 @@ private NavigableSet<ContainerID> getCandidateContainers(DatanodeDetails node) {
       if (excludeContainers != null) {
         idSet.removeAll(excludeContainers);
       }
+      Set<ContainerID> excludeContainersForSource = excludeContainersDueToFailure.get(node);
+      if (excludeContainersForSource != null) {
+        idSet.removeAll(excludeContainersForSource);
+      }
       if (selectedContainers != null) {
         idSet.removeAll(selectedContainers);
       }

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java

@@ -553,6 +553,9 @@ private boolean processMoveSelection(DatanodeDetails source,
           containerID,
           containerToSourceMap.get(containerID),
           containerToTargetMap.get(containerID));
+      // add source back to queue as a different container can be selected in next run.
+      findSourceStrategy.addBackSourceDataNode(source);
+      selectionCriteria.addToExcludeDueToFailContainers(source, moveSelection.getContainerID());
       return false;
     }
 
@@ -563,6 +566,9 @@ private boolean processMoveSelection(DatanodeDetails source,
     } catch (ContainerNotFoundException e) {
       LOG.warn("Could not get container {} from Container Manager before " +
           "starting a container move", containerID, e);
+      // add source back to queue as a different container can be selected in next run.
+      findSourceStrategy.addBackSourceDataNode(source);
+      selectionCriteria.addToExcludeDueToFailContainers(source, moveSelection.getContainerID());
       return false;
     }
     LOG.info("ContainerBalancer is trying to move container {} with size " +
@@ -862,13 +868,22 @@ private boolean moveContainer(DatanodeDetails source,
     } catch (ContainerNotFoundException e) {
       LOG.warn("Could not find Container {} for container move",
           containerID, e);
+      // add source back to queue
+      findSourceStrategy.addBackSourceDataNode(source);
+      selectionCriteria.addToExcludeDueToFailContainers(source, moveSelection.getContainerID());
       metrics.incrementNumContainerMovesFailedInLatestIteration(1);
       return false;
-    } catch (NodeNotFoundException | TimeoutException |
-             ContainerReplicaNotFoundException e) {
+    } catch (NodeNotFoundException | TimeoutException e) {
       LOG.warn("Container move failed for container {}", containerID, e);
       metrics.incrementNumContainerMovesFailedInLatestIteration(1);
       return false;
+    } catch (ContainerReplicaNotFoundException e) {
+      LOG.warn("Container move failed for container {}", containerID, e);
+      metrics.incrementNumContainerMovesFailedInLatestIteration(1);
+      // add source back to queue for replica not found only
+      findSourceStrategy.addBackSourceDataNode(source);
+      selectionCriteria.addToExcludeDueToFailContainers(source, moveSelection.getContainerID());
+      return false;
     }
 
     /*
@@ -881,6 +896,13 @@ private boolean moveContainer(DatanodeDetails source,
       } else {
         MoveManager.MoveResult result = future.join();
         moveSelectionToFutureMap.put(moveSelection, future);
+        if (result == MoveManager.MoveResult.REPLICATION_FAIL_NOT_EXIST_IN_SOURCE ||
+            result == MoveManager.MoveResult.REPLICATION_FAIL_EXIST_IN_TARGET ||
+            result == MoveManager.MoveResult.REPLICATION_FAIL_CONTAINER_NOT_CLOSED ||
+            result == MoveManager.MoveResult.REPLICATION_FAIL_INFLIGHT_DELETION ||
+            result == MoveManager.MoveResult.REPLICATION_FAIL_INFLIGHT_REPLICATION) {
+          findSourceStrategy.addBackSourceDataNode(source);
+        }
         return result == MoveManager.MoveResult.COMPLETED;
       }
     } else {
@@ -1098,6 +1120,11 @@ Set<DatanodeDetails> getSelectedTargets() {
     return selectedTargets;
   }
 
+  @VisibleForTesting
+  Set<DatanodeDetails> getSelectedSources() {
+    return selectedSources;
+  }
+
   @VisibleForTesting
   int getCountDatanodesInvolvedPerIteration() {
     return countDatanodesInvolvedPerIteration;

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceGreedy.java

@@ -107,7 +107,7 @@ public void increaseSizeLeaving(DatanodeDetails dui, long size) {
     if (currentSize != null) {
       sizeLeavingNode.put(dui, currentSize + size);
       //reorder according to the latest sizeLeavingNode
-      potentialSources.add(nodeManager.getUsageInfo(dui));
+      addBackSourceDataNode(dui);
       return;
     }
     LOG.warn("Cannot find datanode {} in candidate source datanodes",
@@ -138,6 +138,12 @@ public void removeCandidateSourceDataNode(DatanodeDetails dui) {
     potentialSources.removeIf(a -> a.getDatanodeDetails().equals(dui));
   }
 
+  @Override
+  public void addBackSourceDataNode(DatanodeDetails dn) {
+    DatanodeUsageInfo dui = nodeManager.getUsageInfo(dn);
+    potentialSources.add(dui);
+  }
+
   /**
    * Checks if specified size can leave a specified target datanode
    * according to {@link ContainerBalancerConfiguration}

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceStrategy.java

@@ -45,6 +45,16 @@ public interface FindSourceStrategy {
    */
   void removeCandidateSourceDataNode(DatanodeDetails dui);
 
+  /**
+   * add the specified data node to the candidate source
+   * data nodes.
+   * This method does not check whether the specified Datanode is already present in the Collection.
+   * Callers must take the responsibility of checking and removing the Datanode before adding, if required.
+   *
+   * @param dn datanode to be added to potentialSources
+   */
+  void addBackSourceDataNode(DatanodeDetails dn);
+
   /**
    * increase the Leaving size of a candidate source data node.
    */

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/balancer/TestContainerBalancerTask.java

@@ -1114,6 +1114,43 @@ public void balancerShouldExcludeECContainersWhenLegacyRmIsEnabled()
     }
   }
 
+  /**
+   * Tests if balancer is adding the polled source datanode back to potentialSources queue
+   * if a move has failed due to a container related failure, like REPLICATION_FAIL_NOT_EXIST_IN_SOURCE.
+   */
+  @Test
+  public void testSourceDatanodeAddedBack()
+      throws NodeNotFoundException, IOException, IllegalContainerBalancerStateException,
+      InvalidContainerBalancerConfigurationException, TimeoutException, InterruptedException {
+
+    when(moveManager.move(any(ContainerID.class),
+        any(DatanodeDetails.class),
+        any(DatanodeDetails.class)))
+        .thenReturn(CompletableFuture.completedFuture(MoveManager.MoveResult.REPLICATION_FAIL_NOT_EXIST_IN_SOURCE))
+        .thenReturn(CompletableFuture.completedFuture(MoveManager.MoveResult.COMPLETED));
+    balancerConfiguration.setThreshold(10);
+    balancerConfiguration.setIterations(1);
+    balancerConfiguration.setMaxSizeEnteringTarget(10 * STORAGE_UNIT);
+    balancerConfiguration.setMaxSizeToMovePerIteration(100 * STORAGE_UNIT);
+    balancerConfiguration.setMaxDatanodesPercentageToInvolvePerIteration(100);
+    String includeNodes = nodesInCluster.get(0).getDatanodeDetails().getHostName() + "," +
+        nodesInCluster.get(nodesInCluster.size() - 1).getDatanodeDetails().getHostName();
+    balancerConfiguration.setIncludeNodes(includeNodes);
+
+    startBalancer(balancerConfiguration);
+    GenericTestUtils.waitFor(() -> ContainerBalancerTask.IterationResult.ITERATION_COMPLETED ==
+        containerBalancerTask.getIterationResult(), 10, 50);
+
+    assertEquals(2, containerBalancerTask.getCountDatanodesInvolvedPerIteration());
+    assertTrue(containerBalancerTask.getMetrics().getNumContainerMovesCompletedInLatestIteration() >= 1);
+    assertThat(containerBalancerTask.getMetrics().getNumContainerMovesFailed()).isEqualTo(1);
+    assertTrue(containerBalancerTask.getSelectedTargets().contains(nodesInCluster.get(0)
+        .getDatanodeDetails()));
+    assertTrue(containerBalancerTask.getSelectedSources().contains(nodesInCluster.get(nodesInCluster.size() - 1)
+        .getDatanodeDetails()));
+    stopBalancer();
+  }
+
   /**
    * Determines unBalanced nodes, that is, over and under utilized nodes,
    * according to the generated utilization values for nodes and the threshold.