+ sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder
+ .customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName())
+ .shardElectionTimeoutFactor(10));
+
+ leaderTestKit.waitUntilNoLeader(leaderDistributedDataStore.getActorUtils(), "cars");
+
+ // Submit all tx's - the messages should get queued for retry.
+
+ final ListenableFuture<Boolean> writeTx2CanCommit = writeTx2Cohort.canCommit();
+ 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));
+ IntegrationTestKit.findLocalShard(followerDistributedDataStore.getActorUtils(), "cars")
+ .tell(TimeoutNow.INSTANCE, ActorRef.noSender());
+ IntegrationTestKit.findLocalShard(followerDistributedDataStore.getActorUtils(), "people")
+ .tell(TimeoutNow.INSTANCE, ActorRef.noSender());
+
+ followerTestKit.doCommit(writeTx1CanCommit, writeTx1Cohort);
+ followerTestKit.doCommit(writeTx2CanCommit, writeTx2Cohort);
+ followerTestKit.doCommit(writeTx3Cohort);
+ followerTestKit.doCommit(writeTx4Cohort);
+ followerTestKit.doCommit(rwTxCohort);
+
+ // At this point everything is committed and the follower datastore should see 5 transactions, but leader should
+ // only see the initial transactions
+ verifyCarsReadWriteTransactions(leaderDistributedDataStore, 3);
+ verifyCarsReadWriteTransactions(followerDistributedDataStore, 5);
+
+ DOMStoreReadTransaction readTx = leaderDistributedDataStore.newReadOnlyTransaction();
+ verifyCars(readTx, cars.toArray(new MapEntryNode[cars.size()]));
+ verifyNode(readTx, PeopleModel.PERSON_LIST_PATH, people);
+ }
+
+ private static void verifyCarsReadWriteTransactions(final ClientBackedDataStore datastore, final int expected)
+ throws Exception {
+ IntegrationTestKit.verifyShardStats(datastore, "cars",
+ stats -> assertEquals("getReadWriteTransactionCount", expected, stats.getReadWriteTransactionCount()));
+ }
+
+ @Test
+ public void testLeadershipTransferOnShutdown() throws Exception {
+ leaderDatastoreContextBuilder.shardBatchedModificationCount(1);
+ followerDatastoreContextBuilder.shardElectionTimeoutFactor(10).customRaftPolicyImplementation(null);
+ final String testName = "testLeadershipTransferOnShutdown";
+ initDatastores(testName, MODULE_SHARDS_CARS_PEOPLE_1_2_3, CARS_AND_PEOPLE);
+
+ final IntegrationTestKit follower2TestKit = new IntegrationTestKit(follower2System,
+ DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build()).operationTimeoutInMillis(500),
+ commitTimeout);
+ try (var follower2DistributedDataStore = follower2TestKit.setupDataStore(testParameter, testName,
+ MODULE_SHARDS_CARS_PEOPLE_1_2_3, false)) {
+
+ followerTestKit.waitForMembersUp("member-3");
+ follower2TestKit.waitForMembersUp("member-1", "member-2");
+
+ // Create and submit a couple tx's so they're pending.
+
+ DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
+ writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+ writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
+ writeTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
+ final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
+
+ // FIXME: this assertion should be made in an explicit Shard test
+ // IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars",
+ // stats -> assertEquals("getTxCohortCacheSize", 1, stats.getTxCohortCacheSize()));
+
+ writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
+ final MapEntryNode car = CarsModel.newCarEntry("optima", Uint64.valueOf(20000));
+ writeTx.write(CarsModel.newCarPath("optima"), car);
+ final DOMStoreThreePhaseCommitCohort cohort2 = writeTx.ready();
+
+ // FIXME: this assertion should be made in an explicit Shard test
+ // IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars",
+ // stats -> assertEquals("getTxCohortCacheSize", 2, stats.getTxCohortCacheSize()));
+
+ // Gracefully stop the leader via a Shutdown message.
+
+ sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder
+ .shardElectionTimeoutFactor(100));
+
+ final FiniteDuration duration = FiniteDuration.create(5, TimeUnit.SECONDS);
+ final Future<ActorRef> future = leaderDistributedDataStore.getActorUtils().findLocalShardAsync("cars");
+ final ActorRef leaderActor = Await.result(future, duration);
+
+ final Future<Boolean> stopFuture = Patterns.gracefulStop(leaderActor, duration, Shutdown.INSTANCE);
+
+ // Commit the 2 transactions. They should finish and succeed.
+
+ followerTestKit.doCommit(cohort1);
+ followerTestKit.doCommit(cohort2);
+
+ // Wait for the leader actor stopped.
+
+ final Boolean stopped = Await.result(stopFuture, duration);
+ assertEquals("Stopped", Boolean.TRUE, stopped);
+
+ // Verify leadership was transferred by reading the committed data from the other nodes.
+
+ verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car);
+ verifyCars(follower2DistributedDataStore.newReadOnlyTransaction(), car);
+ }
+ }
+
+ @Test
+ public void testTransactionWithIsolatedLeader() throws Exception {
+ // Set the isolated leader check interval high so we can control the switch to IsolatedLeader.
+ leaderDatastoreContextBuilder.shardIsolatedLeaderCheckIntervalInMillis(10000000);
+ final String testName = "testTransactionWithIsolatedLeader";
+ initDatastoresWithCars(testName);
+
+ // Tx that is submitted after the follower is stopped but before the leader transitions to IsolatedLeader.
+ final DOMStoreWriteTransaction preIsolatedLeaderWriteTx = leaderDistributedDataStore.newWriteOnlyTransaction();
+ preIsolatedLeaderWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+
+ // Tx that is submitted after the leader transitions to IsolatedLeader.
+ final DOMStoreWriteTransaction noShardLeaderWriteTx = leaderDistributedDataStore.newWriteOnlyTransaction();
+ noShardLeaderWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+
+ // Tx that is submitted after the follower is reinstated.
+ final DOMStoreWriteTransaction successWriteTx = leaderDistributedDataStore.newWriteOnlyTransaction();
+ successWriteTx.merge(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+
+ // Stop the follower
+ followerTestKit.watch(followerDistributedDataStore.getActorUtils().getShardManager());
+ followerDistributedDataStore.close();
+ followerTestKit.expectTerminated(followerDistributedDataStore.getActorUtils().getShardManager());
+
+ // 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()));
+
+ final var noShardLeaderCohort = noShardLeaderWriteTx.ready();
+ // tell-based canCommit() does not have a real timeout and hence continues
+ final var canCommit = noShardLeaderCohort.canCommit();
+ Uninterruptibles.sleepUninterruptibly(commitTimeout, TimeUnit.SECONDS);
+ assertFalse(canCommit.isDone());
+
+ sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder
+ .shardElectionTimeoutFactor(100));
+
+ final DOMStoreThreePhaseCommitCohort successTxCohort = successWriteTx.ready();
+
+ followerDistributedDataStore = followerTestKit.setupDataStore(testParameter, testName,
+ MODULE_SHARDS_CARS_ONLY_1_2, false, CARS);
+
+ leaderTestKit.doCommit(preIsolatedLeaderTxCohort);
+ leaderTestKit.doCommit(successTxCohort);
+
+ // continuation of canCommit(): readied transaction will complete commit, but will report an OLFE
+ final var ex = assertThrows(ExecutionException.class,
+ () -> canCommit.get(commitTimeout, TimeUnit.SECONDS)).getCause();
+ assertThat(ex, instanceOf(OptimisticLockFailedException.class));
+ assertEquals("Optimistic lock failed for path " + CarsModel.BASE_PATH, ex.getMessage());
+ final var cause = ex.getCause();
+ assertThat(cause, instanceOf(ConflictingModificationAppliedException.class));
+ final var cmae = (ConflictingModificationAppliedException) cause;
+ assertEquals("Node was created by other transaction.", cmae.getMessage());
+ assertEquals(CarsModel.BASE_PATH, cmae.getPath());
+ }
+
+ @Test
+ public void testTransactionWithShardLeaderNotResponding() throws Exception {
+ followerDatastoreContextBuilder.frontendRequestTimeoutInSeconds(2);
+ followerDatastoreContextBuilder.shardElectionTimeoutFactor(50);
+ initDatastoresWithCars("testTransactionWithShardLeaderNotResponding");
+
+ // Do an initial read to get the primary shard info cached.
+
+ final DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
+ readTx.read(CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS);
+
+ // Shutdown the leader and try to create a new tx.
+
+ TestKit.shutdownActorSystem(leaderSystem, true);
+
+ followerDatastoreContextBuilder.operationTimeoutInMillis(50).shardElectionTimeoutFactor(1);
+ sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder);
+
+ final DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
+
+ rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+
+ final var ex = assertThrows(ExecutionException.class, () -> followerTestKit.doCommit(rwTx.ready()));
+ assertThat("Unexpected exception: " + Throwables.getStackTraceAsString(ex.getCause()),
+ Throwables.getRootCause(ex), instanceOf(RequestTimeoutException.class));
+ }
+
+ @Test
+ public void testTransactionWithCreateTxFailureDueToNoLeader() throws Exception {
+ followerDatastoreContextBuilder.frontendRequestTimeoutInSeconds(2);
+ initDatastoresWithCars("testTransactionWithCreateTxFailureDueToNoLeader");
+
+ // Do an initial read to get the primary shard info cached.
+
+ final DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
+ readTx.read(CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS);
+
+ // Shutdown the leader and try to create a new tx.
+
+ TestKit.shutdownActorSystem(leaderSystem, true);
+
+ Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS);
+
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder
+ .operationTimeoutInMillis(10).shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null));
+
+ final DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
+
+ rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+
+ final var ex = assertThrows(ExecutionException.class, () -> followerTestKit.doCommit(rwTx.ready()));
+ assertThat("Unexpected exception: " + Throwables.getStackTraceAsString(ex.getCause()),
+ Throwables.getRootCause(ex), instanceOf(RequestTimeoutException.class));
+ }
+
+ @Test
+ public void testTransactionRetryWithInitialAskTimeoutExOnCreateTx() throws Exception {
+ followerDatastoreContextBuilder.backendAlivenessTimerIntervalInSeconds(2);
+ String testName = "testTransactionRetryWithInitialAskTimeoutExOnCreateTx";
+ 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);
+
+ try (var ds = follower2TestKit.setupDataStore(testParameter, testName, MODULE_SHARDS_CARS_1_2_3, false, CARS)) {
+
+ followerTestKit.waitForMembersUp("member-1", "member-3");
+ follower2TestKit.waitForMembersUp("member-1", "member-2");
+
+ // Do an initial read to get the primary shard info cached.
+
+ final DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
+ readTx.read(CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS);
+
+ // Shutdown the leader and try to create a new tx.
+
+ TestKit.shutdownActorSystem(leaderSystem, true);
+
+ Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS);
+
+ sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder
+ .operationTimeoutInMillis(500).shardElectionTimeoutFactor(5).customRaftPolicyImplementation(null));
+
+ final DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
+
+ rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+
+ followerTestKit.doCommit(rwTx.ready());
+ }
+ }
+
+ @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 var ds2 = follower2TestKit.setupDataStore(testParameter, testName, MODULE_SHARDS_CARS_1_2_3, false, CARS);
+
+ followerTestKit.waitForMembersUp("member-1", "member-3");
+ follower2TestKit.waitForMembersUp("member-1", "member-2");
+
+ // behavior is controlled by akka.coordinated-shutdown.run-by-actor-system-terminate configuration option
+ TestKit.shutdownActorSystem(follower2System, true);
+
+ ActorRef cars = leaderDistributedDataStore.getActorUtils().findLocalShard("cars").orElseThrow();
+ final 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(() -> containsUnreachable(leaderCluster, follower2Member));
+ await().atMost(10, TimeUnit.SECONDS)
+ .until(() -> containsUnreachable(followerCluster, follower2Member));
+
+ ActorRef followerCars = followerDistributedDataStore.getActorUtils().findLocalShard("cars").orElseThrow();
+
+ // 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();
+ }
+
+ private static Boolean containsUnreachable(final Cluster cluster, final Member member) {
+ // unreachableMembers() returns scala.collection.immutable.Set, but we are using scala.collection.Set to fix JDT
+ // see https://bugs.eclipse.org/bugs/show_bug.cgi?id=468276#c32
+ final Set<Member> members = cluster.readView().unreachableMembers();
+ return members.contains(member);
+ }
+
+ @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.of());
+ 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);
+
+ assertEquals(Optional.of(carsNode), leaderDistributedDataStore.newReadOnlyTransaction().read(
+ CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS));
+
+ verifySnapshot(InMemorySnapshotStore.waitForSavedSnapshot(leaderCarShardName, Snapshot.class),
+ initialSnapshot, snapshotRoot);
+
+ verifySnapshot(InMemorySnapshotStore.waitForSavedSnapshot(followerCarShardName, Snapshot.class),
+ initialSnapshot, snapshotRoot);
+ }
+
+ @Test
+ public void testReadWriteMessageSlicing() throws Exception {
+ 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);
+ }
+
+ @SuppressWarnings("IllegalCatch")
+ @Test
+ public void testRaftCallbackDuringLeadershipDrop() throws Exception {
+ final String testName = "testRaftCallbackDuringLeadershipDrop";
+ initDatastores(testName, MODULE_SHARDS_CARS_1_2_3, CARS);
+
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+
+ final IntegrationTestKit follower2TestKit = new IntegrationTestKit(follower2System,
+ DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build()).operationTimeoutInMillis(500)
+ .shardLeaderElectionTimeoutInSeconds(3600),
+ commitTimeout);
+
+ final DOMStoreWriteTransaction initialWriteTx = leaderDistributedDataStore.newWriteOnlyTransaction();
+ initialWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+ leaderTestKit.doCommit(initialWriteTx.ready());
+
+ try (var follower2DistributedDataStore = follower2TestKit.setupDataStore(testParameter, testName,
+ MODULE_SHARDS_CARS_1_2_3, false)) {
+
+ final ActorRef member3Cars = ((LocalShardStore) follower2DistributedDataStore).getLocalShards()
+ .getLocalShards().get("cars").getActor();
+ final ActorRef member2Cars = ((LocalShardStore)followerDistributedDataStore).getLocalShards()
+ .getLocalShards().get("cars").getActor();
+ member2Cars.tell(new StartDropMessages<>(AppendEntries.class), null);
+ member3Cars.tell(new StartDropMessages<>(AppendEntries.class), null);
+
+ final DOMStoreWriteTransaction newTx = leaderDistributedDataStore.newWriteOnlyTransaction();
+ newTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
+ final AtomicBoolean submitDone = new AtomicBoolean(false);
+ executor.submit(() -> {
+ try {
+ leaderTestKit.doCommit(newTx.ready());
+ submitDone.set(true);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ final ActorRef leaderCars = ((LocalShardStore) leaderDistributedDataStore).getLocalShards()
+ .getLocalShards().get("cars").getActor();
+ await().atMost(10, TimeUnit.SECONDS)
+ .until(() -> ((OnDemandRaftState) leaderDistributedDataStore.getActorUtils()
+ .executeOperation(leaderCars, GetOnDemandRaftState.INSTANCE)).getLastIndex() >= 1);
+
+ final OnDemandRaftState raftState = (OnDemandRaftState)leaderDistributedDataStore.getActorUtils()
+ .executeOperation(leaderCars, GetOnDemandRaftState.INSTANCE);
+
+ // Simulate a follower not receiving heartbeats but still being able to send messages ie RequestVote with
+ // new term(switching to candidate after election timeout)
+ leaderCars.tell(new RequestVote(raftState.getCurrentTerm() + 1,
+ "member-3-shard-cars-testRaftCallbackDuringLeadershipDrop", -1,
+ -1), member3Cars);
+
+ member2Cars.tell(new StopDropMessages<>(AppendEntries.class), null);
+ member3Cars.tell(new StopDropMessages<>(AppendEntries.class), null);
+
+ await("Is tx stuck in COMMIT_PENDING")
+ .atMost(10, TimeUnit.SECONDS).untilAtomic(submitDone, equalTo(true));
+
+ }
+
+ executor.shutdownNow();
+ }
+
+ @Test
+ public void testSnapshotOnRootOverwrite() throws Exception {
+ initDatastores("testSnapshotOnRootOverwrite", "module-shards-default-cars-member1-and-2.conf",
+ new String[] {"cars", "default"},
+ leaderDatastoreContextBuilder.snapshotOnRootOverwrite(true),
+ followerDatastoreContextBuilder.snapshotOnRootOverwrite(true));
+
+ leaderTestKit.waitForMembersUp("member-2");
+ final ContainerNode rootNode = Builders.containerBuilder()
+ .withNodeIdentifier(NodeIdentifier.create(SchemaContext.NAME))
+ .withChild(CarsModel.create())
+ .build();
+
+ leaderTestKit.testWriteTransaction(leaderDistributedDataStore, YangInstanceIdentifier.of(), rootNode);
+
+ // FIXME: CONTROLLER-2020: ClientBackedDatastore does not have stable indexes/term,
+ // the snapshot index seems to fluctuate
+ assumeTrue(false);
+ IntegrationTestKit.verifyShardState(leaderDistributedDataStore, "cars",
+ state -> assertEquals(1, state.getSnapshotIndex()));
+
+ IntegrationTestKit.verifyShardState(followerDistributedDataStore, "cars",
+ state -> assertEquals(1, state.getSnapshotIndex()));
+
+ verifySnapshot("member-1-shard-cars-testSnapshotOnRootOverwrite", 1);
+ verifySnapshot("member-2-shard-cars-testSnapshotOnRootOverwrite", 1);
+
+ for (int i = 0; i < 10; i++) {
+ leaderTestKit.testWriteTransaction(leaderDistributedDataStore, CarsModel.newCarPath("car " + i),
+ CarsModel.newCarEntry("car " + i, Uint64.ONE));
+ }
+
+ // fake snapshot causes the snapshotIndex to move
+ IntegrationTestKit.verifyShardState(leaderDistributedDataStore, "cars",
+ state -> assertEquals(10, state.getSnapshotIndex()));
+ IntegrationTestKit.verifyShardState(followerDistributedDataStore, "cars",
+ state -> assertEquals(10, state.getSnapshotIndex()));
+
+ // however the real snapshot still has not changed and was taken at index 1
+ verifySnapshot("member-1-shard-cars-testSnapshotOnRootOverwrite", 1);
+ verifySnapshot("member-2-shard-cars-testSnapshotOnRootOverwrite", 1);
+
+ // root overwrite so expect a snapshot
+ leaderTestKit.testWriteTransaction(leaderDistributedDataStore, YangInstanceIdentifier.of(), rootNode);
+
+ // this was a real snapshot so everything should be in it(1(DisableTrackingPayload) + 1 + 10 + 1)
+ IntegrationTestKit.verifyShardState(leaderDistributedDataStore, "cars",
+ state -> assertEquals(12, state.getSnapshotIndex()));
+ IntegrationTestKit.verifyShardState(followerDistributedDataStore, "cars",
+ state -> assertEquals(12, state.getSnapshotIndex()));
+
+ verifySnapshot("member-1-shard-cars-testSnapshotOnRootOverwrite", 12);
+ verifySnapshot("member-2-shard-cars-testSnapshotOnRootOverwrite", 12);
+ }
+
+ private static void verifySnapshot(final String persistenceId, final long lastAppliedIndex) {
+ await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
+ List<Snapshot> snap = InMemorySnapshotStore.getSnapshots(persistenceId, Snapshot.class);
+ assertEquals(1, snap.size());
+ assertEquals(lastAppliedIndex, snap.get(0).getLastAppliedIndex());
+ }
+ );
+ }
+
+ 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().orElseThrow());
+ }