ZookeeperCheckpointManager implements ICheckPointManager { public ZookeeperCheckpointManager(CustomizedConsumerConfig config, String groupId) { this.commitZkClient = ZkUtils.createZkClient( config.getProperty(Constants.COMMIT_ZOOKEEPER_SERVER_CONFIG), config.getInt(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000), config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 10000)); this.groupId = groupId; this.commitExecutor = Executors.newFixedThreadPool(10); KafkaUReplicatorMetricsReporter.get().registerMetric(COMMIT_FAILURE_METER_NAME, commitFailure); } ZookeeperCheckpointManager(CustomizedConsumerConfig config, String groupId); void commitOffset(Map<TopicPartition, Long> topicPartitionOffsets); Long fetchOffset(TopicPartition topicPartition); @Override void shutdown();  }
@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()); } }