+ @Test
+ public void testSemiReachableCandidateNotDroppingLeader() throws Exception {
+ final String testName = "testSemiReachableCandidateNotDroppingLeader";
+ initDatastores(testName, MODULE_SHARDS_CARS_1_2_3, CARS);
+
+ final DatastoreContext.Builder follower2DatastoreContextBuilder = DatastoreContext.newBuilder()
+ .shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(10);
+ final IntegrationTestKit follower2TestKit = new IntegrationTestKit(
+ follower2System, follower2DatastoreContextBuilder, commitTimeout);
+
+ final AbstractDataStore ds2 =
+ follower2TestKit.setupAbstractDataStore(
+ testParameter, testName, MODULE_SHARDS_CARS_1_2_3, false, CARS);
+
+ followerTestKit.waitForMembersUp("member-1", "member-3");
+ follower2TestKit.waitForMembersUp("member-1", "member-2");
+
+ TestKit.shutdownActorSystem(follower2System);
+
+ ActorRef cars = leaderDistributedDataStore.getActorUtils().findLocalShard("cars").get();
+ OnDemandRaftState initialState = (OnDemandRaftState) leaderDistributedDataStore.getActorUtils()
+ .executeOperation(cars, GetOnDemandRaftState.INSTANCE);
+
+ Cluster leaderCluster = Cluster.get(leaderSystem);
+ Cluster followerCluster = Cluster.get(followerSystem);
+ Cluster follower2Cluster = Cluster.get(follower2System);
+
+ Member follower2Member = follower2Cluster.readView().self();
+
+ await().atMost(10, TimeUnit.SECONDS)
+ .until(() -> leaderCluster.readView().unreachableMembers().contains(follower2Member));
+ await().atMost(10, TimeUnit.SECONDS)
+ .until(() -> followerCluster.readView().unreachableMembers().contains(follower2Member));
+
+ ActorRef followerCars = followerDistributedDataStore.getActorUtils().findLocalShard("cars").get();
+
+ // to simulate a follower not being able to receive messages, but still being able to send messages and becoming
+ // candidate, we can just send a couple of RequestVotes to both leader and follower.
+ cars.tell(new RequestVote(initialState.getCurrentTerm() + 1, "member-3-shard-cars", -1, -1), null);
+ followerCars.tell(new RequestVote(initialState.getCurrentTerm() + 1, "member-3-shard-cars", -1, -1), null);
+ cars.tell(new RequestVote(initialState.getCurrentTerm() + 3, "member-3-shard-cars", -1, -1), null);
+ followerCars.tell(new RequestVote(initialState.getCurrentTerm() + 3, "member-3-shard-cars", -1, -1), null);
+
+ OnDemandRaftState stateAfter = (OnDemandRaftState) leaderDistributedDataStore.getActorUtils()
+ .executeOperation(cars, GetOnDemandRaftState.INSTANCE);
+ OnDemandRaftState followerState = (OnDemandRaftState) followerDistributedDataStore.getActorUtils()
+ .executeOperation(cars, GetOnDemandRaftState.INSTANCE);
+
+ assertEquals(initialState.getCurrentTerm(), stateAfter.getCurrentTerm());
+ assertEquals(initialState.getCurrentTerm(), followerState.getCurrentTerm());
+
+ ds2.close();
+ }
+
+ @Test
+ public void testInstallSnapshot() throws Exception {
+ final String testName = "testInstallSnapshot";
+ final String leaderCarShardName = "member-1-shard-cars-" + testName;
+ final String followerCarShardName = "member-2-shard-cars-" + testName;
+
+ // Setup a saved snapshot on the leader. The follower will startup with no data and the leader should
+ // install a snapshot to sync the follower.
+
+ DataTree tree = new InMemoryDataTreeFactory().create(DataTreeConfiguration.DEFAULT_CONFIGURATION,
+ SchemaContextHelper.full());
+
+ final ContainerNode carsNode = CarsModel.newCarsNode(
+ CarsModel.newCarsMapNode(CarsModel.newCarEntry("optima", Uint64.valueOf(20000))));
+ AbstractShardTest.writeToStore(tree, CarsModel.BASE_PATH, carsNode);
+
+ final NormalizedNode<?, ?> snapshotRoot = AbstractShardTest.readStore(tree, YangInstanceIdentifier.empty());
+ final Snapshot initialSnapshot = Snapshot.create(
+ new ShardSnapshotState(new MetadataShardDataTreeSnapshot(snapshotRoot)),
+ Collections.emptyList(), 5, 1, 5, 1, 1, null, null);
+ InMemorySnapshotStore.addSnapshot(leaderCarShardName, initialSnapshot);
+
+ InMemorySnapshotStore.addSnapshotSavedLatch(leaderCarShardName);
+ InMemorySnapshotStore.addSnapshotSavedLatch(followerCarShardName);
+
+ initDatastoresWithCars(testName);
+
+ final Optional<NormalizedNode<?, ?>> readOptional = leaderDistributedDataStore.newReadOnlyTransaction().read(
+ CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS);
+ assertTrue("isPresent", readOptional.isPresent());
+ assertEquals("Node", carsNode, readOptional.get());
+
+ verifySnapshot(InMemorySnapshotStore.waitForSavedSnapshot(leaderCarShardName, Snapshot.class),
+ initialSnapshot, snapshotRoot);
+
+ verifySnapshot(InMemorySnapshotStore.waitForSavedSnapshot(followerCarShardName, Snapshot.class),
+ initialSnapshot, snapshotRoot);
+ }
+
+ @Test
+ public void testReadWriteMessageSlicing() throws Exception {
+ // The slicing is only implemented for tell-based protocol
+ Assume.assumeTrue(ClientBackedDataStore.class.isAssignableFrom(testParameter));
+
+ leaderDatastoreContextBuilder.maximumMessageSliceSize(100);
+ followerDatastoreContextBuilder.maximumMessageSliceSize(100);
+ initDatastoresWithCars("testLargeReadReplySlicing");
+
+ final DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
+
+ final NormalizedNode<?, ?> carsNode = CarsModel.create();
+ rwTx.write(CarsModel.BASE_PATH, carsNode);
+
+ verifyNode(rwTx, CarsModel.BASE_PATH, carsNode);
+ }
+
+ private static void verifySnapshot(final Snapshot actual, final Snapshot expected,
+ final NormalizedNode<?, ?> expRoot) {
+ assertEquals("Snapshot getLastAppliedTerm", expected.getLastAppliedTerm(), actual.getLastAppliedTerm());
+ assertEquals("Snapshot getLastAppliedIndex", expected.getLastAppliedIndex(), actual.getLastAppliedIndex());
+ assertEquals("Snapshot getLastTerm", expected.getLastTerm(), actual.getLastTerm());
+ assertEquals("Snapshot getLastIndex", expected.getLastIndex(), actual.getLastIndex());
+ assertEquals("Snapshot state type", ShardSnapshotState.class, actual.getState().getClass());
+ MetadataShardDataTreeSnapshot shardSnapshot =
+ (MetadataShardDataTreeSnapshot) ((ShardSnapshotState)actual.getState()).getSnapshot();
+ assertEquals("Snapshot root node", expRoot, shardSnapshot.getRootNode().get());
+ }
+