import org.opendaylight.controller.cluster.raft.messages.AddServerReply;
import org.opendaylight.controller.cluster.raft.messages.ServerChangeStatus;
import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
+import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
MockitoAnnotations.initMocks(this);
InMemoryJournal.clear();
+ InMemorySnapshotStore.clear();
if(mockShardActor == null) {
mockShardName = new ShardIdentifier(Shard.DEFAULT_NAME, "member-1", "config").toString();
@After
public void tearDown() {
InMemoryJournal.clear();
+ InMemorySnapshotStore.clear();
}
private Props newShardMgrProps() {
}
private Props newShardMgrProps(Configuration config) {
- return ShardManager.props(new MockClusterWrapper(), config,
- newDatastoreContextFactory(datastoreContextBuilder.build()), ready, primaryShardInfoCache);
+ return TestShardManager.builder(datastoreContextBuilder).configuration(config).props();
}
private Props newPropsShardMgrWithMockShardActor() {
private static final long serialVersionUID = 1L;
@Override
public ShardManager create() throws Exception {
- return new ForwardingShardManager(clusterWrapper, config, newDatastoreContextFactory(
- datastoreContextBuilder.build()), ready, name, shardActor, primaryShardInfoCache);
+ return new ForwardingShardManager(ShardManager.builder().cluster(clusterWrapper).configuration(config).
+ datastoreContextFactory(newDatastoreContextFactory(datastoreContextBuilder.build())).
+ waitTillReadyCountdownLatch(ready).primaryShardInfoCache(primaryShardInfoCache), name, shardActor);
}
};
return Props.create(new DelegatingShardManagerCreator(creator)).withDispatcher(Dispatchers.DefaultDispatcherId());
}
+ private TestShardManager newTestShardManager() {
+ return newTestShardManager(newShardMgrProps());
+ }
+
+ private TestShardManager newTestShardManager(Props props) {
+ TestActorRef<TestShardManager> shardManagerActor = TestActorRef.create(getSystem(), props);
+ TestShardManager shardManager = shardManagerActor.underlyingActor();
+ shardManager.waitForRecoveryComplete();
+ return shardManager;
+ }
+
@Test
public void testPerShardDatastoreContext() throws Exception {
final DatastoreContextFactory mockFactory = newDatastoreContextFactory(
private static final long serialVersionUID = 1L;
@Override
public ShardManager create() throws Exception {
- return new ShardManager(new MockClusterWrapper(), mockConfig, mockFactory, ready, primaryShardInfoCache) {
+ return new ShardManager(ShardManager.builder().cluster(new MockClusterWrapper()).configuration(mockConfig).
+ datastoreContextFactory(mockFactory).waitTillReadyCountdownLatch(ready).
+ primaryShardInfoCache(primaryShardInfoCache)) {
@Override
protected ActorRef newShardActor(SchemaContext schemaContext, ShardInformation info) {
Entry<ActorRef, DatastoreContext> entry = shardInfoMap.get(info.getShardName());
InMemoryJournal.addDeleteMessagesCompleteLatch(shardMgrID);
new JavaTestKit(getSystem()) {{
- TestActorRef<TestShardManager> shardManager = TestActorRef.create(getSystem(),
- Props.create(new TestShardManagerCreator(shardMrgIDSuffix)));
+ TestActorRef<TestShardManager> shardManager = TestActorRef.create(getSystem(), newShardMgrProps());
shardManager.underlyingActor().waitForRecoveryComplete();
InMemoryJournal.waitForDeleteMessagesComplete(shardMgrID);
@Test
public void testRoleChangeNotificationAndShardLeaderStateChangedReleaseReady() throws Exception {
- new JavaTestKit(getSystem()) {
- {
- TestActorRef<ShardManager> shardManager = TestActorRef.create(getSystem(), newShardMgrProps());
+ TestShardManager shardManager = newTestShardManager();
- String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
- shardManager.underlyingActor().onReceiveCommand(new RoleChangeNotification(
- memberId, RaftState.Candidate.name(), RaftState.Leader.name()));
+ String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
+ shardManager.onReceiveCommand(new RoleChangeNotification(
+ memberId, RaftState.Candidate.name(), RaftState.Leader.name()));
- verify(ready, never()).countDown();
+ verify(ready, never()).countDown();
- shardManager.underlyingActor().onReceiveCommand(new ShardLeaderStateChanged(memberId, memberId,
- Optional.of(mock(DataTree.class)), DataStoreVersions.CURRENT_VERSION));
+ shardManager.onReceiveCommand(new ShardLeaderStateChanged(memberId, memberId,
+ Optional.of(mock(DataTree.class)), DataStoreVersions.CURRENT_VERSION));
- verify(ready, times(1)).countDown();
-
- }};
+ verify(ready, times(1)).countDown();
}
@Test
public void testRoleChangeNotificationToFollowerWithShardLeaderStateChangedReleaseReady() throws Exception {
- new JavaTestKit(getSystem()) {
- {
- TestActorRef<ShardManager> shardManager = TestActorRef.create(getSystem(), newShardMgrProps());
-
- String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
- shardManager.underlyingActor().onReceiveCommand(new RoleChangeNotification(
- memberId, null, RaftState.Follower.name()));
+ new JavaTestKit(getSystem()) {{
+ TestShardManager shardManager = newTestShardManager();
- verify(ready, never()).countDown();
+ String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
+ shardManager.onReceiveCommand(new RoleChangeNotification(
+ memberId, null, RaftState.Follower.name()));
- shardManager.underlyingActor().onReceiveCommand(MockClusterWrapper.createMemberUp("member-2", getRef().path().toString()));
+ verify(ready, never()).countDown();
- shardManager.underlyingActor().onReceiveCommand(new ShardLeaderStateChanged(memberId,
- "member-2-shard-default-" + shardMrgIDSuffix, Optional.of(mock(DataTree.class)),
- DataStoreVersions.CURRENT_VERSION));
+ shardManager.onReceiveCommand(MockClusterWrapper.createMemberUp("member-2", getRef().path().toString()));
- verify(ready, times(1)).countDown();
+ shardManager.onReceiveCommand(new ShardLeaderStateChanged(memberId,
+ "member-2-shard-default-" + shardMrgIDSuffix, Optional.of(mock(DataTree.class)),
+ DataStoreVersions.CURRENT_VERSION));
- }};
+ verify(ready, times(1)).countDown();
+ }};
}
@Test
public void testReadyCountDownForMemberUpAfterLeaderStateChanged() throws Exception {
- new JavaTestKit(getSystem()) {
- {
- TestActorRef<ShardManager> shardManager = TestActorRef.create(getSystem(), newShardMgrProps());
-
- String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
- shardManager.underlyingActor().onReceiveCommand(new RoleChangeNotification(
- memberId, null, RaftState.Follower.name()));
+ new JavaTestKit(getSystem()) {{
+ TestShardManager shardManager = newTestShardManager();
- verify(ready, never()).countDown();
+ String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
+ shardManager.onReceiveCommand(new RoleChangeNotification(memberId, null, RaftState.Follower.name()));
- shardManager.underlyingActor().onReceiveCommand(new ShardLeaderStateChanged(memberId,
- "member-2-shard-default-" + shardMrgIDSuffix, Optional.of(mock(DataTree.class)),
- DataStoreVersions.CURRENT_VERSION));
+ verify(ready, never()).countDown();
- shardManager.underlyingActor().onReceiveCommand(MockClusterWrapper.createMemberUp("member-2", getRef().path().toString()));
+ shardManager.onReceiveCommand(new ShardLeaderStateChanged(memberId,
+ "member-2-shard-default-" + shardMrgIDSuffix, Optional.of(mock(DataTree.class)),
+ DataStoreVersions.CURRENT_VERSION));
- verify(ready, times(1)).countDown();
+ shardManager.onReceiveCommand(MockClusterWrapper.createMemberUp("member-2", getRef().path().toString()));
- }};
+ verify(ready, times(1)).countDown();
+ }};
}
@Test
public void testRoleChangeNotificationDoNothingForUnknownShard() throws Exception {
- new JavaTestKit(getSystem()) {
- {
- TestActorRef<ShardManager> shardManager = TestActorRef.create(getSystem(), newShardMgrProps());
-
- shardManager.underlyingActor().onReceiveCommand(new RoleChangeNotification(
- "unknown", RaftState.Candidate.name(), RaftState.Leader.name()));
+ TestShardManager shardManager = newTestShardManager();
- verify(ready, never()).countDown();
+ shardManager.onReceiveCommand(new RoleChangeNotification(
+ "unknown", RaftState.Candidate.name(), RaftState.Leader.name()));
- }};
+ verify(ready, never()).countDown();
}
-
@Test
public void testByDefaultSyncStatusIsFalse() throws Exception{
- final Props persistentProps = newShardMgrProps();
- final TestActorRef<ShardManager> shardManager =
- TestActorRef.create(getSystem(), persistentProps);
-
- ShardManager shardManagerActor = shardManager.underlyingActor();
+ TestShardManager shardManager = newTestShardManager();
- assertEquals(false, shardManagerActor.getMBean().getSyncStatus());
+ assertEquals(false, shardManager.getMBean().getSyncStatus());
}
@Test
public void testWhenShardIsLeaderSyncStatusIsTrue() throws Exception{
- final Props persistentProps = ShardManager.props(new MockClusterWrapper(), new MockConfiguration(),
- newDatastoreContextFactory(DatastoreContext.newBuilder().persistent(true).build()), ready,
- primaryShardInfoCache);
- final TestActorRef<ShardManager> shardManager =
- TestActorRef.create(getSystem(), persistentProps);
-
- ShardManager shardManagerActor = shardManager.underlyingActor();
- shardManagerActor.onReceiveCommand(new RoleChangeNotification("member-1-shard-default-unknown",
+ TestShardManager shardManager = newTestShardManager();
+
+ shardManager.onReceiveCommand(new RoleChangeNotification("member-1-shard-default-" + shardMrgIDSuffix,
RaftState.Follower.name(), RaftState.Leader.name()));
- assertEquals(true, shardManagerActor.getMBean().getSyncStatus());
+ assertEquals(true, shardManager.getMBean().getSyncStatus());
}
@Test
public void testWhenShardIsCandidateSyncStatusIsFalse() throws Exception{
- final Props persistentProps = newShardMgrProps();
- final TestActorRef<ShardManager> shardManager =
- TestActorRef.create(getSystem(), persistentProps);
+ TestShardManager shardManager = newTestShardManager();
- ShardManager shardManagerActor = shardManager.underlyingActor();
- shardManagerActor.onReceiveCommand(new RoleChangeNotification("member-1-shard-default-unknown",
+ String shardId = "member-1-shard-default-" + shardMrgIDSuffix;
+ shardManager.onReceiveCommand(new RoleChangeNotification(shardId,
RaftState.Follower.name(), RaftState.Candidate.name()));
- assertEquals(false, shardManagerActor.getMBean().getSyncStatus());
+ assertEquals(false, shardManager.getMBean().getSyncStatus());
// Send a FollowerInitialSyncStatus with status = true for the replica whose current state is candidate
- shardManagerActor.onReceiveCommand(new FollowerInitialSyncUpStatus(true, "member-1-shard-default-unknown"));
+ shardManager.onReceiveCommand(new FollowerInitialSyncUpStatus(
+ true, shardId));
- assertEquals(false, shardManagerActor.getMBean().getSyncStatus());
+ assertEquals(false, shardManager.getMBean().getSyncStatus());
}
@Test
public void testWhenShardIsFollowerSyncStatusDependsOnFollowerInitialSyncStatus() throws Exception{
- final Props persistentProps = ShardManager.props(new MockClusterWrapper(), new MockConfiguration(),
- newDatastoreContextFactory(DatastoreContext.newBuilder().persistent(true).build()), ready,
- primaryShardInfoCache);
- final TestActorRef<ShardManager> shardManager =
- TestActorRef.create(getSystem(), persistentProps);
-
- ShardManager shardManagerActor = shardManager.underlyingActor();
- shardManagerActor.onReceiveCommand(new RoleChangeNotification("member-1-shard-default-unknown",
+ TestShardManager shardManager = newTestShardManager();
+
+ String shardId = "member-1-shard-default-" + shardMrgIDSuffix;
+ shardManager.onReceiveCommand(new RoleChangeNotification(shardId,
RaftState.Candidate.name(), RaftState.Follower.name()));
// Initially will be false
- assertEquals(false, shardManagerActor.getMBean().getSyncStatus());
+ assertEquals(false, shardManager.getMBean().getSyncStatus());
// Send status true will make sync status true
- shardManagerActor.onReceiveCommand(new FollowerInitialSyncUpStatus(true, "member-1-shard-default-unknown"));
+ shardManager.onReceiveCommand(new FollowerInitialSyncUpStatus(true, shardId));
- assertEquals(true, shardManagerActor.getMBean().getSyncStatus());
+ assertEquals(true, shardManager.getMBean().getSyncStatus());
// Send status false will make sync status false
- shardManagerActor.onReceiveCommand(new FollowerInitialSyncUpStatus(false, "member-1-shard-default-unknown"));
+ shardManager.onReceiveCommand(new FollowerInitialSyncUpStatus(false, shardId));
- assertEquals(false, shardManagerActor.getMBean().getSyncStatus());
+ assertEquals(false, shardManager.getMBean().getSyncStatus());
}
@Test
public void testWhenMultipleShardsPresentSyncStatusMustBeTrueForAllShards() throws Exception{
- final Props persistentProps = ShardManager.props(new MockClusterWrapper(),
- new MockConfiguration() {
- @Override
- public List<String> getMemberShardNames(String memberName) {
- return Arrays.asList("default", "astronauts");
- }
- },
- newDatastoreContextFactory(DatastoreContext.newBuilder().persistent(true).build()), ready,
- primaryShardInfoCache);
- final TestActorRef<ShardManager> shardManager =
- TestActorRef.create(getSystem(), persistentProps);
-
- ShardManager shardManagerActor = shardManager.underlyingActor();
+ TestShardManager shardManager = newTestShardManager(newShardMgrProps(new MockConfiguration() {
+ @Override
+ public List<String> getMemberShardNames(String memberName) {
+ return Arrays.asList("default", "astronauts");
+ }
+ }));
// Initially will be false
- assertEquals(false, shardManagerActor.getMBean().getSyncStatus());
+ assertEquals(false, shardManager.getMBean().getSyncStatus());
// Make default shard leader
- shardManagerActor.onReceiveCommand(new RoleChangeNotification("member-1-shard-default-unknown",
+ String defaultShardId = "member-1-shard-default-" + shardMrgIDSuffix;
+ shardManager.onReceiveCommand(new RoleChangeNotification(defaultShardId,
RaftState.Follower.name(), RaftState.Leader.name()));
// default = Leader, astronauts is unknown so sync status remains false
- assertEquals(false, shardManagerActor.getMBean().getSyncStatus());
+ assertEquals(false, shardManager.getMBean().getSyncStatus());
// Make astronauts shard leader as well
- shardManagerActor.onReceiveCommand(new RoleChangeNotification("member-1-shard-astronauts-unknown",
+ String astronautsShardId = "member-1-shard-astronauts-" + shardMrgIDSuffix;
+ shardManager.onReceiveCommand(new RoleChangeNotification(astronautsShardId,
RaftState.Follower.name(), RaftState.Leader.name()));
// Now sync status should be true
- assertEquals(true, shardManagerActor.getMBean().getSyncStatus());
+ assertEquals(true, shardManager.getMBean().getSyncStatus());
// Make astronauts a Follower
- shardManagerActor.onReceiveCommand(new RoleChangeNotification("member-1-shard-astronauts-unknown",
+ shardManager.onReceiveCommand(new RoleChangeNotification(astronautsShardId,
RaftState.Leader.name(), RaftState.Follower.name()));
// Sync status is not true
- assertEquals(false, shardManagerActor.getMBean().getSyncStatus());
+ assertEquals(false, shardManager.getMBean().getSyncStatus());
// Make the astronauts follower sync status true
- shardManagerActor.onReceiveCommand(new FollowerInitialSyncUpStatus(true, "member-1-shard-astronauts-unknown"));
+ shardManager.onReceiveCommand(new FollowerInitialSyncUpStatus(true, astronautsShardId));
// Sync status is now true
- assertEquals(true, shardManagerActor.getMBean().getSyncStatus());
+ assertEquals(true, shardManager.getMBean().getSyncStatus());
}
AddServer.class);
String addServerId = "member-1-shard-astronauts-" + shardMrgIDSuffix;
assertEquals("AddServer serverId", addServerId, addServerMsg.getNewServerId());
-
+ newReplicaShardManager.underlyingActor()
+ .verifySnapshotPersisted(Sets.newHashSet("default", "astronauts"));
expectMsgClass(duration("5 seconds"), Status.Success.class);
}};
}
+ @Test
+ public void testShardPersistenceWithRestoredData() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ MockConfiguration mockConfig =
+ new MockConfiguration(ImmutableMap.<String, List<String>>builder().
+ put("default", Arrays.asList("member-1", "member-2")).
+ put("astronauts", Arrays.asList("member-2")).
+ put("people", Arrays.asList("member-1", "member-2")).build());
+ String[] restoredShards = {"default", "astronauts"};
+ ShardManagerSnapshot snapshot = new ShardManagerSnapshot(Arrays.asList(restoredShards));
+ InMemorySnapshotStore.addSnapshot(shardMgrID, snapshot);
+
+ //create shardManager to come up with restored data
+ TestActorRef<TestShardManager> newRestoredShardManager = TestActorRef.create(getSystem(),
+ newShardMgrProps(mockConfig));
+
+ newRestoredShardManager.underlyingActor().waitForRecoveryComplete();
+
+ newRestoredShardManager.tell(new FindLocalShard("people", false), getRef());
+ LocalShardNotFound notFound = expectMsgClass(duration("5 seconds"), LocalShardNotFound.class);
+ assertEquals("for uninitialized shard", "people", notFound.getShardName());
+
+ //Verify a local shard is created for the restored shards,
+ //although we expect a NotInitializedException for the shards as the actor initialization
+ //message is not sent for them
+ newRestoredShardManager.tell(new FindLocalShard("default", false), getRef());
+ expectMsgClass(duration("5 seconds"), NotInitializedException.class);
+
+ newRestoredShardManager.tell(new FindLocalShard("astronauts", false), getRef());
+ expectMsgClass(duration("5 seconds"), NotInitializedException.class);
+ }};
+ }
+
+
private static class TestShardManager extends ShardManager {
private final CountDownLatch recoveryComplete = new CountDownLatch(1);
- TestShardManager(String shardMrgIDSuffix) {
- super(new MockClusterWrapper(), new MockConfiguration(),
- newDatastoreContextFactory(DatastoreContext.newBuilder().dataStoreType(shardMrgIDSuffix).build()),
- ready, new PrimaryShardInfoFutureCache());
+ private TestShardManager(Builder builder) {
+ super(builder);
}
@Override
assertEquals("Recovery complete", true,
Uninterruptibles.awaitUninterruptibly(recoveryComplete, 5, TimeUnit.SECONDS));
}
- }
-
- @SuppressWarnings("serial")
- static class TestShardManagerCreator implements Creator<TestShardManager> {
- String shardMrgIDSuffix;
- TestShardManagerCreator(String shardMrgIDSuffix) {
- this.shardMrgIDSuffix = shardMrgIDSuffix;
+ public static Builder builder(DatastoreContext.Builder datastoreContextBuilder) {
+ return new Builder(datastoreContextBuilder);
}
- @Override
- public TestShardManager create() throws Exception {
- return new TestShardManager(shardMrgIDSuffix);
- }
+ private static class Builder extends ShardManager.Builder {
+ Builder(DatastoreContext.Builder datastoreContextBuilder) {
+ cluster(new MockClusterWrapper()).configuration(new MockConfiguration());
+ datastoreContextFactory(newDatastoreContextFactory(datastoreContextBuilder.build()));
+ waitTillReadyCountdownLatch(ready).primaryShardInfoCache(new PrimaryShardInfoFutureCache());
+ }
+ @Override
+ public Props props() {
+ verify();
+ return Props.create(TestShardManager.class, this);
+ }
+ }
}
private static class DelegatingShardManagerCreator implements Creator<ShardManager> {
private CountDownLatch memberReachableReceived = new CountDownLatch(1);
private final ActorRef shardActor;
private final String name;
+ private final CountDownLatch snapshotPersist = new CountDownLatch(1);
+ private ShardManagerSnapshot snapshot;
- protected ForwardingShardManager(ClusterWrapper cluster, Configuration configuration,
- DatastoreContextFactory factory, CountDownLatch waitTillReadyCountdownLatch, String name,
- ActorRef shardActor, PrimaryShardInfoFutureCache primaryShardInfoCache) {
- super(cluster, configuration, factory, waitTillReadyCountdownLatch, primaryShardInfoCache);
+ public ForwardingShardManager(Builder builder, String name, ActorRef shardActor) {
+ super(builder);
this.shardActor = shardActor;
this.name = name;
}
Uninterruptibles.awaitUninterruptibly(findPrimaryMessageReceived, 5, TimeUnit.SECONDS));
findPrimaryMessageReceived = new CountDownLatch(1);
}
+
+ @Override
+ public void saveSnapshot(Object obj) {
+ snapshot = (ShardManagerSnapshot) obj;
+ snapshotPersist.countDown();
+ }
+
+ void verifySnapshotPersisted(Set<String> shardList) {
+ assertEquals("saveSnapshot invoked", true,
+ Uninterruptibles.awaitUninterruptibly(snapshotPersist, 5, TimeUnit.SECONDS));
+ assertEquals("Shard Persisted", shardList, Sets.newHashSet(snapshot.getShardList()));
+ }
}
private static class MockRespondActor extends MessageCollectorActor {