CheckPointCreator { public void createCheckPoint() throws Exception { Cluster destinationCluster = config.getClusters().get(destnCluster); CheckpointProvider provider = new FSCheckpointProvider( destinationCluster.getCheckpointDir()); boolean isMerge = false; Set<String> mergingStream = destinationCluster .getPrimaryDestinationStreams(); if (mergingStream.contains(stream)) { if (srcCluster == null) { SourceStream srcStream = config.getSourceStreams().get(stream); sourceClusters.addAll(srcStream.getSourceClusters()); } isMerge = true; } else if (destinationCluster.getDestinationStreams().containsKey(stream)) { if (srcCluster == null) { sourceClusters.add(config.getPrimaryClusterForDestinationStream(stream) .getName()); } isMerge = false; } else { LOG.error("Stream " + stream + " is not destination stream of cluster " + destnCluster); } if (srcCluster != null) { sourceClusters.add(srcCluster); } for (String source : sourceClusters) { Cluster srcCluster = config.getClusters().get(source); FileSystem srcFS = FileSystem.get(srcCluster.getHadoopConf()); String checkPointValue; if (isMerge) { checkPointValue = srcCluster.getLocalDestDir(stream, date); } else { checkPointValue = srcCluster.getFinalDestDir(stream, date.getTime()); } Path checkPoinPath = new Path(checkPointValue); if (!srcFS.exists(checkPoinPath)) throw new Exception("Path " + checkPointValue + " doesn't exist,hence checkpoint can't be created for source " + source); provider.checkpoint(getCheckPointKey(stream, source, isMerge), checkPointValue.getBytes()); } } CheckPointCreator(ConduitConfig config, String sourceCluster,
      String destinationCluster, String stream, Date date); void createCheckPoint();  }
@Test(expectedExceptions = { Exception.class }) public void testCheckPointMirrorCheckPointPathDoesntExist() throws Exception { Date date = new Date(); CheckPointCreator creator = new CheckPointCreator(config, null, CLUSTER2, STREAM2, date); creator.createCheckPoint(); }