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;
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;
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.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 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";
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();
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);
}
assertEquals("isPresent", true, optional.isPresent());
CollectionNodeBuilder<MapEntryNode, MapNode> 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(DOMStoreReadTransaction readTx, YangInstanceIdentifier path,
+ NormalizedNode<?, ?> expNode) throws Exception {
Optional<NormalizedNode<?, ?>> optional = readTx.read(path).get(5, TimeUnit.SECONDS);
assertEquals("isPresent", true, optional.isPresent());
assertEquals("Data node", expNode, optional.get());
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 (DistributedDataStore member2Datastore = new IntegrationTestKit(newSystem, leaderDatastoreContextBuilder)
+ .setupDistributedDataStore(testName, "module-shards-member2", true, CARS_AND_PEOPLE)) {
verifyCars(member2Datastore.newReadOnlyTransaction(), car2);
}
}
@Test
- public void testTransactionChainWithMultipleShards() throws Exception{
+ public void testTransactionChainWithMultipleShards() throws Exception {
initDatastoresWithCarsAndPeople("testTransactionChainWithMultipleShards");
DOMStoreTransactionChain txChain = followerDistributedDataStore.createTransactionChain();
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);
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.
// 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);
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 =
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());
}
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());
}
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());
}
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());
}
// Wait for the commit to be replicated to the follower.
MemberNode.verifyRaftState(followerDistributedDataStore, "cars",
- raftState -> assertEquals("getLastApplied", 0, raftState.getLastApplied()));
+ raftState -> assertEquals("getLastApplied", 0, raftState.getLastApplied()));
MemberNode.verifyRaftState(followerDistributedDataStore, "people",
- raftState -> assertEquals("getLastApplied", 0, raftState.getLastApplied()));
+ 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.
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
// 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());
}
readWriteTx.write(CarsModel.newCarPath("car" + carIndex), cars.getLast());
IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars",
- stats -> assertEquals("getReadWriteTransactionCount", 1, stats.getReadWriteTransactionCount()));
+ 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<Boolean> 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);
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",
- stats -> assertEquals("getTxCohortCacheSize", 1, stats.getTxCohortCacheSize()));
+ 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",
- stats -> assertEquals("getTxCohortCacheSize", 2, stats.getTxCohortCacheSize()));
+ 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<ActorRef> future = leaderDistributedDataStore.getActorContext().findLocalShardAsync("cars");
}
@Test
- public void testTransactionWithIsolatedLeader() throws Throwable {
+ 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";
followerTestKit.expectTerminated(followerDistributedDataStore.getActorContext().getShardManager());
// Submit the preIsolatedLeaderWriteTx so it's pending
- DOMStoreThreePhaseCommitCohort preIsolatedLeaderTxCohort = preIsolatedLeaderWriteTx.ready();
+ final DOMStoreThreePhaseCommitCohort preIsolatedLeaderTxCohort = preIsolatedLeaderWriteTx.ready();
// Change the isolated leader check interval low so it changes to IsolatedLeader.
- sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder.
- shardIsolatedLeaderCheckIntervalInMillis(200));
+ sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder
+ .shardIsolatedLeaderCheckIntervalInMillis(200));
MemberNode.verifyRaftState(leaderDistributedDataStore, "cars",
- raftState -> assertEquals("getRaftState", "IsolatedLeader", raftState.getRaftState()));
+ raftState -> assertEquals("getRaftState", "IsolatedLeader", raftState.getRaftState()));
try {
leaderTestKit.doCommit(noShardLeaderWriteTx.ready());
assertEquals("getCause", NoShardLeaderException.class, e.getCause().getClass());
}
- sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder.
- shardElectionTimeoutFactor(100));
+ sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder
+ .shardElectionTimeoutFactor(100));
DOMStoreThreePhaseCommitCohort successTxCohort = successWriteTx.ready();
leaderTestKit.doCommit(successTxCohort);
}
- @Test(expected=AskTimeoutException.class)
- public void testTransactionWithShardLeaderNotResponding() throws Throwable {
+ @Test(expected = AskTimeoutException.class)
+ public void testTransactionWithShardLeaderNotResponding() throws Exception {
initDatastoresWithCars("testTransactionWithShardLeaderNotResponding");
// Do an initial read to get the primary shard info cached.
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.
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();
try {
followerTestKit.doCommit(rwTx.ready());
} catch (ExecutionException e) {
- throw e.getCause();
+ Throwables.propagateIfInstanceOf(e.getCause(), Exception.class);
+ Throwables.propagate(e.getCause());
}
}
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 =
Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS);
- sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder.
- operationTimeoutInMillis(500).shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null));
+ sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder
+ .operationTimeoutInMillis(500).shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null));
DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
Code Review / controller.git / blobdiff