@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)
    {
      // create 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);
  }
}