MockParticipant[] participants = new MockParticipant[n];
for (int i = 0; i < n; i++)
{
String instanceName = "localhost_" + (12918 + i);
participants[i] = new MockParticipant(clusterName, instanceName, ZK_ADDR, null);
participants[i].syncStart();
}
boolean result =
ClusterStateVerifier.verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR,
clusterName));
Assert.assertTrue(result);
// stop all participants
Thread.sleep(1000);
for (int i = 0; i < n; i++)
{
participants[i].syncStop();
}
// enable group message
ZKHelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
Builder keyBuilder = accessor.keyBuilder();
IdealState idealState = accessor.getProperty(keyBuilder.idealStates("TestDB0"));
idealState.setGroupMessageMode(true);
accessor.setProperty(keyBuilder.idealStates("TestDB0"), idealState);
// registry a message listener so we know how many message generated
TestZkChildListener listener = new TestZkChildListener();
_gZkClient.subscribeChildChanges(keyBuilder.messages("localhost_12918").getPath(), listener);
// restart all participants
for (int i = 0; i < n; i++)
{
String instanceName = "localhost_" + (12918 + i);
participants[i] = new MockParticipant(clusterName, instanceName, ZK_ADDR, null);
participants[i].syncStart();
}
result =
ClusterStateVerifier.verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR,