X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2Fentityownership%2FEntityOwnershipShardTest.java;h=e78be76e23cdf27f3c304a33d3fce6a51ed3e364;hp=146916a9621481846c05e832c29d440aa1b32b51;hb=16263ee5c532e75ca6f14ed5fa9053a38694af14;hpb=4f2d31367df3b74d326d4a7c69bd20789d020198 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 146916a962..e78be76e23 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 @@ -7,13 +7,9 @@ */ package org.opendaylight.controller.cluster.datastore.entityownership; -import static org.hamcrest.CoreMatchers.either; -import static org.hamcrest.CoreMatchers.equalTo; import static org.junit.Assert.assertEquals; +import static org.mockito.AdditionalMatchers.or; import static org.mockito.Matchers.any; -import static org.mockito.Matchers.anyBoolean; -import static org.mockito.Matchers.argThat; -import static org.mockito.Matchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.reset; @@ -40,7 +36,6 @@ import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; -import org.hamcrest.Matcher; import org.junit.After; import org.junit.Test; import org.opendaylight.controller.cluster.datastore.AbstractShardTest; @@ -71,7 +66,7 @@ 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; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; @@ -93,6 +88,8 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity3")); private static final YangInstanceIdentifier ENTITY_ID4 = YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity4")); + private static final YangInstanceIdentifier ENTITY_ID5 = + YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity5")); private static final SchemaContext SCHEMA_CONTEXT = SchemaContextHelper.entityOwners(); private static final AtomicInteger NEXT_SHARD_NUM = new AtomicInteger(); private static final String LOCAL_MEMBER_NAME = "member-1"; @@ -115,14 +112,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(entity, false, true); } @Test @@ -141,19 +136,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(entity, false, true); } @Test @@ -180,9 +171,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. @@ -193,10 +183,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(entity, false, true); } @Test @@ -224,18 +211,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(entity, false, true); } @Test @@ -257,9 +240,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(); @@ -275,7 +256,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( @@ -292,7 +273,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( @@ -317,35 +298,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(entity, false, 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, never()).ownershipChanged(any(Entity.class), anyBoolean(), anyBoolean()); // 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(entity, false, true); } @Test @@ -355,7 +330,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 @@ -365,7 +339,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 @@ -373,51 +347,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(Entity.class), anyBoolean(), anyBoolean()); // 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(Entity.class), anyBoolean(), anyBoolean()); // 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(Entity.class), anyBoolean(), anyBoolean()); // 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(entity, false, 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(Entity.class), anyBoolean(), anyBoolean()); // 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); @@ -453,8 +418,6 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { kit.waitUntilLeader(leader); - EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class); - // Send PeerDown and PeerUp with no entities leader.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender()); @@ -462,7 +425,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); @@ -480,10 +443,6 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName1), kit); verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1); - - leader.tell(new RegisterCandidateLocal(candidate, new Entity(ENTITY_TYPE, ENTITY_ID2)), kit.getRef()); - kit.expectMsgClass(SuccessReply.class); - verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, LOCAL_MEMBER_NAME); verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2); // Add candidates for entity3 with peerMember2 as the owner. @@ -491,7 +450,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); @@ -505,6 +464,12 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2); verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2); + // Add only candidate peerMember1 for entity5. + + commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID5, peerMemberName1), kit); + verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID5, peerMemberName1); + verifyOwner(leader, ENTITY_TYPE, ENTITY_ID5, peerMemberName1); + // Kill peerMember2 and send PeerDown - the entities (2, 3, 4) owned by peerMember2 should get a new // owner selected @@ -523,12 +488,12 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1); verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); - verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2); - verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2); - verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2); - verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2); + verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2); + verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2); + verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2); + verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2); - // Reinstate peerMember2 - should become owner again for entity 4 + // Reinstate peerMember2 - no owners should change peer2 = actorFactory.createTestActor(newShardProps(peerId2, ImmutableMap.builder().put(leaderId.toString(), ""). put(peerId1.toString(), "").build(), @@ -538,10 +503,22 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { leader.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender()); peer1.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender()); - verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2); verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME); verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1); verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); + verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, ""); + + // Add back candidate peerMember2 for entities 1, 2, & 3. + + commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, peerMemberName2), kit); + commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName2), kit); + commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName2), kit); + verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2); + verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2); + verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2); + verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); + verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1); + verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME); // Kill peerMember1 and send PeerDown - entity 2 should get a new owner selected @@ -552,7 +529,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { // Verify the reinstated peerMember2 is fully synced. - verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, peerMemberName2); + verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, ""); verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME); verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID2, peerMemberName2); verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); @@ -564,14 +541,14 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { peerMemberName1).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString()); leader.tell(new PeerUp(peerMemberName1, peerId1.toString()), ActorRef.noSender()); - verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2); + verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, ""); verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME); verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2); verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); // Verify the reinstated peerMember1 is fully synced. - verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID4, peerMemberName2); + verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID4, ""); verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME); verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID2, peerMemberName2); verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME); @@ -589,12 +566,70 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { kit.waitUntilLeader(peer2); - verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, peerMemberName2); + verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, ""); verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID3, peerMemberName2); verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID2, peerMemberName2); verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID1, peerMemberName2); } + @Test + public void testLocalCandidateRemovedWithCandidateRegistered() throws Exception { + ShardTestKit kit = new ShardTestKit(getSystem()); + + dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(10000); + ShardIdentifier leaderId = newShardId("leader"); + ShardIdentifier localId = newShardId(LOCAL_MEMBER_NAME); + + TestActorRef shard = actorFactory.createTestActor(Props.create( + TestEntityOwnershipShard.class, localId, + ImmutableMap.builder().put(leaderId.toString(), "".toString()).build(), + dataStoreContextBuilder.build()).withDispatcher(Dispatchers.DefaultDispatcherId())); + + 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); + + kit.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); + + shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef()); + kit.expectMsgClass(SuccessReply.class); + + // Register local candidate + + shard.tell(new RegisterCandidateLocal(entity), kit.getRef()); + kit.expectMsgClass(SuccessReply.class); + verifyCommittedEntityCandidate(shard, entity.getType(), entity.getId(), 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()); + + 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(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(entity), kit.getRef()); + kit.expectMsgClass(SuccessReply.class); + + verifyNoEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME); + Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS); + verifyNoEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME); + } + @Test public void testListenerRegistration() throws Exception { ShardTestKit kit = new ShardTestKit(getSystem()); @@ -608,7 +643,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 @@ -617,24 +651,24 @@ 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(entity1, false, true); + 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(entity2, false, true); + verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity2, false, true, true)); reset(listener); // 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); - verify(listener, never()).ownershipChanged(eq(entity4), anyBoolean(), anyBoolean()); + verify(listener, never()).ownershipChanged(ownershipChange(entity4)); // Register remote candidate for entity1 @@ -645,10 +679,10 @@ 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(entity1, true, false); + verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, true, false, true)); reset(listener); // Unregister the listener, add a candidate for entity3 and verify listener isn't notified @@ -656,41 +690,44 @@ 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); Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS); - verify(listener, never()).ownershipChanged(any(Entity.class), anyBoolean(), anyBoolean()); + verify(listener, never()).ownershipChanged(any(EntityOwnershipChange.class)); - // Re-register the listener and verify it gets notified of current locally owned entities + // 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); - Matcher entityMatcher = either(equalTo(entity2)).or(equalTo(entity3)); - verify(listener, timeout(5000).times(2)).ownershipChanged(argThat(entityMatcher), eq(false), eq(true)); + verify(listener, timeout(5000).times(2)).ownershipChanged(or(ownershipChange(entity2, false, true, true), + ownershipChange(entity3, false, true, true))); Uninterruptibles.sleepUninterruptibly(300, TimeUnit.MILLISECONDS); - verify(listener, never()).ownershipChanged(eq(entity4), anyBoolean(), anyBoolean()); - verify(listener, never()).ownershipChanged(eq(entity1), anyBoolean(), anyBoolean()); - verify(candidate, never()).ownershipChanged(eq(entity2), anyBoolean(), anyBoolean()); - verify(candidate, never()).ownershipChanged(eq(entity3), anyBoolean(), anyBoolean()); + verify(listener, never()).ownershipChanged(ownershipChange(entity4)); + verify(listener, never()).ownershipChanged(ownershipChange(entity1)); } private void commitModification(TestActorRef shard, NormalizedNode node, JavaTestKit sender) { - BatchedModifications modifications = new BatchedModifications("tnx", DataStoreVersions.CURRENT_VERSION, ""); - modifications.setDoCommitOnReady(true); - modifications.setReady(true); - modifications.setTotalMessagesSent(1); + BatchedModifications modifications = newBatchedModifications(); modifications.addModification(new MergeModification(ENTITY_OWNERS_PATH, node)); shard.tell(modifications, sender.getRef()); sender.expectMsgClass(CommitTransactionReply.SERIALIZABLE_CLASS); } + private BatchedModifications newBatchedModifications() { + BatchedModifications modifications = new BatchedModifications("tnx", DataStoreVersions.CURRENT_VERSION, ""); + modifications.setDoCommitOnReady(true); + modifications.setReady(true); + modifications.setTotalMessagesSent(1); + return modifications; + } + private void verifyEntityCandidateRemoved(final TestActorRef shard, String entityType, YangInstanceIdentifier entityId, String candidateName) { verifyNodeRemoved(candidatePath(entityType, entityId, candidateName), @@ -720,6 +757,20 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { }); } + private void verifyNoEntityCandidate(final TestActorRef shard, String entityType, + YangInstanceIdentifier entityId, String candidateName) { + verifyEntityCandidate(entityType, entityId, candidateName, new Function>() { + @Override + public NormalizedNode apply(YangInstanceIdentifier path) { + try { + return AbstractShardTest.readStore(shard, path); + } catch(Exception e) { + throw new AssertionError("Failed to read " + path, e); + } + } + }, false); + } + private void verifyBatchedEntityCandidate(List mods, String entityType, YangInstanceIdentifier entityId, String candidateName) throws Exception { assertEquals("BatchedModifications size", 1, mods.size()); @@ -730,7 +781,7 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest { YangInstanceIdentifier entityId, String candidateName) throws Exception { assertEquals("Modification type", MergeModification.class, mod.getClass()); verifyEntityCandidate(((MergeModification)mod).getData(), entityType, - entityId, candidateName); + entityId, candidateName, true); } private void verifyOwner(final TestActorRef shard, String entityType, YangInstanceIdentifier entityId,