diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
index 028521ef79a..d9b96aba036 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
@@ -396,6 +396,9 @@ Release 2.7.0 - UNRELEASED
     HDFS-7439. Add BlockOpResponseProto's message to the exception messages.
     (Takanobu Asanuma via szetszwo)
 
+    HDFS-7789. DFSck should resolve the path to support cross-FS symlinks.
+    (gera)
+
   OPTIMIZATIONS
 
     HDFS-7454. Reduce memory footprint for AclEntries in NameNode.
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSck.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSck.java
index 5b432909be4..900d8ba293e 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSck.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSck.java
@@ -223,6 +223,14 @@ public class DFSck extends Configured implements Tool {
     return errCode;
   }
   
+
+  private Path getResolvedPath(String dir) throws IOException {
+    Configuration conf = getConf();
+    Path dirPath = new Path(dir);
+    FileSystem fs = dirPath.getFileSystem(conf);
+    return fs.resolvePath(dirPath);
+  }
+
   /**
    * Derive the namenode http address from the current file system,
    * either default or as set by "-fs" in the generic options.
@@ -234,19 +242,12 @@ public class DFSck extends Configured implements Tool {
     Configuration conf = getConf();
 
     //get the filesystem object to verify it is an HDFS system
-    final FileSystem fs;
-    try {
-      fs = target.getFileSystem(conf);
-    } catch (IOException ioe) {
-      System.err.println("FileSystem is inaccessible due to:\n"
-          + StringUtils.stringifyException(ioe));
-      return null;
-    }
+    final FileSystem fs = target.getFileSystem(conf);
     if (!(fs instanceof DistributedFileSystem)) {
       System.err.println("FileSystem is " + fs.getUri());
       return null;
     }
-    
+
     return DFSUtil.getInfoServer(HAUtil.getAddressOfActive(fs), conf,
         DFSUtil.getHttpClientScheme(conf));
   }
@@ -300,8 +301,16 @@ public class DFSck extends Configured implements Tool {
       dir = "/";
     }
 
-    final Path dirpath = new Path(dir);
-    final URI namenodeAddress = getCurrentNamenodeAddress(dirpath);
+    Path dirpath = null;
+    URI namenodeAddress = null;
+    try {
+      dirpath = getResolvedPath(dir);
+      namenodeAddress = getCurrentNamenodeAddress(dirpath);
+    } catch (IOException ioe) {
+      System.err.println("FileSystem is inaccessible due to:\n"
+          + StringUtils.stringifyException(ioe));
+    }
+
     if (namenodeAddress == null) {
       //Error message already output in {@link #getCurrentNamenodeAddress()}
       System.err.println("DFSck exiting.");
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFsck.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFsck.java
index 3c69f1d031b..b8024a75730 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFsck.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFsck.java
@@ -211,10 +211,16 @@ public class TestFsck {
     try {
       // Audit log should contain one getfileinfo and one fsck
       reader = new BufferedReader(new FileReader(auditLogFile));
-      String line = reader.readLine();
-      assertNotNull(line);
-      assertTrue("Expected getfileinfo event not found in audit log",
-          getfileinfoPattern.matcher(line).matches());
+      String line;
+
+      // one extra getfileinfo stems from resolving the path
+      //
+      for (int i = 0; i < 2; i++) {
+        line = reader.readLine();
+        assertNotNull(line);
+        assertTrue("Expected getfileinfo event not found in audit log",
+            getfileinfoPattern.matcher(line).matches());
+      }
       line = reader.readLine();
       assertNotNull(line);
       assertTrue("Expected fsck event not found in audit log", fsckPattern
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFsckWithMultipleNameNodes.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFsckWithMultipleNameNodes.java
index f4cb6249950..124b301cb98 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFsckWithMultipleNameNodes.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFsckWithMultipleNameNodes.java
@@ -18,6 +18,7 @@
 package org.apache.hadoop.hdfs.server.namenode;
 
 import java.io.IOException;
+import java.net.URI;
 import java.util.Random;
 import java.util.concurrent.TimeoutException;
 
@@ -26,6 +27,8 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.viewfs.ConfigUtil;
+import org.apache.hadoop.fs.viewfs.ViewFileSystem;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.DFSTestUtil;
 import org.apache.hadoop.hdfs.HdfsConfiguration;
@@ -119,6 +122,23 @@ public class TestFsckWithMultipleNameNodes {
         LOG.info("result=" + result);
         Assert.assertTrue(result.contains("Status: HEALTHY"));
       }
+
+      // Test viewfs
+      //
+      LOG.info("RUN_TEST 3");
+      final String[] vurls = new String[nNameNodes];
+      for (int i = 0; i < vurls.length; i++) {
+        String link = "/mount/nn_" + i + FILE_NAME;
+        ConfigUtil.addLink(conf, link, new URI(urls[i]));
+        vurls[i] = "viewfs:" + link;
+      }
+
+      for(int i = 0; i < vurls.length; i++) {
+        LOG.info("vurls[" + i + "]=" + vurls[i]);
+        final String result = TestFsck.runFsck(conf, 0, false, vurls[i]);
+        LOG.info("result=" + result);
+        Assert.assertTrue(result.contains("Status: HEALTHY"));
+      }
     } finally {
       cluster.shutdown();
     }