### Question:
HdfsSnapshotReplicator extends Configured implements Tool { protected void invokeCopy(String sourceStorageUrl, String targetStorageUrl, DistributedFileSystem sourceFs, DistributedFileSystem targetFs, String sourceDir, String targetDir, String currentSnapshotName) throws FalconException { try { Configuration jobConf = this.getConf(); DistCpOptions options = getDistCpOptions(sourceStorageUrl, targetStorageUrl, sourceFs, targetFs, sourceDir, targetDir, currentSnapshotName); DistCp distCp = new DistCp(jobConf, options); LOG.info("Started Snapshot based DistCp from {} to {} ", getStagingUri(sourceStorageUrl, sourceDir), getStagingUri(targetStorageUrl, targetDir)); Job distcpJob = distCp.execute(); LOG.info("Distp Hadoop job: {}", distcpJob.getJobID().toString()); LOG.info("Completed Snapshot based DistCp"); } catch (FalconException fe) { throw fe; } catch (Exception e) { throw new FalconException("Unable to replicate HDFS directory using snapshots.", e); } }  static void main(String[] args); @Override int run(String[] args);  }### Answer:
@Test public void replicationTest() throws Exception { sourceStorageUrl = ClusterHelper.getStorageUrl(sourceCluster); targetStorageUrl = ClusterHelper.getStorageUrl(targetCluster); DistributedFileSystem sourceFs = HdfsSnapshotUtil.getSourceFileSystem(cmd, new Configuration(getConf())); DistributedFileSystem targetFs = HdfsSnapshotUtil.getTargetFileSystem(cmd, new Configuration(getConf())); Path dir1 = new Path(sourceDir, "dir1"); miniDfs.mkdir(dir1, fsPermission); miniDfs.createSnapshot(sourceDir, "snapshot1"); invokeCopy(sourceStorageUrl, targetStorageUrl, sourceFs, targetFs, sourceDir.toString(), targetDir.toString(), "snapshot1"); miniDfs.createSnapshot(targetDir, "snapshot1"); Assert.assertTrue(miniDfs.exists(new Path(targetDir, "dir1"))); Path dir2 = new Path(sourceDir, "dir2"); miniDfs.mkdir(dir2, fsPermission); miniDfs.createSnapshot(sourceDir, "snapshot2"); invokeCopy(sourceStorageUrl, targetStorageUrl, sourceFs, targetFs, sourceDir.toString(), targetDir.toString(), "snapshot2"); miniDfs.createSnapshot(targetDir, "snapshot2"); Assert.assertTrue(miniDfs.exists(new Path(targetDir, "dir1"))); Assert.assertTrue(miniDfs.exists(new Path(targetDir, "dir2"))); miniDfs.delete(dir1, true); miniDfs.createSnapshot(sourceDir, "snapshot3"); invokeCopy(sourceStorageUrl, targetStorageUrl, sourceFs, targetFs, sourceDir.toString(), targetDir.toString(), "snapshot3"); miniDfs.createSnapshot(targetDir, "snapshot3"); Assert.assertFalse(miniDfs.exists(new Path(targetDir, "dir1"))); Assert.assertTrue(miniDfs.exists(new Path(targetDir, "dir2"))); }
@Test(dependsOnMethods = "replicationTest", expectedExceptions = FalconException.class, expectedExceptionsMessageRegExp = "Unable to find latest snapshot on targetDir " + "/apps/falcon/snapshot-replication/targetDir") public void removeSnapshotabilityOnTargetTest() throws Exception { miniDfs.deleteSnapshot(targetDir, "snapshot1"); miniDfs.deleteSnapshot(targetDir, "snapshot2"); miniDfs.deleteSnapshot(targetDir, "snapshot3"); miniDfs.disallowSnapshot(targetDir); Path dir1 = new Path(sourceDir, "dir4"); miniDfs.mkdir(dir1, fsPermission); miniDfs.createSnapshot(sourceDir, "snapshot4"); invokeCopy(sourceStorageUrl, targetStorageUrl, miniDfs, miniDfs, sourceDir.toString(), targetDir.toString(), "snapshot4"); }