diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
index 499f5a764a0..350abab352b 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
@@ -2400,6 +2400,51 @@ public class DistributedFileSystem extends FileSystem
     }.resolve(this, absF);
   }
 
+  /**
+   * Get the difference between two snapshots of a directory iteratively.
+   *
+   * @param snapshotDir full path of the directory where snapshots are taken.
+   * @param fromSnapshotName snapshot name of the from point. Null indicates the current tree.
+   * @param toSnapshotName snapshot name of the to point. Null indicates the current tree.
+   * @param snapshotDiffStartPath path relative to the snapshottable root directory from where
+   *     the snapshotdiff computation needs to start.
+   * @param snapshotDiffIndex index in the created or deleted list of the directory at which the
+   *     snapshotdiff computation stopped during the last rpc call. -1 indicates the diff
+   *     computation needs to start right from the start path.
+   * @return the difference report represented as a {@link SnapshotDiffReportListing}.
+   * @throws IOException if an I/O error occurred.
+   */
+  public SnapshotDiffReportListing getSnapshotDiffReportListing(Path snapshotDir,
+      String fromSnapshotName, String toSnapshotName, String snapshotDiffStartPath,
+      int snapshotDiffIndex) throws IOException {
+    statistics.incrementReadOps(1);
+    storageStatistics.incrementOpCounter(OpType.GET_SNAPSHOT_DIFF);
+    Path absF = fixRelativePart(snapshotDir);
+    return new FileSystemLinkResolver<SnapshotDiffReportListing>() {
+
+      @Override
+      public SnapshotDiffReportListing doCall(final Path p) throws IOException {
+        return dfs.getSnapshotDiffReportListing(getPathName(p), fromSnapshotName, toSnapshotName,
+            DFSUtilClient.string2Bytes(snapshotDiffStartPath), snapshotDiffIndex);
+      }
+
+      @Override
+      public SnapshotDiffReportListing next(final FileSystem fs, final Path p)
+          throws IOException {
+        if (fs instanceof DistributedFileSystem) {
+          DistributedFileSystem distributedFileSystem = (DistributedFileSystem)fs;
+          distributedFileSystem.getSnapshotDiffReportListing(p, fromSnapshotName, toSnapshotName,
+              snapshotDiffStartPath, snapshotDiffIndex);
+        } else {
+          throw new UnsupportedOperationException("Cannot perform snapshot"
+              + " operations on a symlink to a non-DistributedFileSystem: "
+              + snapshotDir + " -> " + p);
+        }
+        return null;
+      }
+    }.resolve(this, absF);
+  }
+
   /**
    * Get the close status of a file
    * @param src The path to the file
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/web/WebHdfsFileSystem.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/web/WebHdfsFileSystem.java
index 2cba43b32dd..ee5bf14c83c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/web/WebHdfsFileSystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/web/WebHdfsFileSystem.java
@@ -1458,7 +1458,9 @@ public class WebHdfsFileSystem extends FileSystem
         }.run();
   }
 
-  private SnapshotDiffReportListing getSnapshotDiffReportListing(
+  // This API should be treated as private to WebHdfsFileSystem. Only tests can use it directly.
+  @VisibleForTesting
+  public SnapshotDiffReportListing getSnapshotDiffReportListing(
         String snapshotDir, final String fromSnapshot, final String toSnapshot,
         byte[] startPath, int index) throws IOException {
     return new FsPathResponseRunner<SnapshotDiffReportListing>(
diff --git a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/client/HttpFSFileSystem.java b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/client/HttpFSFileSystem.java
index 9f7dfd4647b..0b4de728444 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/client/HttpFSFileSystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/client/HttpFSFileSystem.java
@@ -53,6 +53,7 @@ import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy;
 import org.apache.hadoop.hdfs.protocol.FsPermissionExtension;
 import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
 import org.apache.hadoop.hdfs.protocol.SnapshotDiffReport;
+import org.apache.hadoop.hdfs.protocol.SnapshotDiffReportListing;
 import org.apache.hadoop.hdfs.protocol.SnapshottableDirectoryStatus;
 import org.apache.hadoop.hdfs.protocol.SnapshotStatus;
 import org.apache.hadoop.hdfs.web.JsonUtilClient;
@@ -136,6 +137,8 @@ public class HttpFSFileSystem extends FileSystem
   public static final String POLICY_NAME_PARAM = "storagepolicy";
   public static final String SNAPSHOT_NAME_PARAM = "snapshotname";
   public static final String OLD_SNAPSHOT_NAME_PARAM = "oldsnapshotname";
+  public static final String SNAPSHOT_DIFF_START_PATH = "snapshotdiffstartpath";
+  public static final String SNAPSHOT_DIFF_INDEX = "snapshotdiffindex";
   public static final String FSACTION_MODE_PARAM = "fsaction";
   public static final String EC_POLICY_NAME_PARAM = "ecpolicy";
 
@@ -266,8 +269,8 @@ public class HttpFSFileSystem extends FileSystem
     RENAMESNAPSHOT(HTTP_PUT), GETSNAPSHOTDIFF(HTTP_GET),
     GETSNAPSHOTTABLEDIRECTORYLIST(HTTP_GET), GETSNAPSHOTLIST(HTTP_GET),
     GETSERVERDEFAULTS(HTTP_GET),
-    CHECKACCESS(HTTP_GET), SETECPOLICY(HTTP_PUT), GETECPOLICY(
-        HTTP_GET), UNSETECPOLICY(HTTP_POST), SATISFYSTORAGEPOLICY(HTTP_PUT);
+    CHECKACCESS(HTTP_GET), SETECPOLICY(HTTP_PUT), GETECPOLICY(HTTP_GET), UNSETECPOLICY(
+        HTTP_POST), SATISFYSTORAGEPOLICY(HTTP_PUT), GETSNAPSHOTDIFFLISTING(HTTP_GET);
 
     private String httpMethod;
 
@@ -1577,6 +1580,22 @@ public class HttpFSFileSystem extends FileSystem
     return JsonUtilClient.toSnapshotDiffReport(json);
   }
 
+  public SnapshotDiffReportListing getSnapshotDiffReportListing(Path path, String snapshotOldName,
+      String snapshotNewName, byte[] snapshotDiffStartPath, Integer snapshotDiffIndex)
+      throws IOException {
+    Map<String, String> params = new HashMap<>();
+    params.put(OP_PARAM, Operation.GETSNAPSHOTDIFFLISTING.toString());
+    params.put(SNAPSHOT_NAME_PARAM, snapshotNewName);
+    params.put(OLD_SNAPSHOT_NAME_PARAM, snapshotOldName);
+    params.put(SNAPSHOT_DIFF_START_PATH, DFSUtilClient.bytes2String(snapshotDiffStartPath));
+    params.put(SNAPSHOT_DIFF_INDEX, snapshotDiffIndex.toString());
+    HttpURLConnection conn = getConnection(
+        Operation.GETSNAPSHOTDIFFLISTING.getMethod(), params, path, true);
+    HttpExceptionUtils.validateResponse(conn, HttpURLConnection.HTTP_OK);
+    JSONObject json = (JSONObject) HttpFSUtils.jsonParse(conn);
+    return JsonUtilClient.toSnapshotDiffReportListing(json);
+  }
+
   public SnapshottableDirectoryStatus[] getSnapshottableDirectoryList()
       throws IOException {
     Map<String, String> params = new HashMap<String, String>();
diff --git a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/FSOperations.java b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/FSOperations.java
index e272cdc71b6..33f2abbb319 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/FSOperations.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/FSOperations.java
@@ -45,6 +45,7 @@ import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy;
 import org.apache.hadoop.hdfs.protocol.HdfsConstants;
 import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
 import org.apache.hadoop.hdfs.protocol.SnapshotDiffReport;
+import org.apache.hadoop.hdfs.protocol.SnapshotDiffReportListing;
 import org.apache.hadoop.hdfs.protocol.SnapshottableDirectoryStatus;
 import org.apache.hadoop.hdfs.protocol.SnapshotStatus;
 import org.apache.hadoop.hdfs.web.JsonUtil;
@@ -1879,6 +1880,65 @@ public final class FSOperations {
     }
   }
 
+  /**
+   *  Executor that performs a getSnapshotDiffListing operation.
+   */
+  @InterfaceAudience.Private
+  public static class FSGetSnapshotDiffListing implements
+      FileSystemAccess.FileSystemExecutor<String> {
+
+    private final Path path;
+    private final String oldSnapshotName;
+    private final String snapshotName;
+    private final String snapshotDiffStartPath;
+    private final int snapshotDiffIndex;
+
+    /**
+     * Creates a getSnapshotDiffListing executor.
+     *
+     * @param path directory path of the snapshots to be examined.
+     * @param oldSnapshotName Older snapshot name.
+     * @param snapshotName Newer snapshot name.
+     * @param snapshotDiffStartPath snapshot diff start path.
+     * @param snapshotDiffIndex snapshot diff index.
+     */
+    public FSGetSnapshotDiffListing(String path, String oldSnapshotName,
+        String snapshotName, String snapshotDiffStartPath, int snapshotDiffIndex) {
+      this.path = new Path(path);
+      this.oldSnapshotName = oldSnapshotName;
+      this.snapshotName = snapshotName;
+      this.snapshotDiffStartPath = snapshotDiffStartPath;
+      this.snapshotDiffIndex = snapshotDiffIndex;
+    }
+
+    /**
+     * Executes the filesystem operation.
+     *
+     * @param fs filesystem instance to use.
+     * @return A serialized JSON string of snapshot diffs.
+     * @throws IOException thrown if an IO error occurred.
+     */
+    @Override
+    public String execute(FileSystem fs) throws IOException {
+      SnapshotDiffReportListing snapshotDiffReportListing = null;
+      if (fs instanceof DistributedFileSystem) {
+        DistributedFileSystem dfs = (DistributedFileSystem) fs;
+        snapshotDiffReportListing =
+            dfs.getSnapshotDiffReportListing(path, oldSnapshotName, snapshotName,
+                snapshotDiffStartPath, snapshotDiffIndex);
+      } else {
+        throw new UnsupportedOperationException("getSnapshotDiffListing is not "
+            + "supported for HttpFs on " + fs.getClass()
+            + ". Please check your fs.defaultFS configuration");
+      }
+      if (snapshotDiffReportListing != null) {
+        return JsonUtil.toJsonString(snapshotDiffReportListing);
+      } else {
+        return "";
+      }
+    }
+  }
+
   /**
    *  Executor that performs a getSnapshottableDirListing operation.
    */
