Future<String> future = c1.putAsync(key, newValue);
barrier.await(10, TimeUnit.SECONDS);
// Block the rebalance confirmation on nonOwnerCache
CheckPoint checkPoint = new CheckPoint();
// We have to wait until non owner has the new topology installed before transferring state
waitUntilToplogyInstalled(c3, checkPoint);
// Now make sure the owners doesn't have the new topology installed
waitUntilBeforeTopologyInstalled(c1, checkPoint);
waitUntilBeforeTopologyInstalled(c2, checkPoint);
// Now force 1 and 3 to be owners so then 3 will get invalidation and state transfer
factory.setOwnerIndexes(0, 2);
EmbeddedCacheManager cm = addClusterEnabledCacheManager(configuration);
Future<Void> join = fork(new Callable<Void>() {
@Override
public Void call() throws Exception {
waitForClusterToForm(cacheName);
log.debug("4th has joined");
return null;
}
});
checkPoint.awaitStrict("post_topology_installed_invoked_" + c3, 10, TimeUnit.SECONDS);
checkPoint.awaitStrict("pre_topology_installed_invoked_" + c1, 10, TimeUnit.SECONDS);
checkPoint.awaitStrict("pre_topology_installed_invoked_" + c2, 10, TimeUnit.SECONDS);
barrier.await(10, TimeUnit.SECONDS);
assertEquals(startValue, future.get(10, TimeUnit.SECONDS));
checkPoint.triggerForever("post_topology_installed_released_" + c3);
checkPoint.triggerForever("pre_topology_installed_released_" + c1);
checkPoint.triggerForever("pre_topology_installed_released_" + c2);
join.get(10, TimeUnit.SECONDS);
assertIsInContainerImmortal(c1, key);
assertIsNotInL1(c2, key);