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;
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;
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;
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);
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 =
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", 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", new RaftStateVerifier() {
- @Override
- public void verify(OnDemandRaftState 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.
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());
}
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<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", 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<ActorRef> future = leaderDistributedDataStore.getActorContext().findLocalShardAsync("cars");
}
@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 {
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.
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();
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 =
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();
private static void sendDatastoreContextUpdate(DistributedDataStore dataStore, final Builder builder) {
final Builder newBuilder = DatastoreContext.newBuilderFrom(builder.build());
DatastoreContextFactory mockContextFactory = Mockito.mock(DatastoreContextFactory.class);
- Answer<DatastoreContext> answer = new Answer<DatastoreContext>() {
- @Override
- public DatastoreContext answer(InvocationOnMock invocation) {
- return newBuilder.build();
- }
- };
+ Answer<DatastoreContext> answer = invocation -> newBuilder.build();
Mockito.doAnswer(answer).when(mockContextFactory).getBaseDatastoreContext();
Mockito.doAnswer(answer).when(mockContextFactory).getShardDatastoreContext(Mockito.anyString());
dataStore.onDatastoreContextUpdated(mockContextFactory);