diff --git a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/HttpFSParametersProvider.java b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/HttpFSParametersProvider.java
index f6c84dcae4e..41009c7b516 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/HttpFSParametersProvider.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/HttpFSParametersProvider.java
@@ -27,6 +27,7 @@ import org.apache.hadoop.lib.service.FileSystemAccess;
 import org.apache.hadoop.lib.wsrs.BooleanParam;
 import org.apache.hadoop.lib.wsrs.EnumParam;
 import org.apache.hadoop.lib.wsrs.EnumSetParam;
+import org.apache.hadoop.lib.wsrs.IntegerParam;
 import org.apache.hadoop.lib.wsrs.LongParam;
 import org.apache.hadoop.lib.wsrs.Param;
 import org.apache.hadoop.lib.wsrs.ParametersProvider;
@@ -112,6 +113,9 @@ public class HttpFSParametersProvider extends ParametersProvider {
     PARAMS_DEF.put(Operation.RENAMESNAPSHOT,
             new Class[] {OldSnapshotNameParam.class,
                 SnapshotNameParam.class});
+    PARAMS_DEF.put(Operation.GETSNAPSHOTDIFFLISTING,
+        new Class[] {OldSnapshotNameParam.class, SnapshotNameParam.class,
+            SnapshotDiffStartPathParam.class, SnapshotDiffIndexParam.class});
     PARAMS_DEF.put(Operation.GETSNAPSHOTDIFF,
         new Class[] {OldSnapshotNameParam.class,
             SnapshotNameParam.class});
@@ -669,6 +673,44 @@ public class HttpFSParametersProvider extends ParametersProvider {
     }
   }
 
+  /**
+   * Class for SnapshotDiffStartPath parameter.
+   */
+  public static class SnapshotDiffStartPathParam extends StringParam {
+
+    /**
+     * Parameter name.
+     */
+    public static final String NAME = HttpFSFileSystem.SNAPSHOT_DIFF_START_PATH;
+
+    /**
+     * Constructor.
+     */
+    public SnapshotDiffStartPathParam() {
+      super(NAME, "");
+    }
+
+  }
+
+  /**
+   * Class for SnapshotDiffStartPath parameter.
+   */
+  public static class SnapshotDiffIndexParam extends IntegerParam {
+
+    /**
+     * Parameter name.
+     */
+    public static final String NAME = HttpFSFileSystem.SNAPSHOT_DIFF_INDEX;
+
+    /**
+     * Constructor.
+     */
+    public SnapshotDiffIndexParam() {
+      super(NAME, null);
+    }
+
+  }
+
   /**
    * Class for FsAction parameter.
    */
diff --git a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/HttpFSServer.java b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/HttpFSServer.java
index d0d76d62899..399ff3bde9c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/HttpFSServer.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/main/java/org/apache/hadoop/fs/http/server/HttpFSServer.java
@@ -450,6 +450,24 @@ public class HttpFSServer {
       response = Response.ok(js).type(MediaType.APPLICATION_JSON).build();
       break;
     }
+    case GETSNAPSHOTDIFFLISTING: {
+      String oldSnapshotName = params.get(OldSnapshotNameParam.NAME,
+          OldSnapshotNameParam.class);
+      String snapshotName = params.get(SnapshotNameParam.NAME,
+          SnapshotNameParam.class);
+      String snapshotDiffStartPath = params
+          .get(HttpFSParametersProvider.SnapshotDiffStartPathParam.NAME,
+              HttpFSParametersProvider.SnapshotDiffStartPathParam.class);
+      Integer snapshotDiffIndex = params.get(HttpFSParametersProvider.SnapshotDiffIndexParam.NAME,
+          HttpFSParametersProvider.SnapshotDiffIndexParam.class);
+      FSOperations.FSGetSnapshotDiffListing command =
+          new FSOperations.FSGetSnapshotDiffListing(path, oldSnapshotName,
+              snapshotName, snapshotDiffStartPath, snapshotDiffIndex);
+      String js = fsExecute(user, command);
+      AUDIT_LOG.info("[{}]", path);
+      response = Response.ok(js).type(MediaType.APPLICATION_JSON).build();
+      break;
+    }
     case GETSNAPSHOTTABLEDIRECTORYLIST: {
       FSOperations.FSGetSnapshottableDirListing command =
           new FSOperations.FSGetSnapshottableDirListing();
diff --git a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/test/java/org/apache/hadoop/fs/http/client/BaseTestHttpFSWith.java b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/test/java/org/apache/hadoop/fs/http/client/BaseTestHttpFSWith.java
index d8dd7c7c362..9ef24ec734a 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/test/java/org/apache/hadoop/fs/http/client/BaseTestHttpFSWith.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-httpfs/src/test/java/org/apache/hadoop/fs/http/client/BaseTestHttpFSWith.java
@@ -42,6 +42,8 @@ import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.AppendTestUtil;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.DFSTestUtil;
+import org.apache.hadoop.hdfs.DFSUtil;
+import org.apache.hadoop.hdfs.DFSUtilClient;
 import org.apache.hadoop.hdfs.DistributedFileSystem;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys;
@@ -50,6 +52,7 @@ import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy;
 import org.apache.hadoop.hdfs.protocol.HdfsConstants;
 import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
 import org.apache.hadoop.hdfs.protocol.SnapshotDiffReport;
+import org.apache.hadoop.hdfs.protocol.SnapshotDiffReportListing;
 import org.apache.hadoop.hdfs.protocol.SnapshotException;
 import org.apache.hadoop.hdfs.protocol.SnapshottableDirectoryStatus;
 import org.apache.hadoop.hdfs.protocol.SnapshotStatus;
@@ -1198,7 +1201,7 @@ public abstract class BaseTestHttpFSWith extends HFSTestCase {
     ALLOW_SNAPSHOT, DISALLOW_SNAPSHOT, DISALLOW_SNAPSHOT_EXCEPTION,
     FILE_STATUS_ATTR, GET_SNAPSHOT_DIFF, GET_SNAPSHOTTABLE_DIRECTORY_LIST,
     GET_SNAPSHOT_LIST, GET_SERVERDEFAULTS, CHECKACCESS, SETECPOLICY,
-    SATISFYSTORAGEPOLICY
+    SATISFYSTORAGEPOLICY, GET_SNAPSHOT_DIFF_LISTING
   }
 
   private void operation(Operation op) throws Exception {
@@ -1335,6 +1338,9 @@ public abstract class BaseTestHttpFSWith extends HFSTestCase {
     case SATISFYSTORAGEPOLICY:
       testStoragePolicySatisfier();
       break;
+    case GET_SNAPSHOT_DIFF_LISTING:
+      testGetSnapshotDiffListing();
+      break;
     }
 
   }
@@ -1607,29 +1613,30 @@ public abstract class BaseTestHttpFSWith extends HFSTestCase {
       Path file2 = new Path(path, "file2");
       testCreate(file2, false);
       fs.createSnapshot(path, "snap2");
-      // Get snapshot diff
-      SnapshotDiffReport diffReport = null;
-      if (fs instanceof HttpFSFileSystem) {
-        HttpFSFileSystem httpFS = (HttpFSFileSystem) fs;
-        diffReport = httpFS.getSnapshotDiffReport(path, "snap1", "snap2");
-      } else if (fs instanceof WebHdfsFileSystem) {
-        WebHdfsFileSystem webHdfsFileSystem = (WebHdfsFileSystem) fs;
-        diffReport = webHdfsFileSystem.getSnapshotDiffReport(path,
-            "snap1", "snap2");
-      } else {
-        Assert.fail(fs.getClass().getSimpleName() +
-            " doesn't support getSnapshotDiff");
+
+      try {
+        // Get snapshot diff
+        SnapshotDiffReport diffReport = null;
+        if (fs instanceof HttpFSFileSystem) {
+          HttpFSFileSystem httpFS = (HttpFSFileSystem) fs;
+          diffReport = httpFS.getSnapshotDiffReport(path, "snap1", "snap2");
+        } else if (fs instanceof WebHdfsFileSystem) {
+          WebHdfsFileSystem webHdfsFileSystem = (WebHdfsFileSystem) fs;
+          diffReport = webHdfsFileSystem.getSnapshotDiffReport(path, "snap1", "snap2");
+        } else {
+          Assert.fail(fs.getClass().getSimpleName() + " doesn't support getSnapshotDiff");
+        }
+        // Verify result with DFS
+        DistributedFileSystem dfs =
+            (DistributedFileSystem) FileSystem.get(path.toUri(), this.getProxiedFSConf());
+        SnapshotDiffReport dfsDiffReport = dfs.getSnapshotDiffReport(path, "snap1", "snap2");
+        Assert.assertEquals(diffReport.toString(), dfsDiffReport.toString());
+      } finally {
+        // Cleanup
+        fs.deleteSnapshot(path, "snap2");
+        fs.deleteSnapshot(path, "snap1");
+        fs.delete(path, true);
       }
-      // Verify result with DFS
-      DistributedFileSystem dfs = (DistributedFileSystem)
-          FileSystem.get(path.toUri(), this.getProxiedFSConf());
-      SnapshotDiffReport dfsDiffReport =
-          dfs.getSnapshotDiffReport(path, "snap1", "snap2");
-      Assert.assertEquals(diffReport.toString(), dfsDiffReport.toString());
-      // Cleanup
-      fs.deleteSnapshot(path, "snap2");
-      fs.deleteSnapshot(path, "snap1");
-      fs.delete(path, true);
     }
   }
 
@@ -1951,4 +1958,102 @@ public abstract class BaseTestHttpFSWith extends HFSTestCase {
       dfs.delete(path1, true);
     }
   }
+
+  private void testGetSnapshotDiffListing() throws Exception {
+    if (!this.isLocalFS()) {
+      // Create a directory with snapshot allowed
+      Path path = new Path("/tmp/tmp-snap-test");
+      createSnapshotTestsPreconditions(path);
+      // Get the FileSystem instance that's being tested
+      FileSystem fs = this.getHttpFSFileSystem();
+      // Check FileStatus
+      Assert.assertTrue(fs.getFileStatus(path).isSnapshotEnabled());
+      // Create a file and take a snapshot
+      Path file1 = new Path(path, "file1");
+      testCreate(file1, false);
+      fs.createSnapshot(path, "snap1");
+      // Create another file and take a snapshot
+      Path file2 = new Path(path, "file2");
+      testCreate(file2, false);
+      fs.createSnapshot(path, "snap2");
+      // Get snapshot diff listing
+      try {
+        SnapshotDiffReportListing diffReportListing = null;
+        byte[] emptyBytes = new byte[] {};
+        if (fs instanceof HttpFSFileSystem) {
+          HttpFSFileSystem httpFS = (HttpFSFileSystem) fs;
+          diffReportListing =
+              httpFS.getSnapshotDiffReportListing(path, "snap1", "snap2", emptyBytes, -1);
+        } else if (fs instanceof WebHdfsFileSystem) {
+          WebHdfsFileSystem webHdfsFileSystem = (WebHdfsFileSystem) fs;
+          diffReportListing = webHdfsFileSystem
+              .getSnapshotDiffReportListing(path.toUri().getPath(), "snap1", "snap2", emptyBytes,
+                  -1);
+        } else {
+          Assert.fail(fs.getClass().getSimpleName() + " doesn't support getSnapshotDiff");
+        }
+        // Verify result with DFS
+        DistributedFileSystem dfs =
+            (DistributedFileSystem) FileSystem.get(path.toUri(), this.getProxiedFSConf());
+        SnapshotDiffReportListing dfsDiffReportListing =
+            dfs.getSnapshotDiffReportListing(path, "snap1", "snap2",
+                DFSUtil.bytes2String(emptyBytes), -1);
+        assertHttpFsReportListingWithDfsClient(diffReportListing, dfsDiffReportListing);
+      } finally {
+        // Cleanup
+        fs.deleteSnapshot(path, "snap2");
+        fs.deleteSnapshot(path, "snap1");
+        fs.delete(path, true);
+      }
+    }
+  }
+
+  private void assertHttpFsReportListingWithDfsClient(SnapshotDiffReportListing diffReportListing,
+      SnapshotDiffReportListing dfsDiffReportListing) {
+    Assert.assertEquals(diffReportListing.getCreateList().size(),
+        dfsDiffReportListing.getCreateList().size());
+    Assert.assertEquals(diffReportListing.getDeleteList().size(),
+        dfsDiffReportListing.getDeleteList().size());
+    Assert.assertEquals(diffReportListing.getModifyList().size(),
+        dfsDiffReportListing.getModifyList().size());
+    Assert.assertEquals(diffReportListing.getIsFromEarlier(),
+        dfsDiffReportListing.getIsFromEarlier());
+    Assert.assertEquals(diffReportListing.getLastIndex(), dfsDiffReportListing.getLastIndex());
+    Assert.assertEquals(DFSUtil.bytes2String(diffReportListing.getLastPath()),
+        DFSUtil.bytes2String(dfsDiffReportListing.getLastPath()));
+    int i = 0;
+    for (SnapshotDiffReportListing.DiffReportListingEntry entry : diffReportListing
+        .getCreateList()) {
+      SnapshotDiffReportListing.DiffReportListingEntry dfsDiffEntry =
+          dfsDiffReportListing.getCreateList().get(i);
+      Assert.assertEquals(entry.getDirId(), dfsDiffEntry.getDirId());
+      Assert.assertEquals(entry.getFileId(), dfsDiffEntry.getFileId());
+      Assert.assertArrayEquals(DFSUtilClient.byteArray2bytes(entry.getSourcePath()),
+          DFSUtilClient.byteArray2bytes(dfsDiffEntry.getSourcePath()));
+      i++;
+    }
+    i = 0;
+    for (SnapshotDiffReportListing.DiffReportListingEntry entry : diffReportListing
+        .getDeleteList()) {
+      SnapshotDiffReportListing.DiffReportListingEntry dfsDiffEntry =
+          dfsDiffReportListing.getDeleteList().get(i);
+      Assert.assertEquals(entry.getDirId(), dfsDiffEntry.getDirId());
+      Assert.assertEquals(entry.getFileId(), dfsDiffEntry.getFileId());
+      Assert.assertArrayEquals(DFSUtilClient.byteArray2bytes(entry.getSourcePath()),
+          DFSUtilClient.byteArray2bytes(dfsDiffEntry.getSourcePath()));
+      i++;
+    }
+    i = 0;
+    for (SnapshotDiffReportListing.DiffReportListingEntry entry : diffReportListing
+        .getModifyList()) {
+      SnapshotDiffReportListing.DiffReportListingEntry dfsDiffEntry =
+          dfsDiffReportListing.getModifyList().get(i);
+      Assert.assertEquals(entry.getDirId(), dfsDiffEntry.getDirId());
+      Assert.assertEquals(entry.getFileId(), dfsDiffEntry.getFileId());
+      Assert.assertArrayEquals(DFSUtilClient.byteArray2bytes(entry.getSourcePath()),
+          DFSUtilClient.byteArray2bytes(dfsDiffEntry.getSourcePath()));
+      i++;
+    }
+  }
+
 }
\ No newline at end of file