2 * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
9 package org.opendaylight.controller.cluster.datastore;
11 import static org.junit.Assert.assertEquals;
12 import static org.junit.Assert.assertNotNull;
13 import static org.junit.Assert.assertNull;
14 import static org.junit.Assert.assertSame;
15 import static org.junit.Assert.assertTrue;
16 import static org.mockito.Mockito.mock;
17 import static org.mockito.Mockito.never;
18 import static org.mockito.Mockito.times;
19 import static org.mockito.Mockito.verify;
20 import akka.actor.ActorRef;
21 import akka.actor.ActorSystem;
22 import akka.actor.AddressFromURIString;
23 import akka.actor.Props;
24 import akka.actor.Status;
25 import akka.cluster.Cluster;
26 import akka.cluster.ClusterEvent;
27 import akka.dispatch.Dispatchers;
28 import akka.japi.Creator;
29 import akka.pattern.Patterns;
30 import akka.persistence.RecoveryCompleted;
31 import akka.testkit.JavaTestKit;
32 import akka.testkit.TestActorRef;
33 import akka.util.Timeout;
34 import com.google.common.base.Optional;
35 import com.google.common.collect.ImmutableMap;
36 import com.google.common.collect.ImmutableSet;
37 import com.google.common.collect.Sets;
38 import com.google.common.util.concurrent.Uninterruptibles;
39 import com.typesafe.config.ConfigFactory;
41 import java.util.Arrays;
42 import java.util.List;
44 import java.util.concurrent.CountDownLatch;
45 import java.util.concurrent.TimeUnit;
46 import org.junit.After;
47 import org.junit.Before;
48 import org.junit.Test;
49 import org.mockito.Mock;
50 import org.mockito.MockitoAnnotations;
51 import org.opendaylight.controller.cluster.datastore.config.Configuration;
52 import org.opendaylight.controller.cluster.datastore.config.ConfigurationImpl;
53 import org.opendaylight.controller.cluster.datastore.config.EmptyModuleShardConfigProvider;
54 import org.opendaylight.controller.cluster.datastore.config.ModuleShardConfiguration;
55 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
56 import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
57 import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException;
58 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
59 import org.opendaylight.controller.cluster.datastore.identifiers.ShardManagerIdentifier;
60 import org.opendaylight.controller.cluster.datastore.messages.ActorInitialized;
61 import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica;
62 import org.opendaylight.controller.cluster.datastore.messages.CreateShard;
63 import org.opendaylight.controller.cluster.datastore.messages.CreateShardReply;
64 import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
65 import org.opendaylight.controller.cluster.datastore.messages.FindPrimary;
66 import org.opendaylight.controller.cluster.datastore.messages.LocalPrimaryShardFound;
67 import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
68 import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
69 import org.opendaylight.controller.cluster.datastore.messages.PeerDown;
70 import org.opendaylight.controller.cluster.datastore.messages.PeerUp;
71 import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
72 import org.opendaylight.controller.cluster.datastore.messages.RemotePrimaryShardFound;
73 import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
74 import org.opendaylight.controller.cluster.datastore.messages.ShardLeaderStateChanged;
75 import org.opendaylight.controller.cluster.datastore.messages.SwitchShardBehavior;
76 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
77 import org.opendaylight.controller.cluster.datastore.utils.MockClusterWrapper;
78 import org.opendaylight.controller.cluster.datastore.utils.MockConfiguration;
79 import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache;
80 import org.opendaylight.controller.cluster.notifications.LeaderStateChanged;
81 import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener;
82 import org.opendaylight.controller.cluster.notifications.RoleChangeNotification;
83 import org.opendaylight.controller.cluster.raft.RaftState;
84 import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus;
85 import org.opendaylight.controller.cluster.raft.base.messages.SwitchBehavior;
86 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
87 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
88 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
89 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
90 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
91 import scala.concurrent.Await;
92 import scala.concurrent.Future;
93 import scala.concurrent.duration.FiniteDuration;
95 public class ShardManagerTest extends AbstractActorTest {
96 private static int ID_COUNTER = 1;
98 private final String shardMrgIDSuffix = "config" + ID_COUNTER++;
99 private final String shardMgrID = "shard-manager-" + shardMrgIDSuffix;
102 private static CountDownLatch ready;
104 private static TestActorRef<MessageCollectorActor> mockShardActor;
106 private static String mockShardName;
108 private final DatastoreContext.Builder datastoreContextBuilder = DatastoreContext.newBuilder().
109 dataStoreType(shardMrgIDSuffix).shardInitializationTimeout(600, TimeUnit.MILLISECONDS)
110 .shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(6);
112 private static ActorRef newMockShardActor(ActorSystem system, String shardName, String memberName) {
113 String name = new ShardIdentifier(shardName, memberName,"config").toString();
114 return TestActorRef.create(system, Props.create(MessageCollectorActor.class), name);
117 private final PrimaryShardInfoFutureCache primaryShardInfoCache = new PrimaryShardInfoFutureCache();
120 public void setUp() {
121 MockitoAnnotations.initMocks(this);
123 InMemoryJournal.clear();
125 if(mockShardActor == null) {
126 mockShardName = new ShardIdentifier(Shard.DEFAULT_NAME, "member-1", "config").toString();
127 mockShardActor = TestActorRef.create(getSystem(), Props.create(MessageCollectorActor.class), mockShardName);
130 mockShardActor.underlyingActor().clear();
134 public void tearDown() {
135 InMemoryJournal.clear();
138 private Props newShardMgrProps() {
139 return newShardMgrProps(new MockConfiguration());
142 private Props newShardMgrProps(Configuration config) {
143 return ShardManager.props(new MockClusterWrapper(), config, datastoreContextBuilder.build(), ready,
144 primaryShardInfoCache);
147 private Props newPropsShardMgrWithMockShardActor() {
148 return newPropsShardMgrWithMockShardActor("shardManager", mockShardActor, new MockClusterWrapper(),
149 new MockConfiguration());
152 private Props newPropsShardMgrWithMockShardActor(final String name, final ActorRef shardActor,
153 final ClusterWrapper clusterWrapper, final Configuration config) {
154 Creator<ShardManager> creator = new Creator<ShardManager>() {
155 private static final long serialVersionUID = 1L;
157 public ShardManager create() throws Exception {
158 return new ForwardingShardManager(clusterWrapper, config, datastoreContextBuilder.build(),
159 ready, name, shardActor, primaryShardInfoCache);
163 return Props.create(new DelegatingShardManagerCreator(creator)).withDispatcher(Dispatchers.DefaultDispatcherId());
167 public void testOnReceiveFindPrimaryForNonExistentShard() throws Exception {
168 new JavaTestKit(getSystem()) {{
169 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
171 shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
173 shardManager.tell(new FindPrimary("non-existent", false), getRef());
175 expectMsgClass(duration("5 seconds"), PrimaryNotFoundException.class);
180 public void testOnReceiveFindPrimaryForLocalLeaderShard() throws Exception {
181 new JavaTestKit(getSystem()) {{
182 String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
184 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
186 shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
187 shardManager.tell(new ActorInitialized(), mockShardActor);
189 DataTree mockDataTree = mock(DataTree.class);
190 shardManager.tell(new ShardLeaderStateChanged(memberId, memberId, Optional.of(mockDataTree),
191 DataStoreVersions.CURRENT_VERSION), getRef());
193 MessageCollectorActor.expectFirstMatching(mockShardActor, RegisterRoleChangeListener.class);
194 shardManager.tell((new RoleChangeNotification(memberId, RaftState.Candidate.name(),
195 RaftState.Leader.name())), mockShardActor);
197 shardManager.tell(new FindPrimary(Shard.DEFAULT_NAME, false), getRef());
199 LocalPrimaryShardFound primaryFound = expectMsgClass(duration("5 seconds"), LocalPrimaryShardFound.class);
200 assertTrue("Unexpected primary path " + primaryFound.getPrimaryPath(),
201 primaryFound.getPrimaryPath().contains("member-1-shard-default"));
202 assertSame("getLocalShardDataTree", mockDataTree, primaryFound.getLocalShardDataTree() );
207 public void testOnReceiveFindPrimaryForNonLocalLeaderShardBeforeMemberUp() throws Exception {
208 new JavaTestKit(getSystem()) {{
209 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
211 shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
212 shardManager.tell(new ActorInitialized(), mockShardActor);
214 String memberId2 = "member-2-shard-default-" + shardMrgIDSuffix;
215 String memberId1 = "member-1-shard-default-" + shardMrgIDSuffix;
216 shardManager.tell(new RoleChangeNotification(memberId1,
217 RaftState.Candidate.name(), RaftState.Follower.name()), mockShardActor);
218 shardManager.tell(new LeaderStateChanged(memberId1, memberId2, DataStoreVersions.CURRENT_VERSION), mockShardActor);
220 shardManager.tell(new FindPrimary(Shard.DEFAULT_NAME, false), getRef());
222 expectMsgClass(duration("5 seconds"), NoShardLeaderException.class);
227 public void testOnReceiveFindPrimaryForNonLocalLeaderShard() throws Exception {
228 new JavaTestKit(getSystem()) {{
229 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
231 shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
232 shardManager.tell(new ActorInitialized(), mockShardActor);
234 String memberId2 = "member-2-shard-default-" + shardMrgIDSuffix;
235 MockClusterWrapper.sendMemberUp(shardManager, "member-2", getRef().path().toString());
237 String memberId1 = "member-1-shard-default-" + shardMrgIDSuffix;
238 shardManager.tell(new RoleChangeNotification(memberId1,
239 RaftState.Candidate.name(), RaftState.Follower.name()), mockShardActor);
240 short leaderVersion = DataStoreVersions.CURRENT_VERSION - 1;
241 shardManager.tell(new ShardLeaderStateChanged(memberId1, memberId2, Optional.<DataTree>absent(),
242 leaderVersion), mockShardActor);
244 shardManager.tell(new FindPrimary(Shard.DEFAULT_NAME, false), getRef());
246 RemotePrimaryShardFound primaryFound = expectMsgClass(duration("5 seconds"), RemotePrimaryShardFound.class);
247 assertTrue("Unexpected primary path " + primaryFound.getPrimaryPath(),
248 primaryFound.getPrimaryPath().contains("member-2-shard-default"));
249 assertEquals("getPrimaryVersion", leaderVersion, primaryFound.getPrimaryVersion());
254 public void testOnReceiveFindPrimaryForUninitializedShard() throws Exception {
255 new JavaTestKit(getSystem()) {{
256 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
258 shardManager.tell(new FindPrimary(Shard.DEFAULT_NAME, false), getRef());
260 expectMsgClass(duration("5 seconds"), NotInitializedException.class);
265 public void testOnReceiveFindPrimaryForInitializedShardWithNoRole() throws Exception {
266 new JavaTestKit(getSystem()) {{
267 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
269 shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
270 shardManager.tell(new ActorInitialized(), mockShardActor);
272 shardManager.tell(new FindPrimary(Shard.DEFAULT_NAME, false), getRef());
274 expectMsgClass(duration("5 seconds"), NoShardLeaderException.class);
279 public void testOnReceiveFindPrimaryForFollowerShardWithNoInitialLeaderId() throws Exception {
280 new JavaTestKit(getSystem()) {{
281 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
283 shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
284 shardManager.tell(new ActorInitialized(), mockShardActor);
286 String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
287 shardManager.tell(new RoleChangeNotification(memberId,
288 RaftState.Candidate.name(), RaftState.Follower.name()), mockShardActor);
290 shardManager.tell(new FindPrimary(Shard.DEFAULT_NAME, false), getRef());
292 expectMsgClass(duration("5 seconds"), NoShardLeaderException.class);
294 DataTree mockDataTree = mock(DataTree.class);
295 shardManager.tell(new ShardLeaderStateChanged(memberId, memberId, Optional.of(mockDataTree),
296 DataStoreVersions.CURRENT_VERSION), mockShardActor);
298 shardManager.tell(new FindPrimary(Shard.DEFAULT_NAME, false), getRef());
300 LocalPrimaryShardFound primaryFound = expectMsgClass(duration("5 seconds"), LocalPrimaryShardFound.class);
301 assertTrue("Unexpected primary path " + primaryFound.getPrimaryPath(),
302 primaryFound.getPrimaryPath().contains("member-1-shard-default"));
303 assertSame("getLocalShardDataTree", mockDataTree, primaryFound.getLocalShardDataTree() );
308 public void testOnReceiveFindPrimaryWaitForShardLeader() throws Exception {
309 new JavaTestKit(getSystem()) {{
310 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
312 shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
314 // We're passing waitUntilInitialized = true to FindPrimary so the response should be
315 // delayed until we send ActorInitialized and RoleChangeNotification.
316 shardManager.tell(new FindPrimary(Shard.DEFAULT_NAME, true), getRef());
318 expectNoMsg(FiniteDuration.create(150, TimeUnit.MILLISECONDS));
320 shardManager.tell(new ActorInitialized(), mockShardActor);
322 expectNoMsg(FiniteDuration.create(150, TimeUnit.MILLISECONDS));
324 String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
325 shardManager.tell(new RoleChangeNotification(memberId,
326 RaftState.Candidate.name(), RaftState.Leader.name()), mockShardActor);
328 expectNoMsg(FiniteDuration.create(150, TimeUnit.MILLISECONDS));
330 DataTree mockDataTree = mock(DataTree.class);
331 shardManager.tell(new ShardLeaderStateChanged(memberId, memberId, Optional.of(mockDataTree),
332 DataStoreVersions.CURRENT_VERSION), mockShardActor);
334 LocalPrimaryShardFound primaryFound = expectMsgClass(duration("5 seconds"), LocalPrimaryShardFound.class);
335 assertTrue("Unexpected primary path " + primaryFound.getPrimaryPath(),
336 primaryFound.getPrimaryPath().contains("member-1-shard-default"));
337 assertSame("getLocalShardDataTree", mockDataTree, primaryFound.getLocalShardDataTree() );
339 expectNoMsg(FiniteDuration.create(200, TimeUnit.MILLISECONDS));
344 public void testOnReceiveFindPrimaryWaitForReadyWithUninitializedShard() throws Exception {
345 new JavaTestKit(getSystem()) {{
346 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
348 shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
350 shardManager.tell(new FindPrimary(Shard.DEFAULT_NAME, true), getRef());
352 expectMsgClass(duration("2 seconds"), NotInitializedException.class);
354 shardManager.tell(new ActorInitialized(), mockShardActor);
356 expectNoMsg(FiniteDuration.create(200, TimeUnit.MILLISECONDS));
361 public void testOnReceiveFindPrimaryWaitForReadyWithCandidateShard() throws Exception {
362 new JavaTestKit(getSystem()) {{
363 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
365 shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
366 shardManager.tell(new ActorInitialized(), mockShardActor);
367 shardManager.tell(new RoleChangeNotification("member-1-shard-default-" + shardMrgIDSuffix,
368 null, RaftState.Candidate.name()), mockShardActor);
370 shardManager.tell(new FindPrimary(Shard.DEFAULT_NAME, true), getRef());
372 expectMsgClass(duration("2 seconds"), NoShardLeaderException.class);
377 public void testOnReceiveFindPrimaryWaitForReadyWithIsolatedLeaderShard() throws Exception {
378 new JavaTestKit(getSystem()) {{
379 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
381 shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
382 shardManager.tell(new ActorInitialized(), mockShardActor);
383 shardManager.tell(new RoleChangeNotification("member-1-shard-default-" + shardMrgIDSuffix,
384 null, RaftState.IsolatedLeader.name()), mockShardActor);
386 shardManager.tell(new FindPrimary(Shard.DEFAULT_NAME, true), getRef());
388 expectMsgClass(duration("2 seconds"), NoShardLeaderException.class);
393 public void testOnReceiveFindPrimaryWaitForReadyWithNoRoleShard() throws Exception {
394 new JavaTestKit(getSystem()) {{
395 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
397 shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
398 shardManager.tell(new ActorInitialized(), mockShardActor);
400 shardManager.tell(new FindPrimary(Shard.DEFAULT_NAME, true), getRef());
402 expectMsgClass(duration("2 seconds"), NoShardLeaderException.class);
407 public void testOnReceiveFindPrimaryForRemoteShard() throws Exception {
408 String shardManagerID = ShardManagerIdentifier.builder().type(shardMrgIDSuffix).build().toString();
410 // Create an ActorSystem ShardManager actor for member-1.
412 final ActorSystem system1 = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
413 Cluster.get(system1).join(AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558"));
415 ActorRef mockShardActor1 = newMockShardActor(system1, Shard.DEFAULT_NAME, "member-1");
417 final TestActorRef<ForwardingShardManager> shardManager1 = TestActorRef.create(system1,
418 newPropsShardMgrWithMockShardActor("shardManager1", mockShardActor1, new ClusterWrapperImpl(system1),
419 new MockConfiguration()), shardManagerID);
421 // Create an ActorSystem ShardManager actor for member-2.
423 final ActorSystem system2 = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member2"));
425 Cluster.get(system2).join(AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558"));
427 final ActorRef mockShardActor2 = newMockShardActor(system2, "astronauts", "member-2");
429 MockConfiguration mockConfig2 = new MockConfiguration(ImmutableMap.<String, List<String>>builder().
430 put("default", Arrays.asList("member-1", "member-2")).
431 put("astronauts", Arrays.asList("member-2")).build());
433 final TestActorRef<ForwardingShardManager> shardManager2 = TestActorRef.create(system2,
434 newPropsShardMgrWithMockShardActor("shardManager2", mockShardActor2, new ClusterWrapperImpl(system2),
435 mockConfig2), shardManagerID);
437 new JavaTestKit(system1) {{
439 shardManager1.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
440 shardManager2.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
442 shardManager2.tell(new ActorInitialized(), mockShardActor2);
444 String memberId2 = "member-2-shard-astronauts-" + shardMrgIDSuffix;
445 short leaderVersion = DataStoreVersions.CURRENT_VERSION - 1;
446 shardManager2.tell(new ShardLeaderStateChanged(memberId2, memberId2,
447 Optional.of(mock(DataTree.class)), leaderVersion), mockShardActor2);
448 shardManager2.tell(new RoleChangeNotification(memberId2,
449 RaftState.Candidate.name(), RaftState.Leader.name()), mockShardActor2);
451 shardManager1.underlyingActor().waitForMemberUp();
453 shardManager1.tell(new FindPrimary("astronauts", false), getRef());
455 RemotePrimaryShardFound found = expectMsgClass(duration("5 seconds"), RemotePrimaryShardFound.class);
456 String path = found.getPrimaryPath();
457 assertTrue("Unexpected primary path " + path, path.contains("member-2-shard-astronauts-config"));
458 assertEquals("getPrimaryVersion", leaderVersion, found.getPrimaryVersion());
460 shardManager2.underlyingActor().verifyFindPrimary();
462 Cluster.get(system2).down(AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558"));
464 shardManager1.underlyingActor().waitForMemberRemoved();
466 shardManager1.tell(new FindPrimary("astronauts", false), getRef());
468 expectMsgClass(duration("5 seconds"), PrimaryNotFoundException.class);
471 JavaTestKit.shutdownActorSystem(system1);
472 JavaTestKit.shutdownActorSystem(system2);
476 public void testShardAvailabilityOnChangeOfMemberReachability() throws Exception {
477 String shardManagerID = ShardManagerIdentifier.builder().type(shardMrgIDSuffix).build().toString();
479 // Create an ActorSystem ShardManager actor for member-1.
481 final ActorSystem system1 = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
482 Cluster.get(system1).join(AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558"));
484 final ActorRef mockShardActor1 = newMockShardActor(system1, Shard.DEFAULT_NAME, "member-1");
486 final TestActorRef<ForwardingShardManager> shardManager1 = TestActorRef.create(system1,
487 newPropsShardMgrWithMockShardActor("shardManager1", mockShardActor1, new ClusterWrapperImpl(system1),
488 new MockConfiguration()), shardManagerID);
490 // Create an ActorSystem ShardManager actor for member-2.
492 final ActorSystem system2 = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member2"));
494 Cluster.get(system2).join(AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558"));
496 final ActorRef mockShardActor2 = newMockShardActor(system2, Shard.DEFAULT_NAME, "member-2");
498 MockConfiguration mockConfig2 = new MockConfiguration(ImmutableMap.<String, List<String>>builder().
499 put("default", Arrays.asList("member-1", "member-2")).build());
501 final TestActorRef<ForwardingShardManager> shardManager2 = TestActorRef.create(system2,
502 newPropsShardMgrWithMockShardActor("shardManager2", mockShardActor2, new ClusterWrapperImpl(system2),
503 mockConfig2), shardManagerID);
505 new JavaTestKit(system1) {{
507 shardManager1.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
508 shardManager2.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
509 shardManager1.tell(new ActorInitialized(), mockShardActor1);
510 shardManager2.tell(new ActorInitialized(), mockShardActor2);
512 String memberId2 = "member-2-shard-default-" + shardMrgIDSuffix;
513 String memberId1 = "member-1-shard-default-" + shardMrgIDSuffix;
514 shardManager1.tell(new ShardLeaderStateChanged(memberId1, memberId2,
515 Optional.of(mock(DataTree.class)), DataStoreVersions.CURRENT_VERSION), mockShardActor1);
516 shardManager1.tell(new RoleChangeNotification(memberId1,
517 RaftState.Candidate.name(), RaftState.Follower.name()), mockShardActor1);
518 shardManager2.tell(new ShardLeaderStateChanged(memberId2, memberId2, Optional.of(mock(DataTree.class)),
519 DataStoreVersions.CURRENT_VERSION),
521 shardManager2.tell(new RoleChangeNotification(memberId2,
522 RaftState.Candidate.name(), RaftState.Leader.name()), mockShardActor2);
523 shardManager1.underlyingActor().waitForMemberUp();
525 shardManager1.tell(new FindPrimary("default", true), getRef());
527 RemotePrimaryShardFound found = expectMsgClass(duration("5 seconds"), RemotePrimaryShardFound.class);
528 String path = found.getPrimaryPath();
529 assertTrue("Unexpected primary path " + path, path.contains("member-2-shard-default-config"));
531 shardManager1.underlyingActor().onReceiveCommand(MockClusterWrapper.
532 createUnreachableMember("member-2", "akka.tcp://cluster-test@127.0.0.1:2558"));
534 shardManager1.underlyingActor().waitForUnreachableMember();
536 PeerDown peerDown = MessageCollectorActor.expectFirstMatching(mockShardActor1, PeerDown.class);
537 assertEquals("getMemberName", "member-2", peerDown.getMemberName());
538 MessageCollectorActor.clearMessages(mockShardActor1);
540 shardManager1.underlyingActor().onReceiveCommand(MockClusterWrapper.
541 createMemberRemoved("member-2", "akka.tcp://cluster-test@127.0.0.1:2558"));
543 MessageCollectorActor.expectFirstMatching(mockShardActor1, PeerDown.class);
545 shardManager1.tell(new FindPrimary("default", true), getRef());
547 expectMsgClass(duration("5 seconds"), NoShardLeaderException.class);
549 shardManager1.underlyingActor().onReceiveCommand(MockClusterWrapper.
550 createReachableMember("member-2", "akka.tcp://cluster-test@127.0.0.1:2558"));
552 shardManager1.underlyingActor().waitForReachableMember();
554 PeerUp peerUp = MessageCollectorActor.expectFirstMatching(mockShardActor1, PeerUp.class);
555 assertEquals("getMemberName", "member-2", peerUp.getMemberName());
556 MessageCollectorActor.clearMessages(mockShardActor1);
558 shardManager1.tell(new FindPrimary("default", true), getRef());
560 RemotePrimaryShardFound found1 = expectMsgClass(duration("5 seconds"), RemotePrimaryShardFound.class);
561 String path1 = found1.getPrimaryPath();
562 assertTrue("Unexpected primary path " + path1, path1.contains("member-2-shard-default-config"));
564 shardManager1.underlyingActor().onReceiveCommand(MockClusterWrapper.
565 createMemberUp("member-2", "akka.tcp://cluster-test@127.0.0.1:2558"));
567 MessageCollectorActor.expectFirstMatching(mockShardActor1, PeerUp.class);
571 JavaTestKit.shutdownActorSystem(system1);
572 JavaTestKit.shutdownActorSystem(system2);
576 public void testShardAvailabilityChangeOnMemberUnreachableAndLeadershipChange() throws Exception {
577 String shardManagerID = ShardManagerIdentifier.builder().type(shardMrgIDSuffix).build().toString();
579 // Create an ActorSystem ShardManager actor for member-1.
581 final ActorSystem system1 = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
582 Cluster.get(system1).join(AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558"));
584 final ActorRef mockShardActor1 = newMockShardActor(system1, Shard.DEFAULT_NAME, "member-1");
586 final TestActorRef<ForwardingShardManager> shardManager1 = TestActorRef.create(system1,
587 newPropsShardMgrWithMockShardActor("shardManager1", mockShardActor1, new ClusterWrapperImpl(system1),
588 new MockConfiguration()), shardManagerID);
590 // Create an ActorSystem ShardManager actor for member-2.
592 final ActorSystem system2 = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member2"));
594 Cluster.get(system2).join(AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558"));
596 final ActorRef mockShardActor2 = newMockShardActor(system2, Shard.DEFAULT_NAME, "member-2");
598 MockConfiguration mockConfig2 = new MockConfiguration(ImmutableMap.<String, List<String>>builder().
599 put("default", Arrays.asList("member-1", "member-2")).build());
601 final TestActorRef<ForwardingShardManager> shardManager2 = TestActorRef.create(system2,
602 newPropsShardMgrWithMockShardActor("shardManager2", mockShardActor2, new ClusterWrapperImpl(system2),
603 mockConfig2), shardManagerID);
605 new JavaTestKit(system1) {{
607 shardManager1.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
608 shardManager2.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
609 shardManager1.tell(new ActorInitialized(), mockShardActor1);
610 shardManager2.tell(new ActorInitialized(), mockShardActor2);
612 String memberId2 = "member-2-shard-default-" + shardMrgIDSuffix;
613 String memberId1 = "member-1-shard-default-" + shardMrgIDSuffix;
614 shardManager1.tell(new ShardLeaderStateChanged(memberId1, memberId2,
615 Optional.of(mock(DataTree.class)), DataStoreVersions.CURRENT_VERSION), mockShardActor1);
616 shardManager1.tell(new RoleChangeNotification(memberId1,
617 RaftState.Candidate.name(), RaftState.Follower.name()), mockShardActor1);
618 shardManager2.tell(new ShardLeaderStateChanged(memberId2, memberId2, Optional.of(mock(DataTree.class)),
619 DataStoreVersions.CURRENT_VERSION),
621 shardManager2.tell(new RoleChangeNotification(memberId2,
622 RaftState.Candidate.name(), RaftState.Leader.name()), mockShardActor2);
623 shardManager1.underlyingActor().waitForMemberUp();
625 shardManager1.tell(new FindPrimary("default", true), getRef());
627 RemotePrimaryShardFound found = expectMsgClass(duration("5 seconds"), RemotePrimaryShardFound.class);
628 String path = found.getPrimaryPath();
629 assertTrue("Unexpected primary path " + path, path.contains("member-2-shard-default-config"));
631 primaryShardInfoCache.putSuccessful("default", new PrimaryShardInfo(system1.actorSelection(
632 mockShardActor1.path()), DataStoreVersions.CURRENT_VERSION, Optional.<DataTree>absent()));
634 shardManager1.underlyingActor().onReceiveCommand(MockClusterWrapper.
635 createUnreachableMember("member-2", "akka.tcp://cluster-test@127.0.0.1:2558"));
637 shardManager1.underlyingActor().waitForUnreachableMember();
639 shardManager1.tell(new FindPrimary("default", true), getRef());
641 expectMsgClass(duration("5 seconds"), NoShardLeaderException.class);
643 assertNull("Expected primaryShardInfoCache entry removed", primaryShardInfoCache.getIfPresent("default"));
645 shardManager1.tell(new ShardLeaderStateChanged(memberId1, memberId1, Optional.of(mock(DataTree.class)),
646 DataStoreVersions.CURRENT_VERSION), mockShardActor1);
647 shardManager1.tell(new RoleChangeNotification(memberId1,
648 RaftState.Follower.name(), RaftState.Leader.name()), mockShardActor1);
650 shardManager1.tell(new FindPrimary("default", true), getRef());
652 LocalPrimaryShardFound found1 = expectMsgClass(duration("5 seconds"), LocalPrimaryShardFound.class);
653 String path1 = found1.getPrimaryPath();
654 assertTrue("Unexpected primary path " + path1, path1.contains("member-1-shard-default-config"));
658 JavaTestKit.shutdownActorSystem(system1);
659 JavaTestKit.shutdownActorSystem(system2);
664 public void testOnReceiveFindLocalShardForNonExistentShard() throws Exception {
665 new JavaTestKit(getSystem()) {{
666 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
668 shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
670 shardManager.tell(new FindLocalShard("non-existent", false), getRef());
672 LocalShardNotFound notFound = expectMsgClass(duration("5 seconds"), LocalShardNotFound.class);
674 assertEquals("getShardName", "non-existent", notFound.getShardName());
679 public void testOnReceiveFindLocalShardForExistentShard() throws Exception {
680 new JavaTestKit(getSystem()) {{
681 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
683 shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
684 shardManager.tell(new ActorInitialized(), mockShardActor);
686 shardManager.tell(new FindLocalShard(Shard.DEFAULT_NAME, false), getRef());
688 LocalShardFound found = expectMsgClass(duration("5 seconds"), LocalShardFound.class);
690 assertTrue("Found path contains " + found.getPath().path().toString(),
691 found.getPath().path().toString().contains("member-1-shard-default-config"));
696 public void testOnReceiveFindLocalShardForNotInitializedShard() throws Exception {
697 new JavaTestKit(getSystem()) {{
698 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
700 shardManager.tell(new FindLocalShard(Shard.DEFAULT_NAME, false), getRef());
702 expectMsgClass(duration("5 seconds"), NotInitializedException.class);
707 public void testOnReceiveFindLocalShardWaitForShardInitialized() throws Exception {
708 new JavaTestKit(getSystem()) {{
709 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
711 shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
713 // We're passing waitUntilInitialized = true to FindLocalShard so the response should be
714 // delayed until we send ActorInitialized.
715 Future<Object> future = Patterns.ask(shardManager, new FindLocalShard(Shard.DEFAULT_NAME, true),
716 new Timeout(5, TimeUnit.SECONDS));
718 shardManager.tell(new ActorInitialized(), mockShardActor);
720 Object resp = Await.result(future, duration("5 seconds"));
721 assertTrue("Expected: LocalShardFound, Actual: " + resp, resp instanceof LocalShardFound);
726 public void testOnRecoveryJournalIsCleaned() {
727 InMemoryJournal.addEntry(shardMgrID, 1L, new ShardManager.SchemaContextModules(
728 ImmutableSet.of("foo")));
729 InMemoryJournal.addEntry(shardMgrID, 2L, new ShardManager.SchemaContextModules(
730 ImmutableSet.of("bar")));
731 InMemoryJournal.addDeleteMessagesCompleteLatch(shardMgrID);
733 new JavaTestKit(getSystem()) {{
734 TestActorRef<TestShardManager> shardManager = TestActorRef.create(getSystem(),
735 Props.create(new TestShardManagerCreator(shardMrgIDSuffix)));
737 shardManager.underlyingActor().waitForRecoveryComplete();
738 InMemoryJournal.waitForDeleteMessagesComplete(shardMgrID);
740 // Journal entries up to the last one should've been deleted
741 Map<Long, Object> journal = InMemoryJournal.get(shardMgrID);
742 synchronized (journal) {
743 assertEquals("Journal size", 0, journal.size());
749 public void testRoleChangeNotificationAndShardLeaderStateChangedReleaseReady() throws Exception {
750 new JavaTestKit(getSystem()) {
752 TestActorRef<ShardManager> shardManager = TestActorRef.create(getSystem(), newShardMgrProps());
754 String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
755 shardManager.underlyingActor().onReceiveCommand(new RoleChangeNotification(
756 memberId, RaftState.Candidate.name(), RaftState.Leader.name()));
758 verify(ready, never()).countDown();
760 shardManager.underlyingActor().onReceiveCommand(new ShardLeaderStateChanged(memberId, memberId,
761 Optional.of(mock(DataTree.class)), DataStoreVersions.CURRENT_VERSION));
763 verify(ready, times(1)).countDown();
769 public void testRoleChangeNotificationToFollowerWithShardLeaderStateChangedReleaseReady() throws Exception {
770 new JavaTestKit(getSystem()) {
772 TestActorRef<ShardManager> shardManager = TestActorRef.create(getSystem(), newShardMgrProps());
774 String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
775 shardManager.underlyingActor().onReceiveCommand(new RoleChangeNotification(
776 memberId, null, RaftState.Follower.name()));
778 verify(ready, never()).countDown();
780 shardManager.underlyingActor().onReceiveCommand(MockClusterWrapper.createMemberUp("member-2", getRef().path().toString()));
782 shardManager.underlyingActor().onReceiveCommand(new ShardLeaderStateChanged(memberId,
783 "member-2-shard-default-" + shardMrgIDSuffix, Optional.of(mock(DataTree.class)),
784 DataStoreVersions.CURRENT_VERSION));
786 verify(ready, times(1)).countDown();
792 public void testReadyCountDownForMemberUpAfterLeaderStateChanged() throws Exception {
793 new JavaTestKit(getSystem()) {
795 TestActorRef<ShardManager> shardManager = TestActorRef.create(getSystem(), newShardMgrProps());
797 String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
798 shardManager.underlyingActor().onReceiveCommand(new RoleChangeNotification(
799 memberId, null, RaftState.Follower.name()));
801 verify(ready, never()).countDown();
803 shardManager.underlyingActor().onReceiveCommand(new ShardLeaderStateChanged(memberId,
804 "member-2-shard-default-" + shardMrgIDSuffix, Optional.of(mock(DataTree.class)),
805 DataStoreVersions.CURRENT_VERSION));
807 shardManager.underlyingActor().onReceiveCommand(MockClusterWrapper.createMemberUp("member-2", getRef().path().toString()));
809 verify(ready, times(1)).countDown();
815 public void testRoleChangeNotificationDoNothingForUnknownShard() throws Exception {
816 new JavaTestKit(getSystem()) {
818 TestActorRef<ShardManager> shardManager = TestActorRef.create(getSystem(), newShardMgrProps());
820 shardManager.underlyingActor().onReceiveCommand(new RoleChangeNotification(
821 "unknown", RaftState.Candidate.name(), RaftState.Leader.name()));
823 verify(ready, never()).countDown();
830 public void testByDefaultSyncStatusIsFalse() throws Exception{
831 final Props persistentProps = newShardMgrProps();
832 final TestActorRef<ShardManager> shardManager =
833 TestActorRef.create(getSystem(), persistentProps);
835 ShardManager shardManagerActor = shardManager.underlyingActor();
837 assertEquals(false, shardManagerActor.getMBean().getSyncStatus());
841 public void testWhenShardIsLeaderSyncStatusIsTrue() throws Exception{
842 final Props persistentProps = ShardManager.props(
843 new MockClusterWrapper(),
844 new MockConfiguration(),
845 DatastoreContext.newBuilder().persistent(true).build(), ready, primaryShardInfoCache);
846 final TestActorRef<ShardManager> shardManager =
847 TestActorRef.create(getSystem(), persistentProps);
849 ShardManager shardManagerActor = shardManager.underlyingActor();
850 shardManagerActor.onReceiveCommand(new RoleChangeNotification("member-1-shard-default-unknown",
851 RaftState.Follower.name(), RaftState.Leader.name()));
853 assertEquals(true, shardManagerActor.getMBean().getSyncStatus());
857 public void testWhenShardIsCandidateSyncStatusIsFalse() throws Exception{
858 final Props persistentProps = newShardMgrProps();
859 final TestActorRef<ShardManager> shardManager =
860 TestActorRef.create(getSystem(), persistentProps);
862 ShardManager shardManagerActor = shardManager.underlyingActor();
863 shardManagerActor.onReceiveCommand(new RoleChangeNotification("member-1-shard-default-unknown",
864 RaftState.Follower.name(), RaftState.Candidate.name()));
866 assertEquals(false, shardManagerActor.getMBean().getSyncStatus());
868 // Send a FollowerInitialSyncStatus with status = true for the replica whose current state is candidate
869 shardManagerActor.onReceiveCommand(new FollowerInitialSyncUpStatus(true, "member-1-shard-default-unknown"));
871 assertEquals(false, shardManagerActor.getMBean().getSyncStatus());
875 public void testWhenShardIsFollowerSyncStatusDependsOnFollowerInitialSyncStatus() throws Exception{
876 final Props persistentProps = ShardManager.props(
877 new MockClusterWrapper(),
878 new MockConfiguration(),
879 DatastoreContext.newBuilder().persistent(true).build(), ready, primaryShardInfoCache);
880 final TestActorRef<ShardManager> shardManager =
881 TestActorRef.create(getSystem(), persistentProps);
883 ShardManager shardManagerActor = shardManager.underlyingActor();
884 shardManagerActor.onReceiveCommand(new RoleChangeNotification("member-1-shard-default-unknown",
885 RaftState.Candidate.name(), RaftState.Follower.name()));
887 // Initially will be false
888 assertEquals(false, shardManagerActor.getMBean().getSyncStatus());
890 // Send status true will make sync status true
891 shardManagerActor.onReceiveCommand(new FollowerInitialSyncUpStatus(true, "member-1-shard-default-unknown"));
893 assertEquals(true, shardManagerActor.getMBean().getSyncStatus());
895 // Send status false will make sync status false
896 shardManagerActor.onReceiveCommand(new FollowerInitialSyncUpStatus(false, "member-1-shard-default-unknown"));
898 assertEquals(false, shardManagerActor.getMBean().getSyncStatus());
903 public void testWhenMultipleShardsPresentSyncStatusMustBeTrueForAllShards() throws Exception{
904 final Props persistentProps = ShardManager.props(
905 new MockClusterWrapper(),
906 new MockConfiguration() {
908 public List<String> getMemberShardNames(String memberName) {
909 return Arrays.asList("default", "astronauts");
912 DatastoreContext.newBuilder().persistent(true).build(), ready, primaryShardInfoCache);
913 final TestActorRef<ShardManager> shardManager =
914 TestActorRef.create(getSystem(), persistentProps);
916 ShardManager shardManagerActor = shardManager.underlyingActor();
918 // Initially will be false
919 assertEquals(false, shardManagerActor.getMBean().getSyncStatus());
921 // Make default shard leader
922 shardManagerActor.onReceiveCommand(new RoleChangeNotification("member-1-shard-default-unknown",
923 RaftState.Follower.name(), RaftState.Leader.name()));
925 // default = Leader, astronauts is unknown so sync status remains false
926 assertEquals(false, shardManagerActor.getMBean().getSyncStatus());
928 // Make astronauts shard leader as well
929 shardManagerActor.onReceiveCommand(new RoleChangeNotification("member-1-shard-astronauts-unknown",
930 RaftState.Follower.name(), RaftState.Leader.name()));
932 // Now sync status should be true
933 assertEquals(true, shardManagerActor.getMBean().getSyncStatus());
935 // Make astronauts a Follower
936 shardManagerActor.onReceiveCommand(new RoleChangeNotification("member-1-shard-astronauts-unknown",
937 RaftState.Leader.name(), RaftState.Follower.name()));
939 // Sync status is not true
940 assertEquals(false, shardManagerActor.getMBean().getSyncStatus());
942 // Make the astronauts follower sync status true
943 shardManagerActor.onReceiveCommand(new FollowerInitialSyncUpStatus(true, "member-1-shard-astronauts-unknown"));
945 // Sync status is now true
946 assertEquals(true, shardManagerActor.getMBean().getSyncStatus());
951 public void testOnReceiveSwitchShardBehavior() throws Exception {
952 new JavaTestKit(getSystem()) {{
953 final ActorRef shardManager = getSystem().actorOf(newPropsShardMgrWithMockShardActor());
955 shardManager.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
956 shardManager.tell(new ActorInitialized(), mockShardActor);
958 shardManager.tell(new SwitchShardBehavior(mockShardName, "Leader", 1000), getRef());
960 SwitchBehavior switchBehavior = MessageCollectorActor.expectFirstMatching(mockShardActor, SwitchBehavior.class);
962 assertEquals(RaftState.Leader, switchBehavior.getNewState());
963 assertEquals(1000, switchBehavior.getNewTerm());
968 public void testOnReceiveCreateShard() {
969 new JavaTestKit(getSystem()) {{
970 datastoreContextBuilder.shardInitializationTimeout(1, TimeUnit.MINUTES).persistent(true);
972 ActorRef shardManager = getSystem().actorOf(newShardMgrProps(
973 new ConfigurationImpl(new EmptyModuleShardConfigProvider())));
975 SchemaContext schemaContext = TestModel.createTestContext();
976 shardManager.tell(new UpdateSchemaContext(schemaContext), ActorRef.noSender());
978 DatastoreContext datastoreContext = DatastoreContext.newBuilder().shardElectionTimeoutFactor(100).
979 persistent(false).build();
980 TestShardPropsCreator shardPropsCreator = new TestShardPropsCreator();
982 ModuleShardConfiguration config = new ModuleShardConfiguration(URI.create("foo-ns"), "foo-module",
983 "foo", null, Arrays.asList("member-1", "member-5", "member-6"));
984 shardManager.tell(new CreateShard(config, shardPropsCreator, datastoreContext), getRef());
986 expectMsgClass(duration("5 seconds"), CreateShardReply.class);
988 shardManager.tell(new FindLocalShard("foo", true), getRef());
990 expectMsgClass(duration("5 seconds"), LocalShardFound.class);
992 assertEquals("isRecoveryApplicable", false, shardPropsCreator.datastoreContext.isPersistent());
993 assertTrue("Epxected ShardPeerAddressResolver", shardPropsCreator.datastoreContext.getShardRaftConfig().
994 getPeerAddressResolver() instanceof ShardPeerAddressResolver);
995 assertEquals("peerMembers", Sets.newHashSet(new ShardIdentifier("foo", "member-5", shardMrgIDSuffix).toString(),
996 new ShardIdentifier("foo", "member-6", shardMrgIDSuffix).toString()),
997 shardPropsCreator.peerAddresses.keySet());
998 assertEquals("ShardIdentifier", new ShardIdentifier("foo", "member-1", shardMrgIDSuffix),
999 shardPropsCreator.shardId);
1000 assertSame("schemaContext", schemaContext, shardPropsCreator.schemaContext);
1002 // Send CreateShard with same name - should fail.
1004 shardManager.tell(new CreateShard(config, shardPropsCreator, null), getRef());
1006 expectMsgClass(duration("5 seconds"), akka.actor.Status.Failure.class);
1011 public void testOnReceiveCreateShardWithNoInitialSchemaContext() {
1012 new JavaTestKit(getSystem()) {{
1013 ActorRef shardManager = getSystem().actorOf(newShardMgrProps(
1014 new ConfigurationImpl(new EmptyModuleShardConfigProvider())));
1016 TestShardPropsCreator shardPropsCreator = new TestShardPropsCreator();
1018 ModuleShardConfiguration config = new ModuleShardConfiguration(URI.create("foo-ns"), "foo-module",
1019 "foo", null, Arrays.asList("member-1"));
1020 shardManager.tell(new CreateShard(config, shardPropsCreator, null), getRef());
1022 expectMsgClass(duration("5 seconds"), CreateShardReply.class);
1024 SchemaContext schemaContext = TestModel.createTestContext();
1025 shardManager.tell(new UpdateSchemaContext(schemaContext), ActorRef.noSender());
1027 shardManager.tell(new FindLocalShard("foo", true), getRef());
1029 expectMsgClass(duration("5 seconds"), LocalShardFound.class);
1031 assertSame("schemaContext", schemaContext, shardPropsCreator.schemaContext);
1032 assertNotNull("schemaContext is null", shardPropsCreator.datastoreContext);
1037 public void testAddShardReplicaForNonExistentShard() throws Exception {
1038 new JavaTestKit(getSystem()) {{
1039 ActorRef shardManager = getSystem().actorOf(newShardMgrProps(
1040 new ConfigurationImpl(new EmptyModuleShardConfigProvider())));
1042 shardManager.tell(new AddShardReplica("model-inventory"), getRef());
1043 Status.Failure resp = expectMsgClass(duration("2 seconds"), Status.Failure.class);
1045 assertEquals("Failure obtained", true,
1046 (resp.cause() instanceof IllegalArgumentException));
1051 public void testAddShardReplicaForAlreadyCreatedShard() throws Exception {
1052 new JavaTestKit(getSystem()) {{
1053 ActorRef shardManager = getSystem().actorOf(newShardMgrProps());
1054 shardManager.tell(new AddShardReplica("default"), getRef());
1055 Status.Failure resp = expectMsgClass(duration("2 seconds"), Status.Failure.class);
1056 assertEquals("Failure obtained", true,
1057 (resp.cause() instanceof IllegalArgumentException));
1062 public void testAddShardReplica() throws Exception {
1063 new JavaTestKit(getSystem()) {{
1064 MockConfiguration mockConfig = new MockConfiguration(ImmutableMap.<String, List<String>>builder().
1065 put("default", Arrays.asList("member-1", "member-2")).
1066 put("astronauts", Arrays.asList("member-2")).build());
1068 ActorRef shardManager = getSystem().actorOf(newShardMgrProps(mockConfig));
1070 shardManager.tell(new AddShardReplica("astronauts"), getRef());
1071 expectMsgClass(duration("2 seconds"), Status.Success.class);
1076 public void testRemoveShardReplicaForNonExistentShard() throws Exception {
1077 new JavaTestKit(getSystem()) {{
1078 ActorRef shardManager = getSystem().actorOf(newShardMgrProps(
1079 new ConfigurationImpl(new EmptyModuleShardConfigProvider())));
1081 shardManager.tell(new RemoveShardReplica("model-inventory"), getRef());
1082 Status.Failure resp = expectMsgClass(duration("2 seconds"), Status.Failure.class);
1083 assertEquals("Failure obtained", true,
1084 (resp.cause() instanceof IllegalArgumentException));
1089 private static class TestShardPropsCreator implements ShardPropsCreator {
1090 ShardIdentifier shardId;
1091 Map<String, String> peerAddresses;
1092 SchemaContext schemaContext;
1093 DatastoreContext datastoreContext;
1096 public Props newProps(ShardIdentifier shardId, Map<String, String> peerAddresses,
1097 DatastoreContext datastoreContext, SchemaContext schemaContext) {
1098 this.shardId = shardId;
1099 this.peerAddresses = peerAddresses;
1100 this.schemaContext = schemaContext;
1101 this.datastoreContext = datastoreContext;
1102 return Shard.props(shardId, peerAddresses, datastoreContext, schemaContext);
1107 private static class TestShardManager extends ShardManager {
1108 private final CountDownLatch recoveryComplete = new CountDownLatch(1);
1110 TestShardManager(String shardMrgIDSuffix) {
1111 super(new MockClusterWrapper(), new MockConfiguration(),
1112 DatastoreContext.newBuilder().dataStoreType(shardMrgIDSuffix).build(), ready,
1113 new PrimaryShardInfoFutureCache());
1117 public void handleRecover(Object message) throws Exception {
1119 super.handleRecover(message);
1121 if(message instanceof RecoveryCompleted) {
1122 recoveryComplete.countDown();
1127 void waitForRecoveryComplete() {
1128 assertEquals("Recovery complete", true,
1129 Uninterruptibles.awaitUninterruptibly(recoveryComplete, 5, TimeUnit.SECONDS));
1133 @SuppressWarnings("serial")
1134 static class TestShardManagerCreator implements Creator<TestShardManager> {
1135 String shardMrgIDSuffix;
1137 TestShardManagerCreator(String shardMrgIDSuffix) {
1138 this.shardMrgIDSuffix = shardMrgIDSuffix;
1142 public TestShardManager create() throws Exception {
1143 return new TestShardManager(shardMrgIDSuffix);
1148 private static class DelegatingShardManagerCreator implements Creator<ShardManager> {
1149 private static final long serialVersionUID = 1L;
1150 private final Creator<ShardManager> delegate;
1152 public DelegatingShardManagerCreator(Creator<ShardManager> delegate) {
1153 this.delegate = delegate;
1157 public ShardManager create() throws Exception {
1158 return delegate.create();
1162 private static class ForwardingShardManager extends ShardManager {
1163 private CountDownLatch findPrimaryMessageReceived = new CountDownLatch(1);
1164 private CountDownLatch memberUpReceived = new CountDownLatch(1);
1165 private CountDownLatch memberRemovedReceived = new CountDownLatch(1);
1166 private CountDownLatch memberUnreachableReceived = new CountDownLatch(1);
1167 private CountDownLatch memberReachableReceived = new CountDownLatch(1);
1168 private final ActorRef shardActor;
1169 private final String name;
1171 protected ForwardingShardManager(ClusterWrapper cluster, Configuration configuration,
1172 DatastoreContext datastoreContext, CountDownLatch waitTillReadyCountdownLatch, String name,
1173 ActorRef shardActor, PrimaryShardInfoFutureCache primaryShardInfoCache) {
1174 super(cluster, configuration, datastoreContext, waitTillReadyCountdownLatch, primaryShardInfoCache);
1175 this.shardActor = shardActor;
1180 public void handleCommand(Object message) throws Exception {
1182 super.handleCommand(message);
1184 if(message instanceof FindPrimary) {
1185 findPrimaryMessageReceived.countDown();
1186 } else if(message instanceof ClusterEvent.MemberUp) {
1187 String role = ((ClusterEvent.MemberUp)message).member().roles().head();
1188 if(!getCluster().getCurrentMemberName().equals(role)) {
1189 memberUpReceived.countDown();
1191 } else if(message instanceof ClusterEvent.MemberRemoved) {
1192 String role = ((ClusterEvent.MemberRemoved)message).member().roles().head();
1193 if(!getCluster().getCurrentMemberName().equals(role)) {
1194 memberRemovedReceived.countDown();
1196 } else if(message instanceof ClusterEvent.UnreachableMember) {
1197 String role = ((ClusterEvent.UnreachableMember)message).member().roles().head();
1198 if(!getCluster().getCurrentMemberName().equals(role)) {
1199 memberUnreachableReceived.countDown();
1201 } else if(message instanceof ClusterEvent.ReachableMember) {
1202 String role = ((ClusterEvent.ReachableMember)message).member().roles().head();
1203 if(!getCluster().getCurrentMemberName().equals(role)) {
1204 memberReachableReceived.countDown();
1211 public String persistenceId() {
1216 protected ActorRef newShardActor(SchemaContext schemaContext, ShardInformation info) {
1220 void waitForMemberUp() {
1221 assertEquals("MemberUp received", true,
1222 Uninterruptibles.awaitUninterruptibly(memberUpReceived, 5, TimeUnit.SECONDS));
1223 memberUpReceived = new CountDownLatch(1);
1226 void waitForMemberRemoved() {
1227 assertEquals("MemberRemoved received", true,
1228 Uninterruptibles.awaitUninterruptibly(memberRemovedReceived, 5, TimeUnit.SECONDS));
1229 memberRemovedReceived = new CountDownLatch(1);
1232 void waitForUnreachableMember() {
1233 assertEquals("UnreachableMember received", true,
1234 Uninterruptibles.awaitUninterruptibly(memberUnreachableReceived, 5, TimeUnit.SECONDS
1236 memberUnreachableReceived = new CountDownLatch(1);
1239 void waitForReachableMember() {
1240 assertEquals("ReachableMember received", true,
1241 Uninterruptibles.awaitUninterruptibly(memberReachableReceived, 5, TimeUnit.SECONDS));
1242 memberReachableReceived = new CountDownLatch(1);
1245 void verifyFindPrimary() {
1246 assertEquals("FindPrimary received", true,
1247 Uninterruptibles.awaitUninterruptibly(findPrimaryMessageReceived, 5, TimeUnit.SECONDS));
1248 findPrimaryMessageReceived = new CountDownLatch(1);