X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2Fentityownership%2FEntityOwnershipShardTest.java;h=af347903f8e9e1fabb4cfba980f682444c368987;hb=a47dd7a5d21ca68804a6d0e2e3ca765f223c2ef4;hp=63ec7cf55e3e9bba4d7c0420d63da39cc237c46b;hpb=eb359c6eaee33eb91ac26ac8a3439be4ae1a974a;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShardTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShardTest.java index 63ec7cf55e..af347903f8 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShardTest.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShardTest.java @@ -14,6 +14,7 @@ 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.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.ENTITY_OWNERS_PATH; import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.candidatePath; @@ -38,6 +39,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.After; import org.junit.Test; +import org.opendaylight.controller.cluster.access.concepts.MemberName; import org.opendaylight.controller.cluster.datastore.AbstractShardTest; import org.opendaylight.controller.cluster.datastore.DataStoreVersions; import org.opendaylight.controller.cluster.datastore.DatastoreContext; @@ -48,6 +50,7 @@ import org.opendaylight.controller.cluster.datastore.entityownership.messages.Re import org.opendaylight.controller.cluster.datastore.entityownership.messages.RegisterListenerLocal; import org.opendaylight.controller.cluster.datastore.entityownership.messages.UnregisterCandidateLocal; import org.opendaylight.controller.cluster.datastore.entityownership.messages.UnregisterListenerLocal; +import org.opendaylight.controller.cluster.datastore.entityownership.selectionstrategy.EntityOwnerSelectionStrategyConfig; import org.opendaylight.controller.cluster.datastore.entityownership.selectionstrategy.LastCandidateSelectionStrategy; import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier; import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications; @@ -61,14 +64,15 @@ import org.opendaylight.controller.cluster.datastore.modification.Modification; import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry; import org.opendaylight.controller.cluster.raft.TestActorFactory; import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout; +import org.opendaylight.controller.cluster.raft.base.messages.TimeoutNow; import org.opendaylight.controller.cluster.raft.messages.AppendEntries; import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply; import org.opendaylight.controller.cluster.raft.messages.RequestVote; import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply; import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper; -import org.opendaylight.controller.md.sal.common.api.clustering.Entity; -import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipChange; -import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListener; +import org.opendaylight.mdsal.eos.dom.api.DOMEntity; +import org.opendaylight.mdsal.eos.dom.api.DOMEntityOwnershipChange; +import org.opendaylight.mdsal.eos.dom.api.DOMEntityOwnershipListener; import org.opendaylight.yangtools.yang.common.QName; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; @@ -109,10 +113,10 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { TestActorRef shard = actorFactory.createTestActor(newShardProps()); - kit.waitUntilLeader(shard); + ShardTestKit.waitUntilLeader(shard); YangInstanceIdentifier entityId = ENTITY_ID1; - Entity entity = new Entity(ENTITY_TYPE, entityId); + DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId); shard.tell(new RegisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); @@ -136,7 +140,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { withDispatcher(Dispatchers.DefaultDispatcherId())); YangInstanceIdentifier entityId = ENTITY_ID1; - Entity entity = new Entity(ENTITY_TYPE, entityId); + DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId); shard.tell(new RegisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); @@ -168,10 +172,10 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { ImmutableMap.builder().put(peerId, peer.path().toString()).build()). withDispatcher(Dispatchers.DefaultDispatcherId())); - kit.waitUntilLeader(shard); + ShardTestKit.waitUntilLeader(shard); YangInstanceIdentifier entityId = ENTITY_ID1; - Entity entity = new Entity(ENTITY_TYPE, entityId); + DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId); shard.tell(new RegisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); @@ -204,14 +208,14 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { ImmutableMap.builder().put(peerId, peer.path().toString()).build()). withDispatcher(Dispatchers.DefaultDispatcherId())); - kit.waitUntilLeader(shard); + ShardTestKit.waitUntilLeader(shard); // Drop AppendEntries and wait enough time for the shard to switch to IsolatedLeader. follower.dropAppendEntries = true; Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS); YangInstanceIdentifier entityId = ENTITY_ID1; - Entity entity = new Entity(ENTITY_TYPE, entityId); + DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId); shard.tell(new RegisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); @@ -241,7 +245,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { shard.tell(new AppendEntries(1L, peerId, -1L, -1L, Collections.emptyList(), -1L, -1L, DataStoreVersions.CURRENT_VERSION), peer); - shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef()); + shard.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef()); kit.expectMsgClass(SuccessReply.class); MockLeader leader = peer.underlyingActor(); @@ -257,7 +261,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { shard.tell(dataStoreContextBuilder.shardTransactionCommitTimeoutInSeconds(1).build(), ActorRef.noSender()); - shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID2)), kit.getRef()); + shard.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, ENTITY_ID2)), kit.getRef()); kit.expectMsgClass(SuccessReply.class); assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly( @@ -274,7 +278,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { for(int i = 1; i <= max; i++) { YangInstanceIdentifier id = YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "test" + i)); entityIds.add(id); - shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, id)), kit.getRef()); + shard.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, id)), kit.getRef()); } assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly( @@ -296,9 +300,9 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { public void testOnUnregisterCandidateLocal() throws Exception { ShardTestKit kit = new ShardTestKit(getSystem()); TestActorRef shard = actorFactory.createTestActor(newShardProps()); - kit.waitUntilLeader(shard); + ShardTestKit.waitUntilLeader(shard); - Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1); + DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1); // Register @@ -328,9 +332,9 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { public void testOwnershipChanges() throws Exception { ShardTestKit kit = new ShardTestKit(getSystem()); TestActorRef shard = actorFactory.createTestActor(newShardProps()); - kit.waitUntilLeader(shard); + ShardTestKit.waitUntilLeader(shard); - Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1); + DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1); ShardDataTree shardDataTree = shard.underlyingActor().getDataStore(); // Add a remote candidate @@ -405,28 +409,28 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { TestActorRef peer1 = actorFactory.createTestActor(newShardProps(peerId1, ImmutableMap.builder().put(leaderId.toString(), ""). put(peerId2.toString(), "").build(), - peerMemberName1).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString()); + peerMemberName1, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString()); TestActorRef peer2 = actorFactory.createTestActor(newShardProps(peerId2, ImmutableMap.builder().put(leaderId.toString(), ""). put(peerId1.toString(), "").build(), - peerMemberName2). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString()); + peerMemberName2, EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString()); TestActorRef leader = actorFactory.createTestActor(newShardProps(leaderId, ImmutableMap.builder().put(peerId1.toString(), peer1.path().toString()). - put(peerId2.toString(), peer2.path().toString()).build(), LOCAL_MEMBER_NAME). + put(peerId2.toString(), peer2.path().toString()).build(), LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString()); - leader.tell(new ElectionTimeout(), leader); + leader.tell(TimeoutNow.INSTANCE, leader); - kit.waitUntilLeader(leader); + ShardTestKit.waitUntilLeader(leader); // Send PeerDown and PeerUp with no entities - leader.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender()); - leader.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender()); + leader.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender()); + leader.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender()); // Add candidates for entity1 with the local leader as the owner - leader.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef()); + leader.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef()); kit.expectMsgClass(SuccessReply.class); verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); @@ -451,7 +455,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName2), kit); verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2); - leader.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID3)), kit.getRef()); + leader.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, ENTITY_ID3)), kit.getRef()); kit.expectMsgClass(SuccessReply.class); verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME); @@ -479,10 +483,10 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { kit.expectMsgClass(JavaTestKit.duration("5 seconds"), Terminated.class); kit.unwatch(peer2); - leader.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender()); + leader.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender()); // Send PeerDown again - should be noop - leader.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender()); - peer1.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender()); + leader.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender()); + peer1.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender()); verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, ""); // no other candidates so should clear verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME); @@ -498,11 +502,11 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { peer2 = actorFactory.createTestActor(newShardProps(peerId2, ImmutableMap.builder().put(leaderId.toString(), ""). put(peerId1.toString(), "").build(), - peerMemberName2). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString()); - leader.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender()); + peerMemberName2, EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString()); + leader.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender()); // Send PeerUp again - should be noop - leader.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender()); - peer1.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender()); + leader.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender()); + peer1.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender()); verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME); verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1); @@ -524,7 +528,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { // Kill peerMember1 and send PeerDown - entity 2 should get a new owner selected peer1.tell(PoisonPill.getInstance(), ActorRef.noSender()); - leader.tell(new PeerDown(peerMemberName1, peerId1.toString()), ActorRef.noSender()); + leader.tell(new PeerDown(peerId1.getMemberName(), peerId1.toString()), ActorRef.noSender()); verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2); @@ -539,8 +543,8 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { peer1 = actorFactory.createTestActor(newShardProps(peerId1, ImmutableMap.builder().put(leaderId.toString(), ""). put(peerId2.toString(), "").build(), - peerMemberName1).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString()); - leader.tell(new PeerUp(peerMemberName1, peerId1.toString()), ActorRef.noSender()); + peerMemberName1, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString()); + leader.tell(new PeerUp(peerId1.getMemberName(), peerId1.toString()), ActorRef.noSender()); verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, ""); verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME); @@ -558,14 +562,14 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { // the entities (1 and 3) previously owned by the local leader member. peer2.tell(new PeerAddressResolved(peerId1.toString(), peer1.path().toString()), ActorRef.noSender()); - peer2.tell(new PeerUp(LOCAL_MEMBER_NAME, leaderId.toString()), ActorRef.noSender()); - peer2.tell(new PeerUp(peerMemberName1, peerId1.toString()), ActorRef.noSender()); + peer2.tell(new PeerUp(leaderId.getMemberName(), leaderId.toString()), ActorRef.noSender()); + peer2.tell(new PeerUp(peerId1.getMemberName(), peerId1.toString()), ActorRef.noSender()); leader.tell(PoisonPill.getInstance(), ActorRef.noSender()); - peer2.tell(new PeerDown(LOCAL_MEMBER_NAME, leaderId.toString()), ActorRef.noSender()); - peer2.tell(new ElectionTimeout(), peer2); + peer2.tell(new PeerDown(leaderId.getMemberName(), leaderId.toString()), ActorRef.noSender()); + peer2.tell(TimeoutNow.INSTANCE, peer2); - kit.waitUntilLeader(peer2); + ShardTestKit.waitUntilLeader(peer2); verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, ""); verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID3, peerMemberName2); @@ -588,15 +592,16 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { TestActorRef leader = actorFactory.createTestActor(newShardProps(leaderId, ImmutableMap.builder().put(localId.toString(), shard.path().toString()).build(), - LOCAL_MEMBER_NAME).withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString()); - leader.tell(new ElectionTimeout(), leader); + leaderId.getMemberName().getName(), EntityOwnerSelectionStrategyConfig.newBuilder().build()) + .withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString()); + leader.tell(TimeoutNow.INSTANCE, leader); - kit.waitUntilLeader(leader); + ShardTestKit.waitUntilLeader(leader); shard.tell(new PeerAddressResolved(leaderId.toString(), leader.path().toString()), ActorRef.noSender()); - Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1); - EntityOwnershipListener listener = mock(EntityOwnershipListener.class); + DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1); + DOMEntityOwnershipListener listener = mock(DOMEntityOwnershipListener.class); shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef()); kit.expectMsgClass(SuccessReply.class); @@ -605,20 +610,20 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { shard.tell(new RegisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); - verifyCommittedEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME); + verifyCommittedEntityCandidate(shard, entity.getType(), entity.getIdentifier(), LOCAL_MEMBER_NAME); verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true)); reset(listener); // Simulate a replicated commit from the leader to remove the local candidate that would occur after a // network partition is healed. - leader.tell(new PeerDown(LOCAL_MEMBER_NAME, localId.toString()), ActorRef.noSender()); + leader.tell(new PeerDown(localId.getMemberName(), localId.toString()), ActorRef.noSender()); verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, true, false, false)); // Since the the shard has a local candidate registered, it should re-add its candidate to the entity. - verifyCommittedEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME); + verifyCommittedEntityCandidate(shard, entity.getType(), entity.getIdentifier(), LOCAL_MEMBER_NAME); verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true)); // Unregister the local candidate and verify it's removed and no re-added. @@ -626,24 +631,24 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { shard.tell(new UnregisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); - verifyNoEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME); + verifyNoEntityCandidate(shard, entity.getType(), entity.getIdentifier(), LOCAL_MEMBER_NAME); Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS); - verifyNoEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME); + verifyNoEntityCandidate(shard, entity.getType(), entity.getIdentifier(), LOCAL_MEMBER_NAME); } @Test public void testListenerRegistration() throws Exception { ShardTestKit kit = new ShardTestKit(getSystem()); TestActorRef shard = actorFactory.createTestActor(newShardProps()); - kit.waitUntilLeader(shard); + ShardTestKit.waitUntilLeader(shard); ShardDataTree shardDataTree = shard.underlyingActor().getDataStore(); String otherEntityType = "otherEntityType"; - Entity entity1 = new Entity(ENTITY_TYPE, ENTITY_ID1); - Entity entity2 = new Entity(ENTITY_TYPE, ENTITY_ID2); - Entity entity3 = new Entity(ENTITY_TYPE, ENTITY_ID3); - Entity entity4 = new Entity(otherEntityType, ENTITY_ID3); - EntityOwnershipListener listener = mock(EntityOwnershipListener.class); + DOMEntity entity1 = new DOMEntity(ENTITY_TYPE, ENTITY_ID1); + DOMEntity entity2 = new DOMEntity(ENTITY_TYPE, ENTITY_ID2); + DOMEntity entity3 = new DOMEntity(ENTITY_TYPE, ENTITY_ID3); + DOMEntity entity4 = new DOMEntity(otherEntityType, ENTITY_ID3); + DOMEntityOwnershipListener listener = mock(DOMEntityOwnershipListener.class); // Register listener @@ -674,9 +679,9 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { // Register remote candidate for entity1 String remoteMemberName = "remoteMember"; - writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, entity1.getId(), remoteMemberName), + writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, entity1.getIdentifier(), remoteMemberName), shardDataTree); - verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entity1.getId(), remoteMemberName); + verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entity1.getIdentifier(), remoteMemberName); // Unregister the local candidate for entity1 and verify listener is notified @@ -694,9 +699,9 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { shard.tell(new RegisterCandidateLocal(entity3), kit.getRef()); kit.expectMsgClass(SuccessReply.class); - verifyOwner(shard, ENTITY_TYPE, entity3.getId(), LOCAL_MEMBER_NAME); + verifyOwner(shard, ENTITY_TYPE, entity3.getIdentifier(), LOCAL_MEMBER_NAME); Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS); - verify(listener, never()).ownershipChanged(any(EntityOwnershipChange.class)); + verify(listener, never()).ownershipChanged(any(DOMEntityOwnershipChange.class)); // Re-register the listener and verify it gets notified of currently owned entities @@ -709,7 +714,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { ownershipChange(entity3, false, true, true))); Uninterruptibles.sleepUninterruptibly(300, TimeUnit.MILLISECONDS); verify(listener, never()).ownershipChanged(ownershipChange(entity4)); - verify(listener, never()).ownershipChanged(ownershipChange(entity1)); + verify(listener, times(1)).ownershipChanged(ownershipChange(entity1)); } private static void commitModification(TestActorRef shard, NormalizedNode node, @@ -718,11 +723,11 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { modifications.addModification(new MergeModification(ENTITY_OWNERS_PATH, node)); shard.tell(modifications, sender.getRef()); - sender.expectMsgClass(CommitTransactionReply.SERIALIZABLE_CLASS); + sender.expectMsgClass(CommitTransactionReply.class); } private static BatchedModifications newBatchedModifications() { - BatchedModifications modifications = new BatchedModifications("tnx", DataStoreVersions.CURRENT_VERSION, ""); + BatchedModifications modifications = new BatchedModifications(nextTransactionId(), DataStoreVersions.CURRENT_VERSION); modifications.setDoCommitOnReady(true); modifications.setReady(true); modifications.setTotalMessagesSent(1); @@ -803,44 +808,49 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { return newShardProps(Collections.emptyMap()); } - private Props newShardProps(Map peers) { - return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME); + private Props newShardProps(EntityOwnerSelectionStrategyConfig strategyConfig, Map peers) { + return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, strategyConfig); } - private Props newShardProps(ShardIdentifier shardId, Map peers, String memberName) { + private Props newShardProps(Map peers) { + return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build()); + } + + private Props newShardProps(ShardIdentifier shardId, Map peers, String memberName, + EntityOwnerSelectionStrategyConfig config) { return EntityOwnershipShard.newBuilder().id(shardId).peerAddresses(peers). datastoreContext(dataStoreContextBuilder.build()).schemaContext(SCHEMA_CONTEXT). - localMemberName(memberName).props().withDispatcher(Dispatchers.DefaultDispatcherId()); + localMemberName(MemberName.forName(memberName)).ownerSelectionStrategyConfig(config).props() + .withDispatcher(Dispatchers.DefaultDispatcherId()); } private static ShardIdentifier newShardId(String memberName) { - return ShardIdentifier.builder().memberName(memberName).shardName("entity-ownership"). - type("operational" + NEXT_SHARD_NUM.getAndIncrement()).build(); + return ShardIdentifier.create("entity-ownership", MemberName.forName(memberName), + "operational" + NEXT_SHARD_NUM.getAndIncrement()); } - public static class TestEntityOwnershipShard extends EntityOwnershipShard { + private static class TestEntityOwnershipShard extends EntityOwnershipShard { TestEntityOwnershipShard(ShardIdentifier name, Map peerAddresses, DatastoreContext datastoreContext) { super(newBuilder().id(name).peerAddresses(peerAddresses).datastoreContext(datastoreContext). - schemaContext(SCHEMA_CONTEXT).localMemberName(LOCAL_MEMBER_NAME)); + schemaContext(SCHEMA_CONTEXT).localMemberName(name.getMemberName())); } @Override - public void onReceiveCommand(Object message) throws Exception { + public void handleCommand(Object message) { if(!(message instanceof ElectionTimeout)) { - super.onReceiveCommand(message); + super.handleCommand(message); } } - - } - public static class MockFollower extends UntypedActor { + private static class MockFollower extends UntypedActor { volatile boolean grantVote; volatile boolean dropAppendEntries; private final String myId; + @SuppressWarnings("unused") public MockFollower(String myId) { this(myId, true); } @@ -875,20 +885,71 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { @Test - public void testDelayedEntityOwnerSelection() throws Exception { + public void testDelayedEntityOwnerSelectionWhenMaxPeerRequestsReceived() throws Exception { ShardTestKit kit = new ShardTestKit(getSystem()); - TestActorRef shard = actorFactory.createTestActor(newShardProps()); - shard.underlyingActor().addEntityOwnerSelectionStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500); - kit.waitUntilLeader(shard); + EntityOwnerSelectionStrategyConfig.Builder builder + = EntityOwnerSelectionStrategyConfig.newBuilder().addStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500); - Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1); + String peerId = newShardId("follower").toString(); + TestActorRef peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId, false). + withDispatcher(Dispatchers.DefaultDispatcherId()), peerId); + + peer.underlyingActor().grantVote = true; + + TestActorRef shard = actorFactory.createTestActor(newShardProps(builder.build(), + ImmutableMap.of(peerId.toString(), peer.path().toString()))); + ShardTestKit.waitUntilLeader(shard); + + DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1); ShardDataTree shardDataTree = shard.underlyingActor().getDataStore(); // Add a remote candidate - String remoteMemberName1 = "remoteMember1"; + String remoteMemberName1 = "follower"; writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree); + // Register local + + shard.tell(new RegisterCandidateLocal(entity), kit.getRef()); + kit.expectMsgClass(SuccessReply.class); + + // Verify the local candidate becomes owner + + verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1); + verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); + verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); + } + + @Test + public void testDelayedEntityOwnerSelection() throws Exception { + ShardTestKit kit = new ShardTestKit(getSystem()); + EntityOwnerSelectionStrategyConfig.Builder builder + = EntityOwnerSelectionStrategyConfig.newBuilder().addStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500); + + String follower1Id = newShardId("follower1").toString(); + TestActorRef follower1 = actorFactory.createTestActor(Props.create(MockFollower.class, follower1Id, false). + withDispatcher(Dispatchers.DefaultDispatcherId()), follower1Id); + + follower1.underlyingActor().grantVote = true; + + String follower2Id = newShardId("follower").toString(); + TestActorRef follower2 = actorFactory.createTestActor(Props.create(MockFollower.class, follower2Id, false). + withDispatcher(Dispatchers.DefaultDispatcherId()), follower2Id); + + follower2.underlyingActor().grantVote = true; + + + TestActorRef shard = actorFactory.createTestActor(newShardProps(builder.build(), + ImmutableMap.of(follower1Id.toString(), follower2.path().toString(), follower2Id.toString(), follower2.path().toString()))); + ShardTestKit.waitUntilLeader(shard); + + DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1); + ShardDataTree shardDataTree = shard.underlyingActor().getDataStore(); + + // Add a remote candidate + + String remoteMemberName1 = "follower"; + writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree); // Register local @@ -902,7 +963,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); } - public static class MockLeader extends UntypedActor { + private static class MockLeader extends UntypedActor { volatile CountDownLatch modificationsReceived = new CountDownLatch(1); List receivedModifications = new ArrayList<>(); volatile boolean sendReply = true; @@ -924,7 +985,8 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { } } - getSender().tell(CommitTransactionReply.INSTANCE.toSerializable(), getSelf()); + getSender().tell(CommitTransactionReply.instance(DataStoreVersions.CURRENT_VERSION). + toSerializable(), getSelf()); } else { sendReply = true; }