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=eb649207f7d67c3553a22f2bd4d16ef2eb7b8097;hb=d54ecf76d2bb2a06b0d94a8d4b8ad6ca0d7830e0;hp=c2fbfea98fa3cbf4c70068a89df78b8b1c2028aa;hpb=1f3f61ea49191bb0ada2d4de831a10f0a38a104d;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 c2fbfea98f..eb649207f7 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 @@ -48,6 +48,8 @@ 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; import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply; @@ -66,7 +68,6 @@ 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.EntityOwnershipCandidate; import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipChange; import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListener; import org.opendaylight.yangtools.yang.common.QName; @@ -113,14 +114,12 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { YangInstanceIdentifier entityId = ENTITY_ID1; Entity entity = new Entity(ENTITY_TYPE, entityId); - EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class); - shard.tell(new RegisterCandidateLocal(candidate, entity), kit.getRef()); + shard.tell(new RegisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME); verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME); - verify(candidate, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true)); } @Test @@ -139,19 +138,15 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { YangInstanceIdentifier entityId = ENTITY_ID1; Entity entity = new Entity(ENTITY_TYPE, entityId); - EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class); - shard.tell(new RegisterCandidateLocal(candidate, entity), kit.getRef()); + shard.tell(new RegisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); // Now grant the vote so the shard becomes the leader. This should retry the commit. peer.underlyingActor().grantVote = true; verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME); - verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME); - - verify(candidate, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true)); } @Test @@ -178,9 +173,8 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { YangInstanceIdentifier entityId = ENTITY_ID1; Entity entity = new Entity(ENTITY_TYPE, entityId); - EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class); - shard.tell(new RegisterCandidateLocal(candidate, entity), kit.getRef()); + shard.tell(new RegisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); // Wait enough time for the commit to timeout. @@ -191,10 +185,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { follower.dropAppendEntries = false; verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME); - verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME); - - verify(candidate, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true)); } @Test @@ -222,18 +213,14 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { YangInstanceIdentifier entityId = ENTITY_ID1; Entity entity = new Entity(ENTITY_TYPE, entityId); - EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class); - shard.tell(new RegisterCandidateLocal(candidate, entity), kit.getRef()); + shard.tell(new RegisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); // Resume AppendEntries - the candidate write should now be committed. follower.dropAppendEntries = false; verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME); - verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME); - - verify(candidate, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true)); } @Test @@ -255,9 +242,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { shard.tell(new AppendEntries(1L, peerId, -1L, -1L, Collections.emptyList(), -1L, -1L, DataStoreVersions.CURRENT_VERSION), peer); - EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class); - - shard.tell(new RegisterCandidateLocal(candidate, new Entity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef()); + shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef()); kit.expectMsgClass(SuccessReply.class); MockLeader leader = peer.underlyingActor(); @@ -273,7 +258,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { shard.tell(dataStoreContextBuilder.shardTransactionCommitTimeoutInSeconds(1).build(), ActorRef.noSender()); - shard.tell(new RegisterCandidateLocal(candidate, new Entity(ENTITY_TYPE, ENTITY_ID2)), kit.getRef()); + shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID2)), kit.getRef()); kit.expectMsgClass(SuccessReply.class); assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly( @@ -290,7 +275,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(candidate, new Entity(ENTITY_TYPE, id)), kit.getRef()); + shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, id)), kit.getRef()); } assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly( @@ -315,35 +300,29 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { kit.waitUntilLeader(shard); Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1); - EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class); // Register - shard.tell(new RegisterCandidateLocal(candidate, entity), kit.getRef()); + shard.tell(new RegisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); - verify(candidate, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true)); // Unregister - reset(candidate); - - shard.tell(new UnregisterCandidateLocal(candidate, entity), kit.getRef()); + shard.tell(new UnregisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, ""); - //verify(candidate).ownershipChanged(entity, true, false, false); // Register again - shard.tell(new RegisterCandidateLocal(candidate, entity), kit.getRef()); + shard.tell(new RegisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); - verify(candidate, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true)); } @Test @@ -353,7 +332,6 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { kit.waitUntilLeader(shard); Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1); - EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class); ShardDataTree shardDataTree = shard.underlyingActor().getDataStore(); // Add a remote candidate @@ -363,7 +341,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { // Register local - shard.tell(new RegisterCandidateLocal(candidate, entity), kit.getRef()); + shard.tell(new RegisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); // Verify the remote candidate becomes owner @@ -371,51 +349,42 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1); verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1); - verify(candidate, never()).ownershipChanged(any(EntityOwnershipChange.class)); // Add another remote candidate and verify ownership doesn't change - reset(candidate); String remoteMemberName2 = "remoteMember2"; writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree); verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2); Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS); verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1); - verify(candidate, never()).ownershipChanged(any(EntityOwnershipChange.class)); // Remove the second remote candidate and verify ownership doesn't change - reset(candidate); deleteNode(candidatePath(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree); verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2); Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS); verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1); - verify(candidate, never()).ownershipChanged(any(EntityOwnershipChange.class)); // Remove the first remote candidate and verify the local candidate becomes owner - reset(candidate); deleteNode(candidatePath(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree); verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1); verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); - verify(candidate, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true)); // Add the second remote candidate back and verify ownership doesn't change - reset(candidate); writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree); verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2); Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS); verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); - verify(candidate, never()).ownershipChanged(any(EntityOwnershipChange.class)); // Unregister the local candidate and verify the second remote candidate becomes owner - shard.tell(new UnregisterCandidateLocal(candidate, entity), kit.getRef()); + shard.tell(new UnregisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); @@ -437,22 +406,20 @@ 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); kit.waitUntilLeader(leader); - EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class); - // Send PeerDown and PeerUp with no entities leader.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender()); @@ -460,7 +427,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { // Add candidates for entity1 with the local leader as the owner - leader.tell(new RegisterCandidateLocal(candidate, new Entity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef()); + leader.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef()); kit.expectMsgClass(SuccessReply.class); verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); @@ -485,7 +452,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(candidate, new Entity(ENTITY_TYPE, ENTITY_ID3)), kit.getRef()); + leader.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID3)), kit.getRef()); kit.expectMsgClass(SuccessReply.class); verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME); @@ -532,7 +499,7 @@ 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()); + peerMemberName2, EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString()); leader.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender()); // Send PeerUp again - should be noop leader.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender()); @@ -573,7 +540,7 @@ 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()); + peerMemberName1, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString()); leader.tell(new PeerUp(peerMemberName1, peerId1.toString()), ActorRef.noSender()); verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, ""); @@ -622,7 +589,7 @@ 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()); + LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString()); leader.tell(new ElectionTimeout(), leader); kit.waitUntilLeader(leader); @@ -630,31 +597,34 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { shard.tell(new PeerAddressResolved(leaderId.toString(), leader.path().toString()), ActorRef.noSender()); Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1); - EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class); + EntityOwnershipListener listener = mock(EntityOwnershipListener.class); + + shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef()); + kit.expectMsgClass(SuccessReply.class); // Register local candidate - shard.tell(new RegisterCandidateLocal(candidate, entity), kit.getRef()); + shard.tell(new RegisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); verifyCommittedEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME); - verify(candidate, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true)); - reset(candidate); + 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()); - verify(candidate, timeout(5000)).ownershipChanged(ownershipChange(entity, true, false, false)); + 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); - verify(candidate, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true)); + verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true)); // Unregister the local candidate and verify it's removed and no re-added. - shard.tell(new UnregisterCandidateLocal(candidate, entity), kit.getRef()); + shard.tell(new UnregisterCandidateLocal(entity), kit.getRef()); kit.expectMsgClass(SuccessReply.class); verifyNoEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME); @@ -675,7 +645,6 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { Entity entity3 = new Entity(ENTITY_TYPE, ENTITY_ID3); Entity entity4 = new Entity(otherEntityType, ENTITY_ID3); EntityOwnershipListener listener = mock(EntityOwnershipListener.class); - EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class); // Register listener @@ -684,12 +653,12 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { // Register a couple candidates for the desired entity type and verify listener is notified. - shard.tell(new RegisterCandidateLocal(candidate, entity1), kit.getRef()); + shard.tell(new RegisterCandidateLocal(entity1), kit.getRef()); kit.expectMsgClass(SuccessReply.class); verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, false, true, true)); - shard.tell(new RegisterCandidateLocal(candidate, entity2), kit.getRef()); + shard.tell(new RegisterCandidateLocal(entity2), kit.getRef()); kit.expectMsgClass(SuccessReply.class); verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity2, false, true, true)); @@ -697,7 +666,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { // Register another candidate for another entity type and verify listener is not notified. - shard.tell(new RegisterCandidateLocal(candidate, entity4), kit.getRef()); + shard.tell(new RegisterCandidateLocal(entity4), kit.getRef()); kit.expectMsgClass(SuccessReply.class); Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS); @@ -712,7 +681,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { // Unregister the local candidate for entity1 and verify listener is notified - shard.tell(new UnregisterCandidateLocal(candidate, entity1), kit.getRef()); + shard.tell(new UnregisterCandidateLocal(entity1), kit.getRef()); kit.expectMsgClass(SuccessReply.class); verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, true, false, true)); @@ -723,7 +692,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { shard.tell(new UnregisterListenerLocal(listener, ENTITY_TYPE), kit.getRef()); kit.expectMsgClass(SuccessReply.class); - shard.tell(new RegisterCandidateLocal(candidate, entity3), kit.getRef()); + shard.tell(new RegisterCandidateLocal(entity3), kit.getRef()); kit.expectMsgClass(SuccessReply.class); verifyOwner(shard, ENTITY_TYPE, entity3.getId(), LOCAL_MEMBER_NAME); @@ -732,7 +701,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { // Re-register the listener and verify it gets notified of currently owned entities - reset(listener, candidate); + reset(listener); shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef()); kit.expectMsgClass(SuccessReply.class); @@ -742,11 +711,9 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { Uninterruptibles.sleepUninterruptibly(300, TimeUnit.MILLISECONDS); verify(listener, never()).ownershipChanged(ownershipChange(entity4)); verify(listener, never()).ownershipChanged(ownershipChange(entity1)); - verify(candidate, never()).ownershipChanged(ownershipChange(entity2)); - verify(candidate, never()).ownershipChanged(ownershipChange(entity3)); } - private void commitModification(TestActorRef shard, NormalizedNode node, + private static void commitModification(TestActorRef shard, NormalizedNode node, JavaTestKit sender) { BatchedModifications modifications = newBatchedModifications(); modifications.addModification(new MergeModification(ENTITY_OWNERS_PATH, node)); @@ -755,7 +722,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { sender.expectMsgClass(CommitTransactionReply.SERIALIZABLE_CLASS); } - private BatchedModifications newBatchedModifications() { + private static BatchedModifications newBatchedModifications() { BatchedModifications modifications = new BatchedModifications("tnx", DataStoreVersions.CURRENT_VERSION, ""); modifications.setDoCommitOnReady(true); modifications.setReady(true); @@ -819,8 +786,8 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { entityId, candidateName, true); } - private void verifyOwner(final TestActorRef shard, String entityType, YangInstanceIdentifier entityId, - String localMemberName) { + private static void verifyOwner(final TestActorRef shard, String entityType, + YangInstanceIdentifier entityId, String localMemberName) { verifyOwner(localMemberName, entityType, entityId, new Function>() { @Override public NormalizedNode apply(YangInstanceIdentifier path) { @@ -837,16 +804,23 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { return newShardProps(Collections.emptyMap()); } + private Props newShardProps(EntityOwnerSelectionStrategyConfig strategyConfig) { + return newShardProps(newShardId(LOCAL_MEMBER_NAME), Collections.emptyMap(), + LOCAL_MEMBER_NAME, strategyConfig); + } + private Props newShardProps(Map peers) { - return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME); + return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build()); } - private Props newShardProps(ShardIdentifier shardId, Map peers, String memberName) { - return EntityOwnershipShard.props(shardId, peers, dataStoreContextBuilder.build(), - SCHEMA_CONTEXT, memberName); + 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).ownerSelectionStrategyConfig(config).props().withDispatcher(Dispatchers.DefaultDispatcherId()); } - private ShardIdentifier newShardId(String memberName) { + private static ShardIdentifier newShardId(String memberName) { return ShardIdentifier.builder().memberName(memberName).shardName("entity-ownership"). type("operational" + NEXT_SHARD_NUM.getAndIncrement()).build(); } @@ -855,7 +829,8 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { TestEntityOwnershipShard(ShardIdentifier name, Map peerAddresses, DatastoreContext datastoreContext) { - super(name, peerAddresses, datastoreContext, SCHEMA_CONTEXT, LOCAL_MEMBER_NAME); + super(newBuilder().id(name).peerAddresses(peerAddresses).datastoreContext(datastoreContext). + schemaContext(SCHEMA_CONTEXT).localMemberName(LOCAL_MEMBER_NAME)); } @Override @@ -905,6 +880,36 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { } } + + @Test + public void testDelayedEntityOwnerSelection() throws Exception { + ShardTestKit kit = new ShardTestKit(getSystem()); + EntityOwnerSelectionStrategyConfig.Builder builder + = EntityOwnerSelectionStrategyConfig.newBuilder().addStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500); + TestActorRef shard = actorFactory.createTestActor(newShardProps(builder.build())); + kit.waitUntilLeader(shard); + + Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1); + ShardDataTree shardDataTree = shard.underlyingActor().getDataStore(); + + // Add a remote candidate + + String remoteMemberName1 = "remoteMember1"; + 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); + } + public static class MockLeader extends UntypedActor { volatile CountDownLatch modificationsReceived = new CountDownLatch(1); List receivedModifications = new ArrayList<>();