HDFS-17050. Erasure coding: fix bug for invalidating duplicated block when two ec block at the same datanode but different storage.

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/JvmPauseMonitor.java

@@ -195,11 +195,11 @@ public void run() {
 
         if (extraSleepTime > warnThresholdMs) {
           ++numGcWarnThresholdExceeded;
-          LOG.warn(formatMessage(
+          LOG.debug(formatMessage(
               extraSleepTime, gcTimesAfterSleep, gcTimesBeforeSleep));
         } else if (extraSleepTime > infoThresholdMs) {
           ++numGcInfoThresholdExceeded;
-          LOG.info(formatMessage(
+          LOG.debug(formatMessage(
               extraSleepTime, gcTimesAfterSleep, gcTimesBeforeSleep));
         }
         totalGcExtraSleepTime += extraSleepTime;

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockInfo.java

@@ -259,7 +259,7 @@ DatanodeStorageInfo findStorageInfo(DatanodeDescriptor dn) {
    */
   int findStorageInfo(DatanodeStorageInfo storageInfo) {
     int len = getCapacity();
-    for(int idx = 0; idx < len; idx++) {
+    for (int idx = 0; idx < len; idx++) {
       DatanodeStorageInfo cur = getStorageInfo(idx);
       if (cur == storageInfo) {
         return idx;

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockInfoStriped.java

@@ -6,9 +6,9 @@
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -128,8 +128,25 @@ boolean addStorage(DatanodeStorageInfo storage, Block reportedBlock) {
     DatanodeStorageInfo old = getStorageInfo(index);
     if (old != null && !old.equals(storage)) { // over replicated
       // check if the storage has been stored
+      boolean blockIdNotEquals = false;
+      long blockGroupId = BlockIdManager.convertToStripedID(reportedBlock.getBlockId() - blockIndex);
+      Iterator<BlockInfo> blockIterator = old.getBlockIterator();
+      while (blockIterator.hasNext()) {
+        BlockInfo blockInfo = blockIterator.next();
+        if (!blockInfo.isStriped()) {
+          continue;
+        } else {
+          if (BlockIdManager.convertToStripedID(blockInfo.getBlockId()) == blockGroupId) {
+            Block blockOnOldStorage = ((BlockInfoStriped) blockInfo).getBlockOnStorage(old);
+            if (blockOnOldStorage.getBlockId() != reportedBlock.getBlockId()) {
+              blockIdNotEquals = true;
+              break;
+            }
+          }
+        }
+      }
       int i = findStorageInfo(storage);
-      if (i == -1) {
+      if (i == -1 || blockIdNotEquals) {
         index = findSlot();
       } else {
         return true;

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java

@@ -2550,16 +2550,18 @@ DatanodeDescriptor[] chooseSourceDatanodes(BlockInfo block,
 
     BitSet liveBitSet = null;
     BitSet decommissioningBitSet = null;
+    HashSet<DatanodeDescriptor> alreadyCorruptedSet = null;
     if (isStriped) {
       int blockNum = ((BlockInfoStriped) block).getTotalBlockNum();
       liveBitSet = new BitSet(blockNum);
       decommissioningBitSet = new BitSet(blockNum);
+      alreadyCorruptedSet = new HashSet<>();
     }
 
     for (DatanodeStorageInfo storage : blocksMap.getStorages(block)) {
       final DatanodeDescriptor node = getDatanodeDescriptorFromStorage(storage);
       final StoredReplicaState state = checkReplicaOnStorage(numReplicas, block,
-          storage, corruptReplicas.getNodes(block), false);
+          storage, corruptReplicas.getNodes(block), false, alreadyCorruptedSet);
       if (state == StoredReplicaState.LIVE) {
         if (storage.getStorageType() == StorageType.PROVIDED) {
           storage = new DatanodeStorageInfo(node, storage.getStorageID(),
@@ -4543,25 +4545,32 @@ public NumberReplicas countNodes(BlockInfo b) {
   NumberReplicas countNodes(BlockInfo b, boolean inStartupSafeMode) {
     NumberReplicas numberReplicas = new NumberReplicas();
     Collection<DatanodeDescriptor> nodesCorrupt = corruptReplicas.getNodes(b);
+    HashSet<DatanodeDescriptor> alreadyCorruptSet = null;
     if (b.isStriped()) {
+      alreadyCorruptSet = new HashSet<>();
       countReplicasForStripedBlock(numberReplicas, (BlockInfoStriped) b,
-          nodesCorrupt, inStartupSafeMode);
+          nodesCorrupt, inStartupSafeMode, alreadyCorruptSet);
     } else {
       for (DatanodeStorageInfo storage : blocksMap.getStorages(b)) {
         checkReplicaOnStorage(numberReplicas, b, storage, nodesCorrupt,
-            inStartupSafeMode);
+            inStartupSafeMode, alreadyCorruptSet);
       }
     }
     return numberReplicas;
   }
 
   private StoredReplicaState checkReplicaOnStorage(NumberReplicas counters,
       BlockInfo b, DatanodeStorageInfo storage,
-      Collection<DatanodeDescriptor> nodesCorrupt, boolean inStartupSafeMode) {
+      Collection<DatanodeDescriptor> nodesCorrupt, boolean inStartupSafeMode,
+      HashSet<DatanodeDescriptor> alreadyCorrupt) {
     final StoredReplicaState s;
     if (storage.getState() == State.NORMAL) {
       final DatanodeDescriptor node = storage.getDatanodeDescriptor();
-      if (nodesCorrupt != null && nodesCorrupt.contains(node)) {
+      if (nodesCorrupt != null && nodesCorrupt.contains(node) &&
+          (alreadyCorrupt == null || !alreadyCorrupt.contains(node))) {
+        if (alreadyCorrupt != null) {
+          alreadyCorrupt.add(node);
+        }
         s = StoredReplicaState.CORRUPT;
       } else if (inStartupSafeMode) {
         s = StoredReplicaState.LIVE;
@@ -4607,12 +4616,12 @@ private StoredReplicaState checkReplicaOnStorage(NumberReplicas counters,
    */
   private void countReplicasForStripedBlock(NumberReplicas counters,
       BlockInfoStriped block, Collection<DatanodeDescriptor> nodesCorrupt,
-      boolean inStartupSafeMode) {
+      boolean inStartupSafeMode, HashSet<DatanodeDescriptor> alreadyCorrupt) {
     BitSet liveBitSet = new BitSet(block.getTotalBlockNum());
     BitSet decommissioningBitSet = new BitSet(block.getTotalBlockNum());
     for (StorageAndBlockIndex si : block.getStorageAndIndexInfos()) {
       StoredReplicaState state = checkReplicaOnStorage(counters, block,
-          si.getStorage(), nodesCorrupt, inStartupSafeMode);
+          si.getStorage(), nodesCorrupt, inStartupSafeMode, alreadyCorrupt);
       countLiveAndDecommissioningReplicas(counters, state, liveBitSet,
           decommissioningBitSet, si.getBlockIndex());
     }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeStorageInfo.java

@@ -262,9 +262,19 @@ public AddBlockResult addBlock(BlockInfo b, Block reportedBlock) {
 
     if (otherStorage != null) {
       if (otherStorage != this) {
-        // The block belongs to a different storage. Remove it first.
-        otherStorage.removeBlock(b);
-        result = AddBlockResult.REPLACED;
+        if (!b.isStriped()) {
+          // The block belongs to a different storage. Remove it first.
+          otherStorage.removeBlock(b);
+          result = AddBlockResult.REPLACED;
+        } else {
+          long reportBlockId = reportedBlock.getBlockId();
+          Block blockOnOtherStorage = ((BlockInfoStriped) b).getBlockOnStorage(otherStorage);
+          if (reportBlockId == blockOnOtherStorage.getBlockId()) {
+            // The block belongs to a different storage. Remove it first.
+            otherStorage.removeBlock(b);
+            result = AddBlockResult.REPLACED;
+          }
+        }
       } else {
         // The block is already associated with this storage.
         return AddBlockResult.ALREADY_EXIST;