@Test
public void testZookeeperCheckpointManager() {

  CustomizedConsumerConfig consumerConfig = new CustomizedConsumerConfig(new Properties());
  consumerConfig.setProperty("commit.zookeeper.connect", SRC_CLUSTER_ZK);
  String groupId = "ureplicator-cluster1-cluster2";
  ZookeeperCheckpointManager checkpointManager = new ZookeeperCheckpointManager(consumerConfig,
      groupId);
  for (TopicPartition topicPartition : offsetCommitMap.keySet()) {
    Long offset = checkpointManager.fetchOffset(topicPartition);
    Assert.assertEquals(offset, Long.valueOf(-1));
  }
  checkpointManager.commitOffset(offsetCommitMap);
  for (Map.Entry<TopicPartition, Long> entry : offsetCommitMap.entrySet()) {
    Long offset = checkpointManager.fetchOffset(entry.getKey());
    Assert.assertEquals(offset, entry.getValue());
  }
}