*/
package org.opendaylight.controller.cluster.datastore.shardmanager;
+import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.timeout;
-import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.AddressFromURIString;
+import akka.actor.PoisonPill;
import akka.actor.Props;
import akka.actor.Status;
import akka.actor.Status.Failure;
import akka.testkit.TestActorRef;
import akka.testkit.javadsl.TestKit;
import akka.util.Timeout;
-import com.google.common.base.Function;
import com.google.common.base.Stopwatch;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.SettableFuture;
import com.google.common.util.concurrent.Uninterruptibles;
-import java.net.URI;
import java.time.Duration;
import java.util.AbstractMap;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
+import java.util.function.Function;
import java.util.stream.Collectors;
+import org.junit.After;
import org.junit.AfterClass;
+import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
import org.opendaylight.controller.cluster.access.concepts.MemberName;
-import org.opendaylight.controller.cluster.datastore.AbstractShardManagerTest;
+import org.opendaylight.controller.cluster.datastore.AbstractClusterRefActorTest;
import org.opendaylight.controller.cluster.datastore.ClusterWrapperImpl;
import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
import org.opendaylight.controller.cluster.datastore.DatastoreContext;
import org.opendaylight.controller.cluster.datastore.DatastoreContextFactory;
import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
import org.opendaylight.controller.cluster.datastore.Shard;
+import org.opendaylight.controller.cluster.datastore.config.Configuration;
import org.opendaylight.controller.cluster.datastore.config.ConfigurationImpl;
import org.opendaylight.controller.cluster.datastore.config.EmptyModuleShardConfigProvider;
import org.opendaylight.controller.cluster.datastore.config.ModuleShardConfiguration;
import org.opendaylight.controller.cluster.datastore.messages.LocalPrimaryShardFound;
import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
-import org.opendaylight.controller.cluster.datastore.messages.PeerDown;
-import org.opendaylight.controller.cluster.datastore.messages.PeerUp;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
import org.opendaylight.controller.cluster.datastore.messages.RemotePrimaryShardFound;
import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener;
import org.opendaylight.controller.cluster.notifications.RoleChangeNotification;
import org.opendaylight.controller.cluster.raft.RaftState;
+import org.opendaylight.controller.cluster.raft.TestActorFactory;
import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus;
import org.opendaylight.controller.cluster.raft.base.messages.SwitchBehavior;
import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
import org.opendaylight.controller.cluster.raft.messages.ServerChangeStatus;
import org.opendaylight.controller.cluster.raft.messages.ServerRemoved;
import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy;
+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.concepts.Registration;
+import org.opendaylight.yangtools.yang.common.XMLNamespace;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
-import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext;
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 AbstractShardManagerTest {
+@RunWith(MockitoJUnitRunner.StrictStubs.class)
+public class ShardManagerTest extends AbstractClusterRefActorTest {
private static final Logger LOG = LoggerFactory.getLogger(ShardManagerTest.class);
+ private static final MemberName MEMBER_1 = MemberName.forName("member-1");
private static final MemberName MEMBER_2 = MemberName.forName("member-2");
private static final MemberName MEMBER_3 = MemberName.forName("member-3");
- private static SchemaContext TEST_SCHEMA_CONTEXT;
+ private static int ID_COUNTER = 1;
+ private static ActorRef mockShardActor;
+ private static ShardIdentifier mockShardName;
+ private static SettableFuture<Void> ready;
+ private static EffectiveModelContext TEST_SCHEMA_CONTEXT;
+
+ private final String shardMrgIDSuffix = "config" + ID_COUNTER++;
+ private final TestActorFactory actorFactory = new TestActorFactory(getSystem());
+ private final DatastoreContext.Builder datastoreContextBuilder = DatastoreContext.newBuilder()
+ .dataStoreName(shardMrgIDSuffix).shardInitializationTimeout(600, TimeUnit.MILLISECONDS)
+ .shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(6);
private final String shardMgrID = ShardManagerIdentifier.builder().type(shardMrgIDSuffix).build().toString();
TEST_SCHEMA_CONTEXT = null;
}
+ @Before
+ public void setUp() {
+ ready = SettableFuture.create();
+
+ InMemoryJournal.clear();
+ InMemorySnapshotStore.clear();
+
+ if (mockShardActor == null) {
+ mockShardName = ShardIdentifier.create(Shard.DEFAULT_NAME, MEMBER_1, "config");
+ mockShardActor = getSystem().actorOf(MessageCollectorActor.props(), mockShardName.toString());
+ }
+
+ MessageCollectorActor.clearMessages(mockShardActor);
+ }
+
+ @After
+ public void tearDown() {
+ InMemoryJournal.clear();
+ InMemorySnapshotStore.clear();
+
+ mockShardActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
+ await().atMost(Duration.ofSeconds(10)).until(mockShardActor::isTerminated);
+ mockShardActor = null;
+
+ actorFactory.close();
+ }
+
+ private TestShardManager.Builder newTestShardMgrBuilder() {
+ return TestShardManager.builder(datastoreContextBuilder).distributedDataStore(mock(DistributedDataStore.class));
+ }
+
+ private TestShardManager.Builder newTestShardMgrBuilder(final Configuration config) {
+ return TestShardManager.builder(datastoreContextBuilder).configuration(config)
+ .distributedDataStore(mock(DistributedDataStore.class));
+ }
+
+ private Props newShardMgrProps() {
+ return newShardMgrProps(new MockConfiguration());
+ }
+
+ private Props newShardMgrProps(final Configuration config) {
+ return newTestShardMgrBuilder(config).readinessFuture(ready).props();
+ }
+
private ActorSystem newActorSystem(final String config) {
return newActorSystem("cluster-test", config);
}
return system.actorOf(MessageCollectorActor.props(), name);
}
- private Props newShardMgrProps() {
- return newShardMgrProps(new MockConfiguration());
- }
-
private static DatastoreContextFactory newDatastoreContextFactory(final DatastoreContext datastoreContext) {
DatastoreContextFactory mockFactory = mock(DatastoreContextFactory.class);
doReturn(datastoreContext).when(mockFactory).getBaseDatastoreContext();
}
}
- final Creator<ShardManager> creator = new Creator<ShardManager>() {
+ final Creator<ShardManager> creator = new Creator<>() {
private static final long serialVersionUID = 1L;
@Override
public ShardManager create() {
final TestKit kit = new TestKit(getSystem());
- final ActorRef shardManager = actorFactory.createActor(Props.create(
+ final ActorRef shardManager = actorFactory.createActor(Props.create(ShardManager.class,
new DelegatingShardManagerCreator(creator)).withDispatcher(Dispatchers.DefaultDispatcherId()));
shardManager.tell(new UpdateSchemaContext(TEST_SCHEMA_CONTEXT), kit.getRef());
kit.getRef());
shardManager1.underlyingActor().waitForUnreachableMember();
-
- PeerDown peerDown = MessageCollectorActor.expectFirstMatching(mockShardActor1, PeerDown.class);
- assertEquals("getMemberName", MEMBER_2, peerDown.getMemberName());
MessageCollectorActor.clearMessages(mockShardActor1);
shardManager1.tell(MockClusterWrapper.createMemberRemoved("member-2", "akka://cluster-test@127.0.0.1:2558"),
kit.getRef());
- MessageCollectorActor.expectFirstMatching(mockShardActor1, PeerDown.class);
-
shardManager1.tell(new FindPrimary("default", true), kit.getRef());
kit.expectMsgClass(Duration.ofSeconds(5), NoShardLeaderException.class);
shardManager1.underlyingActor().waitForReachableMember();
- PeerUp peerUp = MessageCollectorActor.expectFirstMatching(mockShardActor1, PeerUp.class);
- assertEquals("getMemberName", MEMBER_2, peerUp.getMemberName());
- MessageCollectorActor.clearMessages(mockShardActor1);
-
shardManager1.tell(new FindPrimary("default", true), kit.getRef());
RemotePrimaryShardFound found1 = kit.expectMsgClass(Duration.ofSeconds(5), RemotePrimaryShardFound.class);
shardManager1.tell(MockClusterWrapper.createMemberUp("member-2", "akka://cluster-test@127.0.0.1:2558"),
kit.getRef());
- MessageCollectorActor.expectFirstMatching(mockShardActor1, PeerUp.class);
-
// Test FindPrimary wait succeeds after reachable member event.
shardManager1.tell(MockClusterWrapper.createUnreachableMember("member-2",
TestShardManager shardManager = newTestShardManager();
String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
- shardManager.onReceiveCommand(new RoleChangeNotification(
+ shardManager.handleCommand(new RoleChangeNotification(
memberId, RaftState.Candidate.name(), RaftState.Leader.name()));
+ assertFalse(ready.isDone());
- verify(ready, never()).countDown();
-
- shardManager.onReceiveCommand(new ShardLeaderStateChanged(memberId, memberId,
+ shardManager.handleCommand(new ShardLeaderStateChanged(memberId, memberId,
mock(DataTree.class), DataStoreVersions.CURRENT_VERSION));
-
- verify(ready, times(1)).countDown();
+ assertTrue(ready.isDone());
}
@Test
TestShardManager shardManager = newTestShardManager();
String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
- shardManager.onReceiveCommand(new RoleChangeNotification(memberId, null, RaftState.Follower.name()));
+ shardManager.handleCommand(new RoleChangeNotification(memberId, null, RaftState.Follower.name()));
+ assertFalse(ready.isDone());
- verify(ready, never()).countDown();
+ shardManager.handleCommand(MockClusterWrapper.createMemberUp("member-2", kit.getRef().path().toString()));
- shardManager.onReceiveCommand(MockClusterWrapper.createMemberUp("member-2", kit.getRef().path().toString()));
-
- shardManager.onReceiveCommand(
+ shardManager.handleCommand(
new ShardLeaderStateChanged(memberId, "member-2-shard-default-" + shardMrgIDSuffix,
mock(DataTree.class), DataStoreVersions.CURRENT_VERSION));
-
- verify(ready, times(1)).countDown();
+ assertTrue(ready.isDone());
}
@Test
TestShardManager shardManager = newTestShardManager();
String memberId = "member-1-shard-default-" + shardMrgIDSuffix;
- shardManager.onReceiveCommand(new RoleChangeNotification(memberId, null, RaftState.Follower.name()));
+ shardManager.handleCommand(new RoleChangeNotification(memberId, null, RaftState.Follower.name()));
+ assertFalse(ready.isDone());
- verify(ready, never()).countDown();
-
- shardManager.onReceiveCommand(
+ shardManager.handleCommand(
new ShardLeaderStateChanged(memberId, "member-2-shard-default-" + shardMrgIDSuffix,
mock(DataTree.class), DataStoreVersions.CURRENT_VERSION));
- shardManager.onReceiveCommand(MockClusterWrapper.createMemberUp("member-2", kit.getRef().path().toString()));
-
- verify(ready, times(1)).countDown();
+ shardManager.handleCommand(MockClusterWrapper.createMemberUp("member-2", kit.getRef().path().toString()));
+ assertTrue(ready.isDone());
}
@Test
public void testRoleChangeNotificationDoNothingForUnknownShard() throws Exception {
TestShardManager shardManager = newTestShardManager();
- shardManager.onReceiveCommand(new RoleChangeNotification("unknown", RaftState.Candidate.name(),
+ shardManager.handleCommand(new RoleChangeNotification("unknown", RaftState.Candidate.name(),
RaftState.Leader.name()));
-
- verify(ready, never()).countDown();
+ assertFalse(ready.isDone());
}
@Test
public void testWhenShardIsLeaderSyncStatusIsTrue() throws Exception {
TestShardManager shardManager = newTestShardManager();
- shardManager.onReceiveCommand(new RoleChangeNotification("member-1-shard-default-" + shardMrgIDSuffix,
+ shardManager.handleCommand(new RoleChangeNotification("member-1-shard-default-" + shardMrgIDSuffix,
RaftState.Follower.name(), RaftState.Leader.name()));
assertTrue(shardManager.getMBean().getSyncStatus());
TestShardManager shardManager = newTestShardManager();
String shardId = "member-1-shard-default-" + shardMrgIDSuffix;
- shardManager.onReceiveCommand(new RoleChangeNotification(shardId,
+ shardManager.handleCommand(new RoleChangeNotification(shardId,
RaftState.Follower.name(), RaftState.Candidate.name()));
assertFalse(shardManager.getMBean().getSyncStatus());
// Send a FollowerInitialSyncStatus with status = true for the replica whose current state is candidate
- shardManager.onReceiveCommand(new FollowerInitialSyncUpStatus(
+ shardManager.handleCommand(new FollowerInitialSyncUpStatus(
true, shardId));
assertFalse(shardManager.getMBean().getSyncStatus());
TestShardManager shardManager = newTestShardManager();
String shardId = "member-1-shard-default-" + shardMrgIDSuffix;
- shardManager.onReceiveCommand(new RoleChangeNotification(shardId,
+ shardManager.handleCommand(new RoleChangeNotification(shardId,
RaftState.Candidate.name(), RaftState.Follower.name()));
// Initially will be false
assertFalse(shardManager.getMBean().getSyncStatus());
// Send status true will make sync status true
- shardManager.onReceiveCommand(new FollowerInitialSyncUpStatus(true, shardId));
+ shardManager.handleCommand(new FollowerInitialSyncUpStatus(true, shardId));
assertTrue(shardManager.getMBean().getSyncStatus());
// Send status false will make sync status false
- shardManager.onReceiveCommand(new FollowerInitialSyncUpStatus(false, shardId));
+ shardManager.handleCommand(new FollowerInitialSyncUpStatus(false, shardId));
assertFalse(shardManager.getMBean().getSyncStatus());
}
// Make default shard leader
String defaultShardId = "member-1-shard-default-" + shardMrgIDSuffix;
- shardManager.onReceiveCommand(new RoleChangeNotification(defaultShardId,
+ shardManager.handleCommand(new RoleChangeNotification(defaultShardId,
RaftState.Follower.name(), RaftState.Leader.name()));
// default = Leader, astronauts is unknown so sync status remains false
// Make astronauts shard leader as well
String astronautsShardId = "member-1-shard-astronauts-" + shardMrgIDSuffix;
- shardManager.onReceiveCommand(new RoleChangeNotification(astronautsShardId,
+ shardManager.handleCommand(new RoleChangeNotification(astronautsShardId,
RaftState.Follower.name(), RaftState.Leader.name()));
// Now sync status should be true
assertTrue(shardManager.getMBean().getSyncStatus());
// Make astronauts a Follower
- shardManager.onReceiveCommand(new RoleChangeNotification(astronautsShardId,
+ shardManager.handleCommand(new RoleChangeNotification(astronautsShardId,
RaftState.Leader.name(), RaftState.Follower.name()));
// Sync status is not true
assertFalse(shardManager.getMBean().getSyncStatus());
// Make the astronauts follower sync status true
- shardManager.onReceiveCommand(new FollowerInitialSyncUpStatus(true, astronautsShardId));
+ shardManager.handleCommand(new FollowerInitialSyncUpStatus(true, astronautsShardId));
// Sync status is now true
assertTrue(shardManager.getMBean().getSyncStatus());
.createActor(newShardMgrProps(new ConfigurationImpl(new EmptyModuleShardConfigProvider()))
.withDispatcher(Dispatchers.DefaultDispatcherId()));
- SchemaContext schemaContext = TEST_SCHEMA_CONTEXT;
+ EffectiveModelContext schemaContext = TEST_SCHEMA_CONTEXT;
shardManager.tell(new UpdateSchemaContext(schemaContext), ActorRef.noSender());
DatastoreContext datastoreContext = DatastoreContext.newBuilder().shardElectionTimeoutFactor(100)
.persistent(false).build();
Shard.Builder shardBuilder = Shard.builder();
- ModuleShardConfiguration config = new ModuleShardConfiguration(URI.create("foo-ns"), "foo-module",
+ ModuleShardConfiguration config = new ModuleShardConfiguration(XMLNamespace.of("foo-ns"), "foo-module",
"foo", null, members("member-1", "member-5", "member-6"));
shardManager.tell(new CreateShard(config, shardBuilder, datastoreContext), kit.getRef());
shardManager.tell(new UpdateSchemaContext(TEST_SCHEMA_CONTEXT), ActorRef.noSender());
Shard.Builder shardBuilder = Shard.builder();
- ModuleShardConfiguration config = new ModuleShardConfiguration(URI.create("foo-ns"), "foo-module",
+ ModuleShardConfiguration config = new ModuleShardConfiguration(XMLNamespace.of("foo-ns"), "foo-module",
"foo", null, members("member-5", "member-6"));
shardManager.tell(new CreateShard(config, shardBuilder, null), kit.getRef());
Shard.Builder shardBuilder = Shard.builder();
- ModuleShardConfiguration config = new ModuleShardConfiguration(URI.create("foo-ns"), "foo-module",
+ ModuleShardConfiguration config = new ModuleShardConfiguration(XMLNamespace.of("foo-ns"), "foo-module",
"foo", null, members("member-1"));
shardManager.tell(new CreateShard(config, shardBuilder, null), kit.getRef());
kit.expectMsgClass(Duration.ofSeconds(5), Success.class);
- SchemaContext schemaContext = TEST_SCHEMA_CONTEXT;
+ EffectiveModelContext schemaContext = TEST_SCHEMA_CONTEXT;
shardManager.tell(new UpdateSchemaContext(schemaContext), ActorRef.noSender());
shardManager.tell(new FindLocalShard("foo", true), kit.getRef());
assertEquals("getType", shardMrgIDSuffix, datastoreSnapshot.getType());
assertNull("Expected null ShardManagerSnapshot", datastoreSnapshot.getShardManagerSnapshot());
- Function<ShardSnapshot, String> shardNameTransformer = ShardSnapshot::getName;
-
assertEquals("Shard names", Sets.newHashSet("shard1", "shard2"), Sets.newHashSet(
- datastoreSnapshot.getShardSnapshots().stream().map(shardNameTransformer).collect(Collectors.toSet())));
+ datastoreSnapshot.getShardSnapshots().stream().map(ShardSnapshot::getName).collect(Collectors.toSet())));
// Add a new replica
datastoreSnapshot = expectMsgClassOrFailure(DatastoreSnapshot.class, kit, "GetSnapshot");
assertEquals("Shard names", Sets.newHashSet("shard1", "shard2", "astronauts"), Sets.newHashSet(
- Lists.transform(datastoreSnapshot.getShardSnapshots(), shardNameTransformer)));
+ Lists.transform(datastoreSnapshot.getShardSnapshots(), ShardSnapshot::getName)));
ShardManagerSnapshot snapshot = datastoreSnapshot.getShardManagerSnapshot();
assertNotNull("Expected ShardManagerSnapshot", snapshot);
.put("astronauts", Collections.<String>emptyList()).build());
ShardManagerSnapshot snapshot =
- new ShardManagerSnapshot(Arrays.asList("shard1", "shard2", "astronauts"), Collections.emptyMap());
+ new ShardManagerSnapshot(Arrays.asList("shard1", "shard2", "astronauts"));
DatastoreSnapshot restoreFromSnapshot = new DatastoreSnapshot(shardMrgIDSuffix, snapshot,
Collections.<ShardSnapshot>emptyList());
TestActorRef<TestShardManager> shardManager = actorFactory.createTestActor(newTestShardMgrBuilder(mockConfig)
// persisted.
String[] restoredShards = { "default", "people" };
ShardManagerSnapshot snapshot =
- new ShardManagerSnapshot(Arrays.asList(restoredShards), Collections.emptyMap());
+ new ShardManagerSnapshot(Arrays.asList(restoredShards));
InMemorySnapshotStore.addSnapshot(shardManagerID, snapshot);
Uninterruptibles.sleepUninterruptibly(2, TimeUnit.MILLISECONDS);
.put("people", Arrays.asList("member-1", "member-2")).build());
String[] restoredShards = {"default", "astronauts"};
ShardManagerSnapshot snapshot =
- new ShardManagerSnapshot(Arrays.asList(restoredShards), Collections.emptyMap());
+ new ShardManagerSnapshot(Arrays.asList(restoredShards));
InMemorySnapshotStore.addSnapshot("shard-manager-" + shardMrgIDSuffix, snapshot);
// create shardManager to come up with restored data
}
Builder shardActor(final ActorRef newShardActor) {
- this.shardActor = newShardActor;
+ shardActor = newShardActor;
return this;
}
AbstractGenericCreator(final Class<C> shardManagerClass) {
this.shardManagerClass = shardManagerClass;
- cluster(new MockClusterWrapper()).configuration(new MockConfiguration()).waitTillReadyCountDownLatch(ready)
+ cluster(new MockClusterWrapper()).configuration(new MockConfiguration()).readinessFuture(ready)
.primaryShardInfoCache(new PrimaryShardInfoFutureCache());
}
Code Review / controller.git / blobdiff