HDFS-17812. Reset curEntry when BlockIteratorImpl checks a new subdir

Author: kokonguyen191

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

@@ -122,7 +122,7 @@ public long getGenerationStampAtblockIdSwitch() {
   }
 
   @VisibleForTesting
-  SequentialBlockIdGenerator getBlockIdGenerator() {
+  public SequentialBlockIdGenerator getBlockIdGenerator() {
     return blockIdGenerator;
   }
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java

@@ -891,6 +891,8 @@ public ExtendedBlock nextBlock() throws IOException {
             }
           }
           state.curFinalizedSubDir = getNextFinalizedSubDir();
+          // Reset cursor so it doesn't carry over from last iteration
+          state.curEntry = null;
           if (state.curFinalizedSubDir == null) {
             state.curFinalizedDir = getNextFinalizedDir();
             if (state.curFinalizedDir == null) {

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestBlockScanner.java

@@ -27,6 +27,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.fail;
 
 import java.io.Closeable;
@@ -48,6 +49,7 @@
 import org.apache.hadoop.hdfs.AppendTestUtil;
 import org.apache.hadoop.hdfs.MiniDFSNNTopology;
 import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
+import org.apache.hadoop.hdfs.server.blockmanagement.SequentialBlockIdGenerator;
 import org.apache.hadoop.hdfs.server.datanode.FsDatasetTestUtils.MaterializedReplica;
 import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
 import org.apache.hadoop.hdfs.server.datanode.VolumeScanner.ScanResultHandler;
@@ -1119,6 +1121,57 @@ public Boolean get() {
     }
   }
 
+  @Test
+  public void testNextBlock() throws Exception {
+    DataNodeFaultInjector oldInjector = DataNodeFaultInjector.get();
+    try {
+      Configuration conf = new Configuration();
+      disableBlockScanner(conf);
+
+      // Need to manually delete block to trigger error
+      final DataNodeFaultInjector injector = new DataNodeFaultInjector() {
+        @Override
+        public void delayDeleteReplica() {
+          while (true) {
+            try {
+              Thread.sleep(1000);
+            } catch (InterruptedException e) {
+              throw new RuntimeException(e);
+            }
+          }
+        }
+      };
+      DataNodeFaultInjector.set(injector);
+
+      TestContext ctx = new TestContext(conf, 1);
+      SequentialBlockIdGenerator blockIdGenerator =
+          ctx.cluster.getNamesystem().getBlockManager().getBlockIdManager().getBlockIdGenerator();
+
+      // /subdir24/subdir4
+      blockIdGenerator.skipTo(1356375042);
+      ctx.createFiles(0, 1, 1);
+      // /subdir24/subdir29
+      blockIdGenerator.skipTo(1356381682);
+      ctx.createFiles(0, 1, 1);
+
+      FsVolumeSpi volume = ctx.volumes.get(0);
+      BlockIterator iter = volume.newBlockIterator(ctx.bpids[0], "test");
+
+      // Get the subdir29 block. It comes before the subdir4 block.
+      ExtendedBlock nextBlock = iter.nextBlock();
+      assertNotNull(nextBlock);
+      FinalizedReplica replica = (FinalizedReplica) ctx.datanode.getFSDataset()
+          .getReplica(ctx.cluster.getNamesystem().getBlockPoolId(), nextBlock.getBlockId());
+      replica.deleteBlockData();
+      replica.deleteMetadata();
+
+      // subdir29 has been cleaned up, nextBlock() will skip directly to subdir4
+      assertNotNull(iter.nextBlock());
+    } finally {
+      DataNodeFaultInjector.set(oldInjector);
+    }
+  }
+
   private static class DelayVolumeScannerResponseToInterrupt extends
       VolumeScannerCBInjector {
     final private long delayAmountNS;