class 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)) {
      // stream is getting merged here
      if (srcCluster == null) {// no src clusters provided;create checkpoint for
        // all src clusters
        SourceStream srcStream = config.getSourceStreams().get(stream);
        sourceClusters.addAll(srcStream.getSourceClusters());
      }
      isMerge = true;
    } else if (destinationCluster.getDestinationStreams().containsKey(stream)) {
      // stream is getting mirrored since its a destination stream and not
      // primary destination
      if (srcCluster == null) {// no src clusters provided;create checkpoint for
        // all src clusters
        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());
    }
  }

  public  CheckPointCreator(ConduitConfig config, String sourceCluster,
      String destinationCluster, String stream, Date date);

}