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 a5d76a4965b686ec5532add511c64d5f87c88865)
(cherry picked from commit f33bcbb687e4dd4ab10d5dad89b1b8eae5788efb)

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

@@ -1363,7 +1363,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;
@@ -73,6 +74,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;
@@ -384,6 +388,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