HDDS-11386. Multithreading bug in ContainerBalancerTask (apache#7339)

aswinshakil · Nov 25, 2024 · 20c4cfa · 20c4cfa
1 parent b090312
commit 20c4cfa
Show file tree

Hide file tree

Showing 3 changed files with 41 additions and 43 deletions.
diff --git a/...src/main/java/org/apache/hadoop/hdds/scm/container/balancer/AbstractFindTargetGreedy.java b/...src/main/java/org/apache/hadoop/hdds/scm/container/balancer/AbstractFindTargetGreedy.java
@@ -32,11 +32,11 @@
 import org.slf4j.Logger;
 
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 
 /**
@@ -56,7 +56,7 @@ protected AbstractFindTargetGreedy(
       ContainerManager containerManager,
       PlacementPolicyValidateProxy placementPolicyValidateProxy,
       NodeManager nodeManager) {
-    sizeEnteringNode = new HashMap<>();
+    sizeEnteringNode = new ConcurrentHashMap<>();
     this.containerManager = containerManager;
     this.placementPolicyValidateProxy = placementPolicyValidateProxy;
     this.nodeManager = nodeManager;

diff --git a/...cm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java b/...cm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java
@@ -51,10 +51,11 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
+import java.util.Queue;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -117,7 +118,7 @@ public class ContainerBalancerTask implements Runnable {
   private IterationResult iterationResult;
   private int nextIterationIndex;
   private boolean delayStart;
-  private List<ContainerBalancerTaskIterationStatusInfo> iterationsStatistic;
+  private Queue<ContainerBalancerTaskIterationStatusInfo> iterationsStatistic;
 
   /**
    * Constructs ContainerBalancerTask with the specified arguments.
@@ -166,7 +167,7 @@ public ContainerBalancerTask(StorageContainerManager scm,
       findTargetStrategy = new FindTargetGreedyByUsageInfo(containerManager,
           placementPolicyValidateProxy, nodeManager);
     }
-    this.iterationsStatistic = new ArrayList<>();
+    this.iterationsStatistic = new ConcurrentLinkedQueue<>();
   }
 
   /**
@@ -308,43 +309,42 @@ private void balance() {
 
   private void saveIterationStatistic(Integer iterationNumber, IterationResult iR) {
     ContainerBalancerTaskIterationStatusInfo iterationStatistic = new ContainerBalancerTaskIterationStatusInfo(
-            iterationNumber,
-            iR.name(),
-            getSizeScheduledForMoveInLatestIteration() / OzoneConsts.GB,
-            metrics.getDataSizeMovedGBInLatestIteration(),
-            metrics.getNumContainerMovesScheduledInLatestIteration(),
-            metrics.getNumContainerMovesCompletedInLatestIteration(),
-            metrics.getNumContainerMovesFailedInLatestIteration(),
-            metrics.getNumContainerMovesTimeoutInLatestIteration(),
-            findTargetStrategy.getSizeEnteringNodes()
-                    .entrySet()
-                    .stream()
-                    .filter(Objects::nonNull)
-                    .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0)
-                    .collect(
-                            Collectors.toMap(
-                                    entry -> entry.getKey().getUuid(),
-                                    entry -> entry.getValue() / OzoneConsts.GB
-                            )
-                    ),
-            findSourceStrategy.getSizeLeavingNodes()
-                    .entrySet()
-                    .stream()
-                    .filter(Objects::nonNull)
-                    .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0)
-                    .collect(
-                            Collectors.toMap(
-                                    entry -> entry.getKey().getUuid(),
-                                    entry -> entry.getValue() / OzoneConsts.GB
-                            )
-                    )
+        iterationNumber,
+        iR.name(),
+        getSizeScheduledForMoveInLatestIteration() / OzoneConsts.GB,
+        metrics.getDataSizeMovedGBInLatestIteration(),
+        metrics.getNumContainerMovesScheduledInLatestIteration(),
+        metrics.getNumContainerMovesCompletedInLatestIteration(),
+        metrics.getNumContainerMovesFailedInLatestIteration(),
+        metrics.getNumContainerMovesTimeoutInLatestIteration(),
+        findTargetStrategy.getSizeEnteringNodes()
+            .entrySet()
+            .stream()
+            .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0)
+            .collect(
+                Collectors.toMap(
+                    entry -> entry.getKey().getUuid(),
+                    entry -> entry.getValue() / OzoneConsts.GB
+                )
+            ),
+        findSourceStrategy.getSizeLeavingNodes()
+            .entrySet()
+            .stream()
+            .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0)
+            .collect(
+                Collectors.toMap(
+                    entry -> entry.getKey().getUuid(),
+                    entry -> entry.getValue() / OzoneConsts.GB
+                )
+            )
     );
-    iterationsStatistic.add(iterationStatistic);
+    iterationsStatistic.offer(iterationStatistic);
   }
 
   public List<ContainerBalancerTaskIterationStatusInfo> getCurrentIterationsStatistic() {
+    List<ContainerBalancerTaskIterationStatusInfo> resultList = new ArrayList<>(iterationsStatistic);
 
-    int lastIterationNumber = iterationsStatistic.stream()
+    int lastIterationNumber = resultList.stream()
         .mapToInt(ContainerBalancerTaskIterationStatusInfo::getIterationNumber)
         .max()
         .orElse(0);
@@ -361,17 +361,16 @@ public List<ContainerBalancerTaskIterationStatusInfo> getCurrentIterationsStatis
         findTargetStrategy.getSizeEnteringNodes()
             .entrySet()
             .stream()
-            .filter(Objects::nonNull)
             .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0)
-            .collect(Collectors.toMap(
+            .collect(
+                Collectors.toMap(
                     entry -> entry.getKey().getUuid(),
                     entry -> entry.getValue() / OzoneConsts.GB
                 )
             ),
         findSourceStrategy.getSizeLeavingNodes()
             .entrySet()
             .stream()
-            .filter(Objects::nonNull)
             .filter(datanodeDetailsLongEntry -> datanodeDetailsLongEntry.getValue() > 0)
             .collect(
                 Collectors.toMap(
@@ -380,7 +379,6 @@ public List<ContainerBalancerTaskIterationStatusInfo> getCurrentIterationsStatis
                 )
             )
     );
-    List<ContainerBalancerTaskIterationStatusInfo> resultList = new ArrayList<>(iterationsStatistic);
     resultList.add(currentIterationStatistic);
     return resultList;
   }

diff --git a/...ver-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceGreedy.java b/...ver-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindSourceGreedy.java
@@ -26,11 +26,11 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.PriorityQueue;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * The selection criteria for selecting source datanodes , the containers of
@@ -46,7 +46,7 @@ public class FindSourceGreedy implements FindSourceStrategy {
   private Double lowerLimit;
 
   FindSourceGreedy(NodeManager nodeManager) {
-    sizeLeavingNode = new HashMap<>();
+    sizeLeavingNode = new ConcurrentHashMap<>();
     potentialSources = new PriorityQueue<>((a, b) -> {
       double currentUsageOfA = a.calculateUtilization(
           -sizeLeavingNode.get(a.getDatanodeDetails()));