X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FDistributedDataStoreRemotingIntegrationTest.java;h=ff44f01402f33f3f57f0bab01d169e78d3eeab1d;hp=0a82a1fffb43299d5b5aa43de5c2011b067d253b;hb=b47ffc2df37ef67559274068d651c86aa666cbc3;hpb=4d1709660b7af992d4c382a2a38debb5c7d64fb9 diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreRemotingIntegrationTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreRemotingIntegrationTest.java index 0a82a1fffb..ff44f01402 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreRemotingIntegrationTest.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreRemotingIntegrationTest.java @@ -15,6 +15,7 @@ import static org.mockito.Matchers.any; import static org.mockito.Matchers.eq; import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.verify; + import akka.actor.ActorRef; import akka.actor.ActorSelection; import akka.actor.ActorSystem; @@ -27,6 +28,7 @@ import akka.pattern.Patterns; import akka.testkit.JavaTestKit; import com.google.common.base.Optional; import com.google.common.base.Supplier; +import com.google.common.base.Throwables; import com.google.common.collect.ImmutableMap; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; @@ -41,16 +43,12 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; import org.mockito.Mockito; -import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier; import org.opendaylight.controller.cluster.databroker.ConcurrentDOMDataBroker; import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder; -import org.opendaylight.controller.cluster.datastore.IntegrationTestKit.ShardStatsVerifier; -import org.opendaylight.controller.cluster.datastore.MemberNode.RaftStateVerifier; import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; import org.opendaylight.controller.cluster.datastore.exceptions.ShardLeaderNotRespondingException; -import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats; import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction; import org.opendaylight.controller.cluster.datastore.messages.GetShardDataTree; @@ -58,9 +56,8 @@ import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransact import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply; import org.opendaylight.controller.cluster.datastore.modification.MergeModification; import org.opendaylight.controller.cluster.datastore.modification.WriteModification; -import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries; -import org.opendaylight.controller.cluster.raft.client.messages.OnDemandRaftState; import org.opendaylight.controller.cluster.raft.client.messages.Shutdown; +import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries; import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy; import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal; import org.opendaylight.controller.md.cluster.datastore.model.CarsModel; @@ -105,8 +102,10 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { private static final String[] CARS_AND_PEOPLE = {"cars", "people"}; private static final String[] CARS = {"cars"}; - private static final Address MEMBER_1_ADDRESS = AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558"); - private static final Address MEMBER_2_ADDRESS = AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2559"); + private static final Address MEMBER_1_ADDRESS = AddressFromURIString.parse( + "akka://cluster-test@127.0.0.1:2558"); + private static final Address MEMBER_2_ADDRESS = AddressFromURIString.parse( + "akka://cluster-test@127.0.0.1:2559"); private static final String MODULE_SHARDS_CARS_ONLY_1_2 = "module-shards-cars-member-1-and-2.conf"; private static final String MODULE_SHARDS_CARS_PEOPLE_1_2 = "module-shards-member1-and-2.conf"; @@ -120,13 +119,13 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2); private final DatastoreContext.Builder followerDatastoreContextBuilder = - DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5). - customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName()); + DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5) + .customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName()); private final TransactionIdentifier tx1 = nextTransactionId(); private final TransactionIdentifier tx2 = nextTransactionId(); - private DistributedDataStore followerDistributedDataStore; - private DistributedDataStore leaderDistributedDataStore; + private AbstractDataStore followerDistributedDataStore; + private AbstractDataStore leaderDistributedDataStore; private IntegrationTestKit followerTestKit; private IntegrationTestKit leaderTestKit; @@ -156,45 +155,48 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { JavaTestKit.shutdownActorSystem(follower2System); } - private void initDatastoresWithCars(String type) { + private void initDatastoresWithCars(final String type) { initDatastores(type, MODULE_SHARDS_CARS_ONLY_1_2, CARS); } - private void initDatastoresWithCarsAndPeople(String type) { + private void initDatastoresWithCarsAndPeople(final String type) { initDatastores(type, MODULE_SHARDS_CARS_PEOPLE_1_2, CARS_AND_PEOPLE); } - private void initDatastores(String type, String moduleShardsConfig, String[] shards) { + private void initDatastores(final String type, final String moduleShardsConfig, final String[] shards) { leaderTestKit = new IntegrationTestKit(leaderSystem, leaderDatastoreContextBuilder); leaderDistributedDataStore = leaderTestKit.setupDistributedDataStore(type, moduleShardsConfig, false, shards); followerTestKit = new IntegrationTestKit(followerSystem, followerDatastoreContextBuilder); - followerDistributedDataStore = followerTestKit.setupDistributedDataStore(type, moduleShardsConfig, false, shards); + followerDistributedDataStore = followerTestKit.setupDistributedDataStore(type, moduleShardsConfig, false, + shards); leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), shards); } - private static void verifyCars(DOMStoreReadTransaction readTx, MapEntryNode... entries) throws Exception { + private static void verifyCars(final DOMStoreReadTransaction readTx, final MapEntryNode... entries) + throws Exception { Optional> optional = readTx.read(CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS); assertEquals("isPresent", true, optional.isPresent()); CollectionNodeBuilder listBuilder = ImmutableNodes.mapNodeBuilder(CarsModel.CAR_QNAME); - for(NormalizedNode entry: entries) { + for (NormalizedNode entry: entries) { listBuilder.withChild((MapEntryNode) entry); } assertEquals("Car list node", listBuilder.build(), optional.get()); } - private static void verifyNode(DOMStoreReadTransaction readTx, YangInstanceIdentifier path, NormalizedNode expNode) - throws Exception { + private static void verifyNode(final DOMStoreReadTransaction readTx, final YangInstanceIdentifier path, + final NormalizedNode expNode) throws Exception { Optional> optional = readTx.read(path).get(5, TimeUnit.SECONDS); assertEquals("isPresent", true, optional.isPresent()); assertEquals("Data node", expNode, optional.get()); } - private static void verifyExists(DOMStoreReadTransaction readTx, YangInstanceIdentifier path) throws Exception { + private static void verifyExists(final DOMStoreReadTransaction readTx, final YangInstanceIdentifier path) + throws Exception { Boolean exists = readTx.exists(path).get(5, TimeUnit.SECONDS); assertEquals("exists", true, exists); } @@ -250,8 +252,8 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { ActorSystem newSystem = ActorSystem.create("reinstated-member2", ConfigFactory.load().getConfig("Member2")); - try (DistributedDataStore member2Datastore = new IntegrationTestKit(newSystem, leaderDatastoreContextBuilder). - setupDistributedDataStore(testName, "module-shards-member2", true, CARS_AND_PEOPLE)) { + try (AbstractDataStore member2Datastore = new IntegrationTestKit(newSystem, leaderDatastoreContextBuilder) + .setupDistributedDataStore(testName, "module-shards-member2", true, CARS_AND_PEOPLE)) { verifyCars(member2Datastore.newReadOnlyTransaction(), car2); } @@ -378,7 +380,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { } @Test - public void testTransactionChainWithMultipleShards() throws Exception{ + public void testTransactionChainWithMultipleShards() throws Exception { initDatastoresWithCarsAndPeople("testTransactionChainWithMultipleShards"); DOMStoreTransactionChain txChain = followerDistributedDataStore.createTransactionChain(); @@ -447,8 +449,8 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { DOMDataWriteTransaction writeTx = txChain.newWriteOnlyTransaction(); ContainerNode invalidData = ImmutableContainerNodeBuilder.create().withNodeIdentifier( - new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME)). - withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build(); + new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME)) + .withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build(); writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, invalidData); @@ -482,8 +484,8 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { writeTx.put(LogicalDatastoreType.CONFIGURATION, PeopleModel.BASE_PATH, PeopleModel.emptyContainer()); ContainerNode invalidData = ImmutableContainerNodeBuilder.create().withNodeIdentifier( - new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME)). - withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build(); + new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME)) + .withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build(); // Note that merge will validate the data and fail but put succeeds b/c deep validation is not // done for put for performance reasons. @@ -523,21 +525,22 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { // Switch the leader to the follower - sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder. - shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null)); + sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder + .shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null)); JavaTestKit.shutdownActorSystem(leaderSystem, null, true); + Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS); followerTestKit.waitUntilNoLeader(followerDistributedDataStore.getActorContext(), CARS); leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1")); Cluster.get(leaderSystem).join(MEMBER_2_ADDRESS); - DatastoreContext.Builder newMember1Builder = DatastoreContext.newBuilder(). - shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5); + DatastoreContext.Builder newMember1Builder = DatastoreContext.newBuilder() + .shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5); IntegrationTestKit newMember1TestKit = new IntegrationTestKit(leaderSystem, newMember1Builder); - try (DistributedDataStore ds = + try (AbstractDataStore ds = newMember1TestKit.setupDistributedDataStore(testName, MODULE_SHARDS_CARS_ONLY_1_2, false, CARS)) { followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorContext(), CARS); @@ -582,7 +585,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { carsFollowerShard.get().tell(readyLocal, followerTestKit.getRef()); Object resp = followerTestKit.expectMsgClass(Object.class); - if(resp instanceof akka.actor.Status.Failure) { + if (resp instanceof akka.actor.Status.Failure) { throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause()); } @@ -601,7 +604,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { carsFollowerShard.get().tell(readyLocal, followerTestKit.getRef()); resp = followerTestKit.expectMsgClass(Object.class); - if(resp instanceof akka.actor.Status.Failure) { + if (resp instanceof akka.actor.Status.Failure) { throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause()); } @@ -649,7 +652,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { carsFollowerShard.get().tell(forwardedReady, followerTestKit.getRef()); Object resp = followerTestKit.expectMsgClass(Object.class); - if(resp instanceof akka.actor.Status.Failure) { + if (resp instanceof akka.actor.Status.Failure) { throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause()); } @@ -669,7 +672,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { carsFollowerShard.get().tell(forwardedReady, followerTestKit.getRef()); resp = followerTestKit.expectMsgClass(Object.class); - if(resp instanceof akka.actor.Status.Failure) { + if (resp instanceof akka.actor.Status.Failure) { throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause()); } @@ -705,12 +708,11 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { // Wait for the commit to be replicated to the follower. - MemberNode.verifyRaftState(followerDistributedDataStore, "cars", new RaftStateVerifier() { - @Override - public void verify(OnDemandRaftState raftState) { - assertEquals("getLastApplied", 0, raftState.getLastApplied()); - } - }); + MemberNode.verifyRaftState(followerDistributedDataStore, "cars", + raftState -> assertEquals("getLastApplied", 0, raftState.getLastApplied())); + + MemberNode.verifyRaftState(followerDistributedDataStore, "people", + raftState -> assertEquals("getLastApplied", 0, raftState.getLastApplied())); // Prepare, ready and canCommit a WO tx that writes to 2 shards. This will become the current tx in // the leader shard. @@ -733,7 +735,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { carIndex++; NormalizedNode people = PeopleModel.newPersonMapNode(); writeTx2.write(PeopleModel.PERSON_LIST_PATH, people); - DOMStoreThreePhaseCommitCohort writeTx2Cohort = writeTx2.ready(); + final DOMStoreThreePhaseCommitCohort writeTx2Cohort = writeTx2.ready(); // Prepare another WO that writes to a single shard and thus will be directly committed on ready. This // tx writes 5 cars so 2 BatchedModidifications messages will be sent initially and cached in the @@ -741,7 +743,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { // sent on ready. DOMStoreWriteTransaction writeTx3 = followerDistributedDataStore.newWriteOnlyTransaction(); - for(int i = 1; i <= 5; i++, carIndex++) { + for (int i = 1; i <= 5; i++, carIndex++) { cars.add(CarsModel.newCarEntry("car" + carIndex, BigInteger.valueOf(carIndex))); writeTx3.write(CarsModel.newCarPath("car" + carIndex), cars.getLast()); } @@ -761,33 +763,30 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { cars.add(CarsModel.newCarEntry("car" + carIndex, BigInteger.valueOf(carIndex))); readWriteTx.write(CarsModel.newCarPath("car" + carIndex), cars.getLast()); - IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars", new ShardStatsVerifier() { - @Override - public void verify(ShardStats stats) { - assertEquals("getReadWriteTransactionCount", 1, stats.getReadWriteTransactionCount()); - } - }); + IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars", + stats -> assertEquals("getReadWriteTransactionCount", 1, stats.getReadWriteTransactionCount())); // Disable elections on the leader so it switches to follower. - sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder. - customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName()). - shardElectionTimeoutFactor(10)); + sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder + .customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName()) + .shardElectionTimeoutFactor(10)); + Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS); leaderTestKit.waitUntilNoLeader(leaderDistributedDataStore.getActorContext(), "cars"); // Submit all tx's - the messages should get queued for retry. ListenableFuture writeTx2CanCommit = writeTx2Cohort.canCommit(); - DOMStoreThreePhaseCommitCohort writeTx3Cohort = writeTx3.ready(); - DOMStoreThreePhaseCommitCohort writeTx4Cohort = writeTx4.ready(); - DOMStoreThreePhaseCommitCohort rwTxCohort = readWriteTx.ready(); + final DOMStoreThreePhaseCommitCohort writeTx3Cohort = writeTx3.ready(); + final DOMStoreThreePhaseCommitCohort writeTx4Cohort = writeTx4.ready(); + final DOMStoreThreePhaseCommitCohort rwTxCohort = readWriteTx.ready(); // Enable elections on the other follower so it becomes the leader, at which point the // tx's should get forwarded from the previous leader to the new leader to complete the commits. - sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder. - customRaftPolicyImplementation(null).shardElectionTimeoutFactor(1)); + sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder + .customRaftPolicyImplementation(null).shardElectionTimeoutFactor(1)); followerTestKit.doCommit(writeTx1CanCommit, writeTx1Cohort); followerTestKit.doCommit(writeTx2CanCommit, writeTx2Cohort); @@ -807,8 +806,9 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { String testName = "testLeadershipTransferOnShutdown"; initDatastores(testName, MODULE_SHARDS_CARS_PEOPLE_1_2_3, CARS_AND_PEOPLE); - IntegrationTestKit follower2TestKit = new IntegrationTestKit(follower2System, followerDatastoreContextBuilder); - try (DistributedDataStore follower2DistributedDataStore = follower2TestKit.setupDistributedDataStore(testName, + IntegrationTestKit follower2TestKit = new IntegrationTestKit(follower2System, + DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build()).operationTimeoutInMillis(100)); + try (AbstractDataStore follower2DistributedDataStore = follower2TestKit.setupDistributedDataStore(testName, MODULE_SHARDS_CARS_PEOPLE_1_2_3, false)) { // Create and submit a couple tx's so they're pending. @@ -817,31 +817,23 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()); writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()); writeTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer()); - DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready(); + final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready(); - IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars", new ShardStatsVerifier() { - @Override - public void verify(ShardStats stats) { - assertEquals("getTxCohortCacheSize", 1, stats.getTxCohortCacheSize()); - } - }); + IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars", + stats -> assertEquals("getTxCohortCacheSize", 1, stats.getTxCohortCacheSize())); writeTx = followerDistributedDataStore.newWriteOnlyTransaction(); MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000)); writeTx.write(CarsModel.newCarPath("optima"), car); - DOMStoreThreePhaseCommitCohort cohort2 = writeTx.ready(); + final DOMStoreThreePhaseCommitCohort cohort2 = writeTx.ready(); - IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars", new ShardStatsVerifier() { - @Override - public void verify(ShardStats stats) { - assertEquals("getTxCohortCacheSize", 2, stats.getTxCohortCacheSize()); - } - }); + IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars", + stats -> assertEquals("getTxCohortCacheSize", 2, stats.getTxCohortCacheSize())); // Gracefully stop the leader via a Shutdown message. - sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder. - shardElectionTimeoutFactor(100)); + sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder + .shardElectionTimeoutFactor(100)); FiniteDuration duration = FiniteDuration.create(5, TimeUnit.SECONDS); Future future = leaderDistributedDataStore.getActorContext().findLocalShardAsync("cars"); @@ -867,48 +859,61 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { } @Test - public void testTransactionWithIsolatedLeader() throws Throwable { - leaderDatastoreContextBuilder.shardIsolatedLeaderCheckIntervalInMillis(200); + public void testTransactionWithIsolatedLeader() throws Exception { + // Set the isolated leader check interval high so we can control the switch to IsolatedLeader. + leaderDatastoreContextBuilder.shardIsolatedLeaderCheckIntervalInMillis(10000000); String testName = "testTransactionWithIsolatedLeader"; initDatastoresWithCars(testName); - DOMStoreWriteTransaction failWriteTx = leaderDistributedDataStore.newWriteOnlyTransaction(); - failWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()); + // Tx that is submitted after the follower is stopped but before the leader transitions to IsolatedLeader. + DOMStoreWriteTransaction preIsolatedLeaderWriteTx = leaderDistributedDataStore.newWriteOnlyTransaction(); + preIsolatedLeaderWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()); + // Tx that is submitted after the leader transitions to IsolatedLeader. + DOMStoreWriteTransaction noShardLeaderWriteTx = leaderDistributedDataStore.newWriteOnlyTransaction(); + noShardLeaderWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()); + + // Tx that is submitted after the follower is reinstated. DOMStoreWriteTransaction successWriteTx = leaderDistributedDataStore.newWriteOnlyTransaction(); successWriteTx.merge(CarsModel.BASE_PATH, CarsModel.emptyContainer()); + // Stop the follower followerTestKit.watch(followerDistributedDataStore.getActorContext().getShardManager()); followerDistributedDataStore.close(); followerTestKit.expectTerminated(followerDistributedDataStore.getActorContext().getShardManager()); - MemberNode.verifyRaftState(leaderDistributedDataStore, "cars", new RaftStateVerifier() { - @Override - public void verify(OnDemandRaftState raftState) { - assertEquals("getRaftState", "IsolatedLeader", raftState.getRaftState()); - } - }); + // Submit the preIsolatedLeaderWriteTx so it's pending + final DOMStoreThreePhaseCommitCohort preIsolatedLeaderTxCohort = preIsolatedLeaderWriteTx.ready(); + + // Change the isolated leader check interval low so it changes to IsolatedLeader. + sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder + .shardIsolatedLeaderCheckIntervalInMillis(200)); + + MemberNode.verifyRaftState(leaderDistributedDataStore, "cars", + raftState -> assertEquals("getRaftState", "IsolatedLeader", raftState.getRaftState())); try { - leaderTestKit.doCommit(failWriteTx.ready()); + leaderTestKit.doCommit(noShardLeaderWriteTx.ready()); fail("Expected NoShardLeaderException"); } catch (ExecutionException e) { assertEquals("getCause", NoShardLeaderException.class, e.getCause().getClass()); } - sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder. - shardElectionTimeoutFactor(100)); + sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder + .shardElectionTimeoutFactor(100)); - DOMStoreThreePhaseCommitCohort writeTxCohort = successWriteTx.ready(); + DOMStoreThreePhaseCommitCohort successTxCohort = successWriteTx.ready(); followerDistributedDataStore = followerTestKit.setupDistributedDataStore(testName, MODULE_SHARDS_CARS_ONLY_1_2, false, CARS); - leaderTestKit.doCommit(writeTxCohort); + leaderTestKit.doCommit(preIsolatedLeaderTxCohort); + leaderTestKit.doCommit(successTxCohort); } - @Test(expected=AskTimeoutException.class) - public void testTransactionWithShardLeaderNotResponding() throws Throwable { + @Test(expected = AskTimeoutException.class) + public void testTransactionWithShardLeaderNotResponding() throws Exception { + followerDatastoreContextBuilder.shardElectionTimeoutFactor(50); initDatastoresWithCars("testTransactionWithShardLeaderNotResponding"); // Do an initial read to get the primary shard info cached. @@ -933,12 +938,13 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { assertTrue("Expected ShardLeaderNotRespondingException cause. Actual: " + e.getCause(), e.getCause() instanceof ShardLeaderNotRespondingException); assertNotNull("Expected a nested cause", e.getCause().getCause()); - throw e.getCause().getCause(); + Throwables.propagateIfInstanceOf(e.getCause().getCause(), Exception.class); + Throwables.propagate(e.getCause().getCause()); } } - @Test(expected=NoShardLeaderException.class) - public void testTransactionWithCreateTxFailureDueToNoLeader() throws Throwable { + @Test(expected = NoShardLeaderException.class) + public void testTransactionWithCreateTxFailureDueToNoLeader() throws Exception { initDatastoresWithCars("testTransactionWithCreateTxFailureDueToNoLeader"); // Do an initial read to get the primary shard info cached. @@ -950,10 +956,12 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { JavaTestKit.shutdownActorSystem(leaderSystem, null, true); + Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS); + Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS); - sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder. - operationTimeoutInMillis(10).shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null)); + sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder + .operationTimeoutInMillis(10).shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null)); DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction(); @@ -962,7 +970,8 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { try { followerTestKit.doCommit(rwTx.ready()); } catch (ExecutionException e) { - throw e.getCause(); + Throwables.propagateIfInstanceOf(e.getCause(), Exception.class); + Throwables.propagate(e.getCause()); } } @@ -971,11 +980,11 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { String testName = "testTransactionRetryWithInitialAskTimeoutExOnCreateTx"; initDatastores(testName, MODULE_SHARDS_CARS_PEOPLE_1_2_3, CARS); - DatastoreContext.Builder follower2DatastoreContextBuilder = DatastoreContext.newBuilder(). - shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5); + DatastoreContext.Builder follower2DatastoreContextBuilder = DatastoreContext.newBuilder() + .shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5); IntegrationTestKit follower2TestKit = new IntegrationTestKit(follower2System, follower2DatastoreContextBuilder); - try (DistributedDataStore ds = + try (AbstractDataStore ds = follower2TestKit.setupDistributedDataStore(testName, MODULE_SHARDS_CARS_PEOPLE_1_2_3, false, CARS)) { followerTestKit.waitForMembersUp("member-1", "member-3"); @@ -990,8 +999,10 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { JavaTestKit.shutdownActorSystem(leaderSystem, null, true); - sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder. - operationTimeoutInMillis(500).shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null)); + Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS); + + sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder + .operationTimeoutInMillis(500).shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null)); DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction(); @@ -1001,15 +1012,10 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { } } - private static void sendDatastoreContextUpdate(DistributedDataStore dataStore, final Builder builder) { + private static void sendDatastoreContextUpdate(final AbstractDataStore dataStore, final Builder builder) { final Builder newBuilder = DatastoreContext.newBuilderFrom(builder.build()); DatastoreContextFactory mockContextFactory = Mockito.mock(DatastoreContextFactory.class); - Answer answer = new Answer() { - @Override - public DatastoreContext answer(InvocationOnMock invocation) { - return newBuilder.build(); - } - }; + Answer answer = invocation -> newBuilder.build(); Mockito.doAnswer(answer).when(mockContextFactory).getBaseDatastoreContext(); Mockito.doAnswer(answer).when(mockContextFactory).getShardDatastoreContext(Mockito.anyString()); dataStore.onDatastoreContextUpdated(mockContextFactory);