HDFS-14303. check block directory logic not correct when there is only meta file, print no meaning warn log. Contributed by qiang Liu.

(cherry picked from commit a5d76a4965)

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

@@ -1374,7 +1374,7 @@ public class FsVolumeImpl implements FsVolumeSpi {
       if (!Block.isBlockFilename(file)) {
         if (isBlockMetaFile(Block.BLOCK_FILE_PREFIX, file.getName())) {
           long blockId = Block.getBlockId(file.getName());
-          verifyFileLocation(file.getParentFile(), bpFinalizedDir,
+          verifyFileLocation(file, bpFinalizedDir,
               blockId);
           report.add(new ScanInfo(blockId, null, file, this));
         }

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

@@ -26,6 +26,7 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
+import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
@@ -74,6 +75,9 @@ import org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.LazyPersistTestCase
 import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.test.GenericTestUtils;
 import org.apache.hadoop.util.Time;
+import org.apache.log4j.Level;
+import org.apache.log4j.SimpleLayout;
+import org.apache.log4j.WriterAppender;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;
@@ -396,6 +400,71 @@ public class TestDirectoryScanner {
     }
   }
 
+  /**
+   * test scan only meta file NOT generate wrong folder structure warn log.
+   */
+  @Test(timeout=600000)
+  public void testScanDirectoryStructureWarn() throws Exception {
+
+    //add a logger stream to check what has printed to log
+    ByteArrayOutputStream loggerStream = new ByteArrayOutputStream();
+    org.apache.log4j.Logger rootLogger =
+        org.apache.log4j.Logger.getRootLogger();
+    rootLogger.setLevel(Level.INFO);
+    WriterAppender writerAppender =
+        new WriterAppender(new SimpleLayout(), loggerStream);
+    rootLogger.addAppender(writerAppender);
+
+    cluster = new MiniDFSCluster
+        .Builder(CONF)
+        .storageTypes(new StorageType[] {
+            StorageType.RAM_DISK, StorageType.DEFAULT })
+        .numDataNodes(1)
+        .build();
+    try {
+      cluster.waitActive();
+      bpid = cluster.getNamesystem().getBlockPoolId();
+      DataNode dataNode = cluster.getDataNodes().get(0);
+      fds = DataNodeTestUtils.getFSDataset(cluster.getDataNodes().get(0));
+      client = cluster.getFileSystem().getClient();
+      scanner = new DirectoryScanner(dataNode, fds, CONF);
+      scanner.setRetainDiffs(true);
+      FsDatasetTestUtil.stopLazyWriter(cluster.getDataNodes().get(0));
+
+      // Create a file file on RAM_DISK
+      createFile(GenericTestUtils.getMethodName(), BLOCK_LENGTH, true);
+
+      // Ensure no difference between volumeMap and disk.
+      scan(1, 0, 0, 0, 0, 0);
+
+      //delete thre block file , left the meta file alone
+      deleteBlockFile();
+
+      //scan to ensure log warn not printed
+      scan(1, 1, 0, 1, 0, 0, 0);
+
+      //ensure the warn log not appear and missing block log do appear
+      String logContent = new String(loggerStream.toByteArray());
+      String missingBlockWarn = "Deleted a metadata file" +
+          " for the deleted block";
+      String dirStructureWarnLog = " found in invalid directory." +
+          "  Expected directory: ";
+      assertFalse("directory check print meaningless warning message",
+          logContent.contains(dirStructureWarnLog));
+      assertTrue("missing block warn log not appear",
+          logContent.contains(missingBlockWarn));
+      LOG.info("check pass");
+
+    } finally {
+      if (scanner != null) {
+        scanner.shutdown();
+        scanner = null;
+      }
+      cluster.shutdown();
+      cluster = null;
+    }
+  }
+
   @Test (timeout=300000)
   public void testDeleteBlockOnTransientStorage() throws Exception {
     cluster = new MiniDFSCluster