import akka.util.Timeout;
import com.google.common.base.Function;
import com.google.common.base.Optional;
+import com.google.common.base.Stopwatch;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import org.apache.commons.lang3.SerializationUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
public class ShardManagerTest extends AbstractActorTest {
+ private static final Logger LOG = LoggerFactory.getLogger(ShardManagerTest.class);
+
private static int ID_COUNTER = 1;
private final String shardMrgIDSuffix = "config" + ID_COUNTER++;
private static String mockShardName;
private final DatastoreContext.Builder datastoreContextBuilder = DatastoreContext.newBuilder().
- dataStoreType(shardMrgIDSuffix).shardInitializationTimeout(600, TimeUnit.MILLISECONDS)
+ dataStoreName(shardMrgIDSuffix).shardInitializationTimeout(600, TimeUnit.MILLISECONDS)
.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(6);
private final Collection<ActorSystem> actorSystems = new ArrayList<>();
return shardManager;
}
+ private void waitForShardInitialized(ActorRef shardManager, String shardName, JavaTestKit kit) {
+ AssertionError last = null;
+ Stopwatch sw = Stopwatch.createStarted();
+ while(sw.elapsed(TimeUnit.SECONDS) <= 5) {
+ try {
+ shardManager.tell(new FindLocalShard(shardName, true), kit.getRef());
+ kit.expectMsgClass(LocalShardFound.class);
+ return;
+ } catch(AssertionError e) {
+ last = e;
+ }
+
+ Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
+ }
+
+ throw last;
+ }
+
+ private <T> T expectMsgClassOrFailure(Class<T> msgClass, JavaTestKit kit, String msg) {
+ Object reply = kit.expectMsgAnyClassOf(JavaTestKit.duration("5 sec"), msgClass, Failure.class);
+ if(reply instanceof Failure) {
+ throw new AssertionError(msg + " failed", ((Failure)reply).cause());
+ }
+
+ return (T)reply;
+ }
+
@Test
public void testPerShardDatastoreContext() throws Exception {
+ LOG.info("testPerShardDatastoreContext starting");
final DatastoreContextFactory mockFactory = newDatastoreContextFactory(
datastoreContextBuilder.shardElectionTimeoutFactor(5).build());
newContext = MessageCollectorActor.expectFirstMatching(topologyShardActor, DatastoreContext.class);
assertEquals("getShardElectionTimeoutFactor", 77, newContext.getShardElectionTimeoutFactor());
+
+ LOG.info("testPerShardDatastoreContext ending");
}
@Test
@Test
public void testOnReceiveFindPrimaryForLocalLeaderShard() throws Exception {
+ LOG.info("testOnReceiveFindPrimaryForLocalLeaderShard starting");
new JavaTestKit(getSystem()) {{
String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
primaryFound.getPrimaryPath().contains("member-1-shard-default"));
assertSame("getLocalShardDataTree", mockDataTree, primaryFound.getLocalShardDataTree());
}};
+
+ LOG.info("testOnReceiveFindPrimaryForLocalLeaderShard ending");
}
@Test
public void testOnReceiveFindPrimaryForNonLocalLeaderShardBeforeMemberUp() throws Exception {
+ LOG.info("testOnReceiveFindPrimaryForNonLocalLeaderShardBeforeMemberUp starting");
new JavaTestKit(getSystem()) {{
final ActorRef shardManager = actorFactory.createActor(newPropsShardMgrWithMockShardActor());
expectMsgClass(duration("5 seconds"), NoShardLeaderException.class);
}};
+
+ LOG.info("testOnReceiveFindPrimaryForNonLocalLeaderShardBeforeMemberUp ending");
}
@Test
public void testOnReceiveFindPrimaryForNonLocalLeaderShard() throws Exception {
+ LOG.info("testOnReceiveFindPrimaryForNonLocalLeaderShard starting");
new JavaTestKit(getSystem()) {{
final ActorRef shardManager = actorFactory.createActor(newPropsShardMgrWithMockShardActor());
primaryFound.getPrimaryPath().contains("member-2-shard-default"));
assertEquals("getPrimaryVersion", leaderVersion, primaryFound.getPrimaryVersion());
}};
+
+ LOG.info("testOnReceiveFindPrimaryForNonLocalLeaderShard ending");
}
@Test
@Test
public void testOnReceiveFindPrimaryForFollowerShardWithNoInitialLeaderId() throws Exception {
+ LOG.info("testOnReceiveFindPrimaryForFollowerShardWithNoInitialLeaderId starting");
new JavaTestKit(getSystem()) {{
final ActorRef shardManager = actorFactory.createActor(newPropsShardMgrWithMockShardActor());
primaryFound.getPrimaryPath().contains("member-1-shard-default"));
assertSame("getLocalShardDataTree", mockDataTree, primaryFound.getLocalShardDataTree());
}};
+
+ LOG.info("testOnReceiveFindPrimaryForFollowerShardWithNoInitialLeaderId starting");
}
@Test
public void testOnReceiveFindPrimaryWaitForShardLeader() throws Exception {
+ LOG.info("testOnReceiveFindPrimaryWaitForShardLeader starting");
new JavaTestKit(getSystem()) {{
final ActorRef shardManager = actorFactory.createActor(newPropsShardMgrWithMockShardActor());
expectNoMsg(FiniteDuration.create(200, TimeUnit.MILLISECONDS));
}};
+
+ LOG.info("testOnReceiveFindPrimaryWaitForShardLeader ending");
}
@Test
public void testOnReceiveFindPrimaryWaitForReadyWithUninitializedShard() throws Exception {
+ LOG.info("testOnReceiveFindPrimaryWaitForReadyWithUninitializedShard starting");
new JavaTestKit(getSystem()) {{
final ActorRef shardManager = actorFactory.createActor(newPropsShardMgrWithMockShardActor());
expectNoMsg(FiniteDuration.create(200, TimeUnit.MILLISECONDS));
}};
+
+ LOG.info("testOnReceiveFindPrimaryWaitForReadyWithUninitializedShard ending");
}
@Test
public void testOnReceiveFindPrimaryWaitForReadyWithCandidateShard() throws Exception {
+ LOG.info("testOnReceiveFindPrimaryWaitForReadyWithCandidateShard starting");
new JavaTestKit(getSystem()) {{
final ActorRef shardManager = actorFactory.createActor(newPropsShardMgrWithMockShardActor());
expectMsgClass(duration("2 seconds"), NoShardLeaderException.class);
}};
+
+ LOG.info("testOnReceiveFindPrimaryWaitForReadyWithCandidateShard ending");
}
@Test
public void testOnReceiveFindPrimaryWaitForReadyWithIsolatedLeaderShard() throws Exception {
+ LOG.info("testOnReceiveFindPrimaryWaitForReadyWithIsolatedLeaderShard starting");
new JavaTestKit(getSystem()) {{
final ActorRef shardManager = actorFactory.createActor(newPropsShardMgrWithMockShardActor());
expectMsgClass(duration("2 seconds"), NoShardLeaderException.class);
}};
+
+ LOG.info("testOnReceiveFindPrimaryWaitForReadyWithIsolatedLeaderShard ending");
}
@Test
public void testOnReceiveFindPrimaryWaitForReadyWithNoRoleShard() throws Exception {
+ LOG.info("testOnReceiveFindPrimaryWaitForReadyWithNoRoleShard starting");
new JavaTestKit(getSystem()) {{
final ActorRef shardManager = actorFactory.createActor(newPropsShardMgrWithMockShardActor());
expectMsgClass(duration("2 seconds"), NoShardLeaderException.class);
}};
+
+ LOG.info("testOnReceiveFindPrimaryWaitForReadyWithNoRoleShard ending");
}
@Test
public void testOnReceiveFindPrimaryForRemoteShard() throws Exception {
+ LOG.info("testOnReceiveFindPrimaryForRemoteShard starting");
String shardManagerID = ShardManagerIdentifier.builder().type(shardMrgIDSuffix).build().toString();
// Create an ActorSystem ShardManager actor for member-1.
expectMsgClass(duration("5 seconds"), PrimaryNotFoundException.class);
}};
+
+ LOG.info("testOnReceiveFindPrimaryForRemoteShard ending");
}
@Test
public void testShardAvailabilityOnChangeOfMemberReachability() throws Exception {
+ LOG.info("testShardAvailabilityOnChangeOfMemberReachability starting");
String shardManagerID = ShardManagerIdentifier.builder().type(shardMrgIDSuffix).build().toString();
// Create an ActorSystem ShardManager actor for member-1.
MessageCollectorActor.expectFirstMatching(mockShardActor1, PeerUp.class);
}};
+
+ LOG.info("testShardAvailabilityOnChangeOfMemberReachability ending");
}
@Test
public void testShardAvailabilityChangeOnMemberUnreachableAndLeadershipChange() throws Exception {
+ LOG.info("testShardAvailabilityChangeOnMemberUnreachableAndLeadershipChange starting");
String shardManagerID = ShardManagerIdentifier.builder().type(shardMrgIDSuffix).build().toString();
// Create an ActorSystem ShardManager actor for member-1.
assertTrue("Unexpected primary path " + path1, path1.contains("member-1-shard-default-config"));
}};
+
+ LOG.info("testShardAvailabilityChangeOnMemberUnreachableAndLeadershipChange ending");
}
@Test
public void testOnReceiveFindLocalShardWaitForShardInitialized() throws Exception {
+ LOG.info("testOnReceiveFindLocalShardWaitForShardInitialized starting");
new JavaTestKit(getSystem()) {{
final ActorRef shardManager = actorFactory.createActor(newPropsShardMgrWithMockShardActor());
Object resp = Await.result(future, duration("5 seconds"));
assertTrue("Expected: LocalShardFound, Actual: " + resp, resp instanceof LocalShardFound);
}};
+
+ LOG.info("testOnReceiveFindLocalShardWaitForShardInitialized starting");
}
@Test
@Test
public void testWhenMultipleShardsPresentSyncStatusMustBeTrueForAllShards() throws Exception{
+ LOG.info("testWhenMultipleShardsPresentSyncStatusMustBeTrueForAllShards starting");
TestShardManager shardManager = newTestShardManager(newShardMgrProps(new MockConfiguration() {
@Override
public List<String> getMemberShardNames(String memberName) {
// Sync status is now true
assertEquals(true, shardManager.getMBean().getSyncStatus());
+ LOG.info("testWhenMultipleShardsPresentSyncStatusMustBeTrueForAllShards ending");
}
@Test
@Test
public void testOnCreateShard() {
+ LOG.info("testOnCreateShard starting");
new JavaTestKit(getSystem()) {{
datastoreContextBuilder.shardInitializationTimeout(1, TimeUnit.MINUTES).persistent(true);
Success success = expectMsgClass(duration("5 seconds"), Success.class);
assertNotNull("Success status is null", success.status());
}};
+
+ LOG.info("testOnCreateShard ending");
}
@Test
public void testOnCreateShardWithLocalMemberNotInShardConfig() {
+ LOG.info("testOnCreateShardWithLocalMemberNotInShardConfig starting");
new JavaTestKit(getSystem()) {{
datastoreContextBuilder.shardInitializationTimeout(1, TimeUnit.MINUTES).persistent(true);
assertEquals("schemaContext", DisableElectionsRaftPolicy.class.getName(),
shardBuilder.getDatastoreContext().getShardRaftConfig().getCustomRaftPolicyImplementationClass());
}};
+
+ LOG.info("testOnCreateShardWithLocalMemberNotInShardConfig ending");
}
@Test
public void testOnCreateShardWithNoInitialSchemaContext() {
+ LOG.info("testOnCreateShardWithNoInitialSchemaContext starting");
new JavaTestKit(getSystem()) {{
ActorRef shardManager = actorFactory.createActor(newShardMgrProps(
new ConfigurationImpl(new EmptyModuleShardConfigProvider())));
assertSame("schemaContext", schemaContext, shardBuilder.getSchemaContext());
assertNotNull("schemaContext is null", shardBuilder.getDatastoreContext());
}};
+
+ LOG.info("testOnCreateShardWithNoInitialSchemaContext ending");
}
@Test
public void testGetSnapshot() throws Throwable {
+ LOG.info("testGetSnapshot starting");
JavaTestKit kit = new JavaTestKit(getSystem());
MockConfiguration mockConfig = new MockConfiguration(ImmutableMap.<String, List<String>>builder().
put("shard1", Arrays.asList("member-1")).
- put("shard2", Arrays.asList("member-1")).build());
+ put("shard2", Arrays.asList("member-1")).
+ put("astronauts", Collections.<String>emptyList()).build());
- ActorRef shardManager = actorFactory.createActor(newShardMgrProps(mockConfig).withDispatcher(
- Dispatchers.DefaultDispatcherId()));
+ TestActorRef<TestShardManager> shardManager = actorFactory.createTestActor(newShardMgrProps(mockConfig).
+ withDispatcher(Dispatchers.DefaultDispatcherId()));
shardManager.tell(GetSnapshot.INSTANCE, kit.getRef());
Failure failure = kit.expectMsgClass(Failure.class);
assertEquals("Failure cause type", IllegalStateException.class, failure.cause().getClass());
- kit = new JavaTestKit(getSystem());
-
shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), ActorRef.noSender());
- shardManager.tell(new FindLocalShard("shard1", true), kit.getRef());
- kit.expectMsgClass(LocalShardFound.class);
- shardManager.tell(new FindLocalShard("shard2", true), kit.getRef());
- kit.expectMsgClass(LocalShardFound.class);
+ waitForShardInitialized(shardManager, "shard1", kit);
+ waitForShardInitialized(shardManager, "shard2", kit);
shardManager.tell(GetSnapshot.INSTANCE, kit.getRef());
- DatastoreSnapshot datastoreSnapshot = kit.expectMsgClass(DatastoreSnapshot.class);
+ DatastoreSnapshot datastoreSnapshot = expectMsgClassOrFailure(DatastoreSnapshot.class, kit, "GetSnapshot");
assertEquals("getType", shardMrgIDSuffix, datastoreSnapshot.getType());
assertNull("Expected null ShardManagerSnapshot", datastoreSnapshot.getShardManagerSnapshot());
assertEquals("Shard names", Sets.newHashSet("shard1", "shard2"), Sets.newHashSet(
Lists.transform(datastoreSnapshot.getShardSnapshots(), shardNameTransformer)));
+
+ // Add a new replica
+
+ JavaTestKit mockShardLeaderKit = new JavaTestKit(getSystem());
+
+ TestShardManager shardManagerInstance = shardManager.underlyingActor();
+ shardManagerInstance.setMessageInterceptor(newFindPrimaryInterceptor(mockShardLeaderKit.getRef()));
+
+ shardManager.tell(new AddShardReplica("astronauts"), kit.getRef());
+ mockShardLeaderKit.expectMsgClass(AddServer.class);
+ mockShardLeaderKit.reply(new AddServerReply(ServerChangeStatus.OK, ""));
+ kit.expectMsgClass(Status.Success.class);
+ waitForShardInitialized(shardManager, "astronauts", kit);
+
+ // Send another GetSnapshot and verify
+
+ shardManager.tell(GetSnapshot.INSTANCE, kit.getRef());
+ datastoreSnapshot = expectMsgClassOrFailure(DatastoreSnapshot.class, kit, "GetSnapshot");
+
+ assertEquals("Shard names", Sets.newHashSet("shard1", "shard2", "astronauts"), Sets.newHashSet(
+ Lists.transform(datastoreSnapshot.getShardSnapshots(), shardNameTransformer)));
+
+ byte[] snapshotBytes = datastoreSnapshot.getShardManagerSnapshot();
+ assertNotNull("Expected ShardManagerSnapshot", snapshotBytes);
+ ShardManagerSnapshot snapshot = SerializationUtils.deserialize(snapshotBytes);
+ assertEquals("Shard names", Sets.newHashSet("shard1", "shard2", "astronauts"),
+ Sets.newHashSet(snapshot.getShardList()));
+
+ LOG.info("testGetSnapshot ending");
+ }
+
+ @Test
+ public void testRestoreFromSnapshot() throws Throwable {
+ LOG.info("testRestoreFromSnapshot starting");
+
+ JavaTestKit kit = new JavaTestKit(getSystem());
+
+ MockConfiguration mockConfig = new MockConfiguration(ImmutableMap.<String, List<String>>builder().
+ put("shard1", Collections.<String>emptyList()).
+ put("shard2", Collections.<String>emptyList()).
+ put("astronauts", Collections.<String>emptyList()).build());
+
+
+ ShardManagerSnapshot snapshot = new ShardManagerSnapshot(Arrays.asList("shard1", "shard2", "astronauts"));
+ DatastoreSnapshot restoreFromSnapshot = new DatastoreSnapshot(shardMrgIDSuffix,
+ SerializationUtils.serialize(snapshot), Collections.<ShardSnapshot>emptyList());
+ TestActorRef<TestShardManager> shardManager = actorFactory.createTestActor(newTestShardMgrBuilder(mockConfig).
+ restoreFromSnapshot(restoreFromSnapshot).props().withDispatcher(Dispatchers.DefaultDispatcherId()));
+
+ shardManager.underlyingActor().waitForRecoveryComplete();
+
+ shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), ActorRef.noSender());
+
+ waitForShardInitialized(shardManager, "shard1", kit);
+ waitForShardInitialized(shardManager, "shard2", kit);
+ waitForShardInitialized(shardManager, "astronauts", kit);
+
+ shardManager.tell(GetSnapshot.INSTANCE, kit.getRef());
+
+ DatastoreSnapshot datastoreSnapshot = expectMsgClassOrFailure(DatastoreSnapshot.class, kit, "GetSnapshot");
+
+ assertEquals("getType", shardMrgIDSuffix, datastoreSnapshot.getType());
+
+ byte[] snapshotBytes = datastoreSnapshot.getShardManagerSnapshot();
+ assertNotNull("Expected ShardManagerSnapshot", snapshotBytes);
+ snapshot = SerializationUtils.deserialize(snapshotBytes);
+ assertEquals("Shard names", Sets.newHashSet("shard1", "shard2", "astronauts"),
+ Sets.newHashSet(snapshot.getShardList()));
+
+ LOG.info("testRestoreFromSnapshot ending");
}
@Test
put("default", Arrays.asList("member-1", "member-2")).
put("astronauts", Arrays.asList("member-2")).build());
- String shardManagerID = ShardManagerIdentifier.builder().type(shardMrgIDSuffix).build().toString();
+ final String shardManagerID = ShardManagerIdentifier.builder().type(shardMrgIDSuffix).build().toString();
+ datastoreContextBuilder.shardManagerPersistenceId(shardManagerID);
// Create an ActorSystem ShardManager actor for member-1.
final ActorSystem system1 = newActorSystem("Member1");
newReplicaShardManager.underlyingActor().waitForMemberUp();
leaderShardManager.underlyingActor().waitForMemberUp();
+ //Have a dummy snapshot to be overwritten by the new data persisted.
+ String[] restoredShards = {"default", "people"};
+ ShardManagerSnapshot snapshot = new ShardManagerSnapshot(Arrays.asList(restoredShards));
+ InMemorySnapshotStore.addSnapshot(shardManagerID, snapshot);
+ Uninterruptibles.sleepUninterruptibly(2, TimeUnit.MILLISECONDS);
+
+ InMemorySnapshotStore.addSnapshotSavedLatch(shardManagerID);
+ InMemorySnapshotStore.addSnapshotDeletedLatch(shardManagerID);
+
//construct a mock response message
AddServerReply response = new AddServerReply(ServerChangeStatus.OK, memberId2);
mockShardLeaderActor.underlyingActor().updateResponse(response);
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);
+
+ InMemorySnapshotStore.waitForSavedSnapshot(shardManagerID, ShardManagerSnapshot.class);
+ InMemorySnapshotStore.waitForDeletedSnapshot(shardManagerID);
+ List<ShardManagerSnapshot> persistedSnapshots =
+ InMemorySnapshotStore.getSnapshots(shardManagerID, ShardManagerSnapshot.class);
+ assertEquals("Number of snapshots persisted", 1, persistedSnapshots.size());
+ ShardManagerSnapshot shardManagerSnapshot = persistedSnapshots.get(0);
+ assertEquals("Persisted local shards", Sets.newHashSet("default", "astronauts"),
+ Sets.newHashSet(shardManagerSnapshot.getShardList()));
}};
+
+ LOG.info("testAddShardReplica ending");
}
@Test
public void testAddShardReplicaWithPreExistingReplicaInRemoteShardLeader() throws Exception {
+ LOG.info("testAddShardReplicaWithPreExistingReplicaInRemoteShardLeader starting");
new JavaTestKit(getSystem()) {{
TestActorRef<TestShardManager> shardManager = actorFactory.createTestActor(
newPropsShardMgrWithMockShardActor(), shardMgrID);
shardManager.tell(new FindLocalShard(Shard.DEFAULT_NAME, false), getRef());
expectMsgClass(duration("5 seconds"), LocalShardFound.class);
}};
+
+ LOG.info("testAddShardReplicaWithPreExistingReplicaInRemoteShardLeader ending");
}
@Test
public void testAddShardReplicaWithPreExistingLocalReplicaLeader() throws Exception {
+ LOG.info("testAddShardReplicaWithPreExistingLocalReplicaLeader starting");
new JavaTestKit(getSystem()) {{
String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
ActorRef shardManager = actorFactory.createActor(newPropsShardMgrWithMockShardActor());
shardManager.tell(new FindLocalShard(Shard.DEFAULT_NAME, false), getRef());
expectMsgClass(duration("5 seconds"), LocalShardFound.class);
}};
+
+ LOG.info("testAddShardReplicaWithPreExistingLocalReplicaLeader ending");
}
@Test
public void testAddShardReplicaWithAddServerReplyFailure() throws Exception {
+ LOG.info("testAddShardReplicaWithAddServerReplyFailure starting");
new JavaTestKit(getSystem()) {{
JavaTestKit mockShardLeaderKit = new JavaTestKit(getSystem());
failure = expectMsgClass(duration("5 seconds"), Failure.class);
assertEquals("Failure cause", NoShardLeaderException.class, failure.cause().getClass());
}};
+
+ LOG.info("testAddShardReplicaWithAddServerReplyFailure ending");
}
@Test
public void testAddShardReplicaWithAlreadyInProgress() throws Exception {
+ LOG.info("testAddShardReplicaWithAlreadyInProgress starting");
new JavaTestKit(getSystem()) {{
JavaTestKit mockShardLeaderKit = new JavaTestKit(getSystem());
JavaTestKit secondRequestKit = new JavaTestKit(getSystem());
secondRequestKit.expectMsgClass(duration("5 seconds"), Failure.class);
}};
+
+ LOG.info("testAddShardReplicaWithAlreadyInProgress ending");
}
@Test
public void testAddShardReplicaWithFindPrimaryTimeout() throws Exception {
+ LOG.info("testAddShardReplicaWithFindPrimaryTimeout starting");
new JavaTestKit(getSystem()) {{
MockConfiguration mockConfig = new MockConfiguration(ImmutableMap.<String, List<String>>builder().
put("astronauts", Arrays.asList("member-2")).build());
assertEquals("Failure obtained", true,
(resp.cause() instanceof RuntimeException));
}};
+
+ LOG.info("testAddShardReplicaWithFindPrimaryTimeout ending");
}
@Test
@Test
public void testServerRemovedShardActorNotRunning() throws Exception {
+ LOG.info("testServerRemovedShardActorNotRunning starting");
new JavaTestKit(getSystem()) {{
MockConfiguration mockConfig =
new MockConfiguration(ImmutableMap.<String, List<String>>builder().
TestActorRef<TestShardManager> shardManager = actorFactory.createTestActor(newShardMgrProps(mockConfig));
shardManager.underlyingActor().waitForRecoveryComplete();
-
shardManager.tell(new FindLocalShard("people", false), getRef());
expectMsgClass(duration("5 seconds"), NotInitializedException.class);
// Removed the default shard replica from member-1
ShardIdentifier.Builder builder = new ShardIdentifier.Builder();
- final ShardIdentifier shardId = builder.shardName("default").memberName("member-1").type("config1").build();
+ ShardIdentifier shardId = builder.shardName("default").memberName("member-1").type(shardMrgIDSuffix).build();
shardManager.tell(new ServerRemoved(shardId.toString()), getRef());
shardManager.underlyingActor().verifySnapshotPersisted(Sets.newHashSet("people"));
}};
+
+ LOG.info("testServerRemovedShardActorNotRunning ending");
}
@Test
public void testServerRemovedShardActorRunning() throws Exception {
+ LOG.info("testServerRemovedShardActorRunning starting");
new JavaTestKit(getSystem()) {{
MockConfiguration mockConfig =
new MockConfiguration(ImmutableMap.<String, List<String>>builder().
put("astronauts", Arrays.asList("member-2")).
put("people", Arrays.asList("member-1", "member-2")).build());
- TestActorRef<MessageCollectorActor> shard = actorFactory.createTestActor(MessageCollectorActor.props());
+ String shardId = ShardIdentifier.builder().shardName("default").memberName("member-1").
+ type(shardMrgIDSuffix).build().toString();
+ TestActorRef<MessageCollectorActor> shard = actorFactory.createTestActor(
+ MessageCollectorActor.props(), shardId);
TestActorRef<TestShardManager> shardManager = actorFactory.createTestActor(
newTestShardMgrBuilder(mockConfig).addShardActor("default", shard).props());
shardManager.underlyingActor().waitForRecoveryComplete();
shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
+ shardManager.tell(new ActorInitialized(), shard);
- shardManager.tell(new FindLocalShard("people", false), getRef());
- expectMsgClass(duration("5 seconds"), NotInitializedException.class);
-
- shardManager.tell(new FindLocalShard("default", false), getRef());
- expectMsgClass(duration("5 seconds"), NotInitializedException.class);
+ waitForShardInitialized(shardManager, "people", this);
+ waitForShardInitialized(shardManager, "default", this);
// Removed the default shard replica from member-1
- ShardIdentifier.Builder builder = new ShardIdentifier.Builder();
- final ShardIdentifier shardId = builder.shardName("default").memberName("member-1").type("config1").build();
- shardManager.tell(new ServerRemoved(shardId.toString()), getRef());
+ shardManager.tell(new ServerRemoved(shardId), getRef());
shardManager.underlyingActor().verifySnapshotPersisted(Sets.newHashSet("people"));
expectMsgClass(duration("5 seconds"), Terminated.class);
}};
+
+ LOG.info("testServerRemovedShardActorRunning ending");
}
@Test
public void testShardPersistenceWithRestoredData() throws Exception {
+ LOG.info("testShardPersistenceWithRestoredData starting");
new JavaTestKit(getSystem()) {{
MockConfiguration mockConfig =
new MockConfiguration(ImmutableMap.<String, List<String>>builder().
newRestoredShardManager.tell(new FindLocalShard("astronauts", false), getRef());
expectMsgClass(duration("5 seconds"), NotInitializedException.class);
}};
+
+ LOG.info("testShardPersistenceWithRestoredData ending");
}
public void saveSnapshot(Object obj) {
snapshot = (ShardManagerSnapshot) obj;
snapshotPersist.countDown();
+ super.saveSnapshot(obj);
}
void verifySnapshotPersisted(Set<String> shardList) {
Code Review / controller.git / blobdiff