/*
- * Copyright (c) 2015 Brocade Communications Systems, Inc. and others. All rights reserved.
+ * Copyright (c) 2015, 2017 Brocade Communications Systems, Inc. and others. All rights reserved.
*
* This program and the accompanying materials are made available under the
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
import akka.pattern.Patterns;
import akka.testkit.JavaTestKit;
import com.google.common.base.Optional;
+import com.google.common.base.Stopwatch;
import com.google.common.base.Supplier;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableMap;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedList;
+import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
import org.junit.After;
import org.junit.Assume;
import org.junit.Before;
testParameter, type, moduleShardsConfig, false, shards);
leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), shards);
+
+ leaderTestKit.waitForMembersUp("member-2");
+ followerTestKit.waitForMembersUp("member-1");
}
private static void verifyCars(final DOMStoreReadTransaction readTx, final MapEntryNode... entries)
initDatastoresWithCars(testName);
final String followerCarShardName = "member-2-shard-cars-" + testName;
- InMemoryJournal.addWriteMessagesCompleteLatch(followerCarShardName, 2, ApplyJournalEntries.class);
DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
assertNotNull("newWriteOnlyTransaction returned null", writeTx);
// Re-instate the follower member 2 as a single-node to verify replication and recovery.
- InMemoryJournal.waitForWriteMessagesComplete(followerCarShardName);
+ // The following is a bit tricky. Before we reinstate the follower we need to ensure it has persisted and
+ // applied and all the log entries from the leader. Since we've verified the car data above we know that
+ // all the transactions have been applied on the leader so we first read and capture its lastAppliedIndex.
+ final AtomicLong leaderLastAppliedIndex = new AtomicLong();
+ IntegrationTestKit.verifyShardState(leaderDistributedDataStore, CARS[0],
+ state -> leaderLastAppliedIndex.set(state.getLastApplied()));
+
+ // Now we need to make sure the follower has persisted the leader's lastAppliedIndex via ApplyJournalEntries.
+ // However we don't know exactly how many ApplyJournalEntries messages there will be as it can differ between
+ // the tell-based and ask-based front-ends. For ask-based there will be exactly 2 ApplyJournalEntries but
+ // tell-based persists additional payloads which could be replicated and applied in a batch resulting in
+ // either 2 or 3 ApplyJournalEntries. To handle this we read the follower's persisted ApplyJournalEntries
+ // until we find the one that encompasses the leader's lastAppliedIndex.
+ Stopwatch sw = Stopwatch.createStarted();
+ boolean done = false;
+ while (!done) {
+ final List<ApplyJournalEntries> entries = InMemoryJournal.get(followerCarShardName,
+ ApplyJournalEntries.class);
+ for (ApplyJournalEntries aje: entries) {
+ if (aje.getToIndex() >= leaderLastAppliedIndex.get()) {
+ done = true;
+ break;
+ }
+ }
+
+ assertTrue("Follower did not persist ApplyJournalEntries containing leader's lastAppliedIndex "
+ + leaderLastAppliedIndex + ". Entries persisted: " + entries, sw.elapsed(TimeUnit.SECONDS) <= 5);
+
+ Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
+ }
- JavaTestKit.shutdownActorSystem(leaderSystem, null, true);
- JavaTestKit.shutdownActorSystem(followerSystem, null, true);
+ JavaTestKit.shutdownActorSystem(leaderSystem, null, Boolean.TRUE);
+ JavaTestKit.shutdownActorSystem(followerSystem, null, Boolean.TRUE);
- final ActorSystem newSystem = ActorSystem.create("reinstated-member2", ConfigFactory.load()
- .getConfig("Member2"));
+ final ActorSystem newSystem = newActorSystem("reinstated-member2", "Member2");
- try (AbstractDataStore member2Datastore = new IntegrationTestKit(newSystem, leaderDatastoreContextBuilder,
- commitTimeout)
- .setupAbstractDataStore(testParameter, testName, "module-shards-member2", true, CARS_AND_PEOPLE)) {
+ try (final AbstractDataStore member2Datastore = new IntegrationTestKit(newSystem, leaderDatastoreContextBuilder)
+ .setupAbstractDataStore(testParameter, testName, "module-shards-member2", true, CARS)) {
verifyCars(member2Datastore.newReadOnlyTransaction(), car2);
}
-
- JavaTestKit.shutdownActorSystem(newSystem);
}
@Test
.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5);
IntegrationTestKit newMember1TestKit = new IntegrationTestKit(leaderSystem, newMember1Builder, commitTimeout);
- try (final AbstractDataStore ds =
+ try (AbstractDataStore ds =
newMember1TestKit.setupAbstractDataStore(
testParameter, testName, MODULE_SHARDS_CARS_ONLY_1_2, false, CARS)) {
final IntegrationTestKit follower2TestKit = new IntegrationTestKit(follower2System,
DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build()).operationTimeoutInMillis(100),
commitTimeout);
- try (final AbstractDataStore follower2DistributedDataStore = follower2TestKit.setupAbstractDataStore(
+ try (AbstractDataStore follower2DistributedDataStore = follower2TestKit.setupAbstractDataStore(
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();
@Test
public void testTransactionRetryWithInitialAskTimeoutExOnCreateTx() throws Exception {
- //TODO remove when test passes also for ClientBackedDataStore
- Assume.assumeTrue(testParameter.equals(DistributedDataStore.class));
String testName = "testTransactionRetryWithInitialAskTimeoutExOnCreateTx";
initDatastores(testName, MODULE_SHARDS_CARS_PEOPLE_1_2_3, CARS);
final IntegrationTestKit follower2TestKit = new IntegrationTestKit(
follower2System, follower2DatastoreContextBuilder, commitTimeout);
- try (final AbstractDataStore ds =
+ try (AbstractDataStore ds =
follower2TestKit.setupAbstractDataStore(
testParameter, testName, MODULE_SHARDS_CARS_PEOPLE_1_2_3, false, CARS)) {
Code Review / controller.git / blobdiff