ClusterCheckpointPersistenceProvider extends
        CheckpointPersistenceProviderAbstract { static public boolean createCluster(String zkAddr, String clusterName) { boolean created = false; ZkClient zkClient = null; try { zkClient = new ZkClient(zkAddr, ZkClient.DEFAULT_SESSION_TIMEOUT, ZkClient.DEFAULT_CONNECTION_TIMEOUT, new ZNRecordSerializer()); ZKHelixAdmin admin = new ZKHelixAdmin(zkClient); admin.addCluster(clusterName, false); created=true; } catch (HelixException e) { LOG.warn("Warn! Cluster might already exist! " + clusterName); created=false; } finally { if (zkClient != null) { zkClient.close(); } } return created; } ClusterCheckpointPersistenceProvider(long id);  ClusterCheckpointPersistenceProvider(String id);  ClusterCheckpointPersistenceProvider(long id, Config config);  ClusterCheckpointPersistenceProvider(String id, Config config);  ClusterCheckpointPersistenceProvider(long id, StaticConfig config);  ClusterCheckpointPersistenceProvider(String id, StaticConfig config); static boolean createCluster(String zkAddr, String clusterName); static void close(String clusterName); @Deprecated /** * * @param sourceNames * @param checkpoint * Persist a checkpoint in legacy location * @throws IOException */ void storeCheckpointLegacy(List<String> sourceNames,
            Checkpoint checkpoint); @Override void storeCheckpoint(List<String> sourceNames, Checkpoint checkpoint); @Deprecated /** * read legacy checkpoint without migration * * @param sources * @return checkpoint if found or null otherwise */ Checkpoint loadCheckpointLegacy(List<String> sources); @Override Checkpoint loadCheckpoint(List<String> sourceNames); Set<String> getSourceNames(); void removeCheckpointLegacy(List<String> sourceNames); @Override void removeCheckpoint(List<String> sourceNames); long getNumWritesSkipped();  }
@Test public void testMultipleClusterCheckpointPersistence() { try { String[] partitionIds = { "1", "2", "3", "4", "5", "6" }; String[] clusters = { "tcluster1", "tcluster2", "tcluster3" }; ArrayList<CheckpointRW> cpRws = new ArrayList<TestClusterCheckpointPersistenceProvider.CheckpointRW>(); for (String c : clusters) { ClusterCheckpointPersistenceProvider.createCluster(zkAddr, c); for (String p : partitionIds) { cpRws.add(new CheckpointRW(c, p, RngUtils .randomPositiveLong())); } } for (CheckpointRW cpRW : cpRws) { cpRW.start(); } for (CheckpointRW cpRW : cpRws) { cpRW.join(10000); Assert.assertFalse(cpRW.hasError()); } } catch (Exception e) { Assert.assertTrue(false); } }