checkIfCancelled();
// Only fetch the data from the cache store if the cache store is not shared
CacheStore cacheStore = stateTransferManager.getCacheStoreForStateTransfer();
if (cacheStore != null) {
for (Object key : cacheStore.loadAllKeys(new ReadOnlyDataContainerBackedKeySet(dataContainer))) {
rebalance(key, null, numOwners, chOld, chNew, cacheStore, states, keysToRemove);
}
} else {
if (trace) log.trace("No cache store or cache store is shared, not rebalancing stored keys");
}
checkIfCancelled();
// Push any remaining state chunks
for (Map.Entry<Address, Collection<InternalCacheEntry>> entry : states.entrySet()) {
pushPartialState(Collections.singleton(entry.getKey()), entry.getValue(), null);
}
// Push locks if the cache is transactional and it is distributed
if (transactionTable != null) {
log.debug("Starting lock migration");
Map<Address, Collection<LockInfo>> locksToMigrate = new HashMap<Address, Collection<LockInfo>>();
rebalanceLocks(numOwners, locksToMigrate, transactionTable.getRemoteTransactions());
rebalanceLocks(numOwners, locksToMigrate, transactionTable.getLocalTransactions());
for (Map.Entry<Address, Collection<LockInfo>> e : locksToMigrate.entrySet()) {
pushPartialState(Collections.singleton(e.getKey()), null, e.getValue());
}
}
// And wait for all the push RPCs to end
finishPushingState();
} else if (initialView) {
// Only remove data from the cache store if the cache store is not shared
CacheStore cacheStore = stateTransferManager.getCacheStoreForStateTransfer();
if (cacheStore != null) {
if (trace) log.trace("Non-shared cache store, cleaning up persisted entries that we don't own after we joined the cache");
for (Object key : cacheStore.loadAllKeys(new ReadOnlyDataContainerBackedKeySet(dataContainer))) {
if (!chNew.isKeyLocalToAddress(self, key, numOwners)) {
keysToRemove.add(key);
}
}
}