/*
* JBoss, Home of Professional Open Source
*
* Distributable under LGPL license.
* See terms of license at gnu.org.
*/
package org.jboss.cache.buddyreplication;
import org.jboss.cache.CacheSPI;
import org.jboss.cache.Fqn;
import org.jboss.cache.util.TestingUtil;
import org.jgroups.JChannel;
import org.jgroups.protocols.DISCARD;
import static org.testng.AssertJUnit.*;
import org.testng.annotations.Test;
import java.util.List;
/**
* Tests behaviour when data owners fail - essentially this tests data gravitation
*
* @author <a href="mailto:manik AT jboss DOT org">Manik Surtani (manik AT jboss DOT org)</a>
*/
@Test(groups = "functional")
public class BuddyReplicationFailoverTest extends BuddyReplicationTestsBase
{
protected boolean optimisticLocks = false;
private String key = "key";
private String value = "value";
BuddyFqnTransformer fqnTransformer = new BuddyFqnTransformer();
public void testDataGravitationKillOwner() throws Exception
{
testDataGravitation(true);
}
public void testDataGravitationDontKillOwner() throws Exception
{
testDataGravitation(false);
}
private void testDataGravitation(boolean killOwner) throws Exception
{
List<CacheSPI<Object, Object>> caches = createCaches(3, false, true, optimisticLocks);
cachesTL.set(caches);
Fqn fqn = Fqn.fromString("/test");
Fqn backupFqn = fqnTransformer.getBackupFqn(caches.get(0).getLocalAddress(), fqn);
TestingUtil.dumpCacheContents(caches);
caches.get(0).put(fqn, key, value);
TestingUtil.dumpCacheContents(caches);
assertEquals("Value should exist", value, caches.get(0).get(fqn, key));
TestingUtil.dumpCacheContents(caches);
// use exists instead of get() to prevent going up the interceptor stack
assertTrue("Should be false", !caches.get(1).exists(fqn));
assertTrue("Should be false", !caches.get(2).exists(fqn));
assertFalse("Should be false", caches.get(0).exists(backupFqn));
assertTrue("Value be true", caches.get(1).exists(backupFqn));
assertFalse("Should be false", caches.get(2).exists(backupFqn));
if (killOwner)
{
System.out.println("***** About to kill original data owner (" + caches.get(0).getLocalAddress() + "). *****");
// forcefully kill data owner.
killChannel(caches.get(0), caches.get(1), 2);
System.out.println("Killed. Testing backup roots.");
TestingUtil.dumpCacheContents(caches);
// assert that the remaining caches have picked new buddies. Cache 1 should have cache 2's backup data.
assert caches.get(1).peek(fqnTransformer.getBackupRoot(caches.get(2).getLocalAddress()), false) != null : "Should have new buddy's backup root.";
assert caches.get(1).peek(fqnTransformer.getBackupRoot(caches.get(1).getLocalAddress()), false) == null : "Should not have self as a backup root.";
assert caches.get(1).peek(fqnTransformer.getBackupRoot(caches.get(0).getLocalAddress()), false) == null : "Should not have dead node as a backup root.";
assert caches.get(1).peek(Fqn.fromRelativeElements(fqnTransformer.getDeadBackupRoot(caches.get(0).getLocalAddress()), 1), false) != null : "Should have dead node as a defunct backup root.";
assert caches.get(2).peek(fqnTransformer.getBackupRoot(caches.get(2).getLocalAddress()), false) == null : "Should not have self as a backup root.";
assert caches.get(2).peek(fqnTransformer.getBackupRoot(caches.get(1).getLocalAddress()), false) != null : "Should have new buddy's backup root.";
assert caches.get(2).peek(fqnTransformer.getBackupRoot(caches.get(0).getLocalAddress()), false) == null : "Should not have dead node as a backup root.";
assert caches.get(2).peek(Fqn.fromRelativeElements(fqnTransformer.getDeadBackupRoot(caches.get(0).getLocalAddress()), 1), false) == null : "Should not have dead node as a defunct backup root.";
}
System.out.println("***** Killed original data owner, about to call a get on a different cache instance. *****");
// according to data gravitation, a call to *any* cache should retrieve the data, and move the data to the new cache.
assertEquals("Value should have gravitated", value, caches.get(2).get(fqn, key));
delay(); // cleanup commands are async
TestingUtil.dumpCacheContents(caches);
// now lets test the eviction part of gravitation
Fqn newBackupFqn = fqnTransformer.getBackupFqn(caches.get(2).getLocalAddress(), fqn);
// use exists instead of get() to prevent going up the interceptor stack
if (!killOwner)
{
assertTrue("Should be false", !caches.get(0).exists(fqn));
}
else
{
assert caches.get(1).peek(fqnTransformer.getBackupRoot(caches.get(2).getLocalAddress()), false) != null : "Should have new buddy's backup root.";
assert caches.get(1).peek(fqnTransformer.getBackupRoot(caches.get(1).getLocalAddress()), false) == null : "Should not have self as a backup root.";
assert caches.get(1).peek(fqnTransformer.getBackupRoot(caches.get(0).getLocalAddress()), false) == null : "Should not have dead node as a backup root.";
assert caches.get(1).peek(Fqn.fromRelativeElements(fqnTransformer.getDeadBackupRoot(caches.get(0).getLocalAddress()), 1), false) == null : "Should not have dead node as a defunct backup root.";
assert caches.get(1).peek(fqnTransformer.getDeadBackupRoot(caches.get(0).getLocalAddress()), false) == null : "Should not have dead node as a defunct backup root.";
assert caches.get(2).peek(fqnTransformer.getBackupRoot(caches.get(2).getLocalAddress()), false) == null : "Should not have self as a backup root.";
assert caches.get(2).peek(fqnTransformer.getBackupRoot(caches.get(1).getLocalAddress()), false) != null : "Should have new buddy's backup root.";
assert caches.get(2).peek(fqnTransformer.getBackupRoot(caches.get(0).getLocalAddress()), false) == null : "Should not have dead node as a backup root.";
assert caches.get(2).peek(Fqn.fromRelativeElements(fqnTransformer.getDeadBackupRoot(caches.get(0).getLocalAddress()), 1), false) == null : "Should not have dead node as a defunct backup root.";
}
assertTrue("Should be false", !caches.get(1).exists(fqn));
// the old backup should no longer exist
if (!killOwner) assertFalse("Should be null", caches.get(0).exists(backupFqn));
assertFalse("Should be null", caches.get(1).exists(backupFqn));
assertFalse("Should be null", caches.get(2).exists(backupFqn));
// and the backup should now exist in caches.get(2)'s buddy which is caches.get(0)
if (killOwner)
{
assertEquals("Value should exist", value, caches.get(1).get(newBackupFqn, key));
}
else
{
assertEquals("Value should exist", value, caches.get(0).get(newBackupFqn, key));
assertFalse("Should be null", caches.get(1).exists(newBackupFqn));
}
assertFalse("Should be null", caches.get(2).exists(newBackupFqn));
}
private void killChannel(CacheSPI cacheToKill, CacheSPI anotherCache, int finalExpectedClusterSize)
{
JChannel channel = (JChannel) cacheToKill.getRPCManager().getChannel();
DISCARD discard = (DISCARD) channel.getProtocolStack().findProtocol(DISCARD.class);
if (discard != null)
{
discard.setDiscardAll(true);
TestingUtil.blockUntilViewReceived(anotherCache, finalExpectedClusterSize, 10000, false);
// an extra second, for some reason we need this.
TestingUtil.sleepThread(1000);
}
}
public void testDataReplicationSuppression() throws Exception
{
List<CacheSPI<Object, Object>> caches = createCaches(3, false, false, optimisticLocks);
cachesTL.set(caches);
Fqn fqn = Fqn.fromString("/test");
Fqn backupFqn = fqnTransformer.getBackupFqn(caches.get(0).getLocalAddress(), fqn);
caches.get(0).put(fqn, key, value);
TestingUtil.dumpCacheContents(caches);
assertEquals("value", caches.get(0).get(fqn, key));
assertFalse(caches.get(0).exists(backupFqn));
assertEquals("value", caches.get(1).get(backupFqn, key));
assertFalse(caches.get(1).exists(fqn));
assertFalse(caches.get(2).exists(fqn));
assertFalse(caches.get(2).exists(backupFqn));
assertNoLocks(caches);
backupFqn = fqnTransformer.getBackupFqn(caches.get(1).getLocalAddress(), fqn);
System.out.println("*** Calling get() on cache[1] with force option");
caches.get(1).getInvocationContext().getOptionOverrides().setForceDataGravitation(true);
assertEquals("value", caches.get(1).get(fqn, key));
delay(); // cleanup commands are async
TestingUtil.dumpCacheContents(caches);
assertFalse(caches.get(1).exists(backupFqn));
assertEquals("value", caches.get(2).get(backupFqn, key));
assertFalse(caches.get(2).exists(fqn));
assertFalse(caches.get(0).exists(fqn));
assertFalse(caches.get(0).exists(backupFqn));
}
public void testSubtreeRetrieval() throws Exception
{
List<CacheSPI<Object, Object>> caches = createCaches(3, false, true, optimisticLocks);
cachesTL.set(caches);
Fqn fqn = Fqn.fromString("/test");
Fqn fqn2 = Fqn.fromString("/test/subtree");
Fqn backupFqn = fqnTransformer.getBackupFqn(caches.get(0).getLocalAddress(), fqn);
Fqn backupFqn2 = fqnTransformer.getBackupFqn(caches.get(0).getLocalAddress(), fqn2);
caches.get(0).put(fqn, key, value);
caches.get(0).put(fqn2, key, value);
// test backup replication to buddy
assertEquals(value, caches.get(0).get(fqn, key));
assertEquals(value, caches.get(0).get(fqn2, key));
assertEquals(value, caches.get(1).get(backupFqn, key));
assertEquals(value, caches.get(1).get(backupFqn2, key));
assertTrue(!caches.get(0).exists(backupFqn));
assertTrue(!caches.get(0).exists(backupFqn2));
assertTrue(!caches.get(1).exists(fqn));
assertTrue(!caches.get(1).exists(fqn2));
assertTrue(!caches.get(2).exists(fqn));
assertTrue(!caches.get(2).exists(fqn2));
assertTrue(!caches.get(2).exists(backupFqn));
assertTrue(!caches.get(2).exists(backupFqn2));
assertNoLocks(caches);
// gravitate to 2:
caches.get(2).getNode(fqn); // expect entire subtree to gravitate.
delay(); // cleanup commands are async
Fqn newBackupFqn = fqnTransformer.getBackupFqn(caches.get(2).getLocalAddress(), fqn);
Fqn newBackupFqn2 = fqnTransformer.getBackupFqn(caches.get(2).getLocalAddress(), fqn2);
assertEquals(value, caches.get(2).get(fqn, key));
assertTrue(caches.get(2).exists(fqn2));
assertEquals(value, caches.get(0).get(newBackupFqn, key));
assertTrue(caches.get(0).exists(newBackupFqn2));
assertTrue(!caches.get(2).exists(newBackupFqn));
assertTrue(!caches.get(2).exists(newBackupFqn2));
assertTrue(!caches.get(0).exists(fqn));
assertTrue(!caches.get(0).exists(fqn2));
assertTrue(!caches.get(1).exists(fqn));
assertTrue(!caches.get(1).exists(fqn2));
assertTrue(!caches.get(1).exists(newBackupFqn));
assertTrue(!caches.get(1).exists(newBackupFqn2));
TestingUtil.dumpCacheContents(caches);
for (CacheSPI<Object, Object> cache : caches)
{
assertTrue(!cache.exists(backupFqn));
assertTrue(!cache.exists(backupFqn2));
}
assertNoLocks(caches);
}
protected void delay()
{
TestingUtil.sleepThread(250);
}
}