*/
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;
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;
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;
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";
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
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
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.
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
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
shard.tell(new AppendEntries(1L, peerId, -1L, -1L, Collections.<ReplicatedLogEntry>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();
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(
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(
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
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
// 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
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);
kit.waitUntilLeader(leader);
- EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class);
-
// Send PeerDown and PeerUp with no entities
leader.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender());
// 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);
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.
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);
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
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.<String, String>builder().put(leaderId.toString(), ""). put(peerId1.toString(), "").build(),
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
// 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);
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);
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<EntityOwnershipShard> shard = actorFactory.createTestActor(Props.create(
+ TestEntityOwnershipShard.class, localId,
+ ImmutableMap.<String, String>builder().put(leaderId.toString(), "".toString()).build(),
+ dataStoreContextBuilder.build()).withDispatcher(Dispatchers.DefaultDispatcherId()));
+
+ TestActorRef<EntityOwnershipShard> leader = actorFactory.createTestActor(newShardProps(leaderId,
+ ImmutableMap.<String, String>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());
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
// 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
// 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
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<Entity> 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<EntityOwnershipShard> 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<EntityOwnershipShard> shard, String entityType,
YangInstanceIdentifier entityId, String candidateName) {
verifyNodeRemoved(candidatePath(entityType, entityId, candidateName),
});
}
+ private void verifyNoEntityCandidate(final TestActorRef<EntityOwnershipShard> shard, String entityType,
+ YangInstanceIdentifier entityId, String candidateName) {
+ verifyEntityCandidate(entityType, entityId, candidateName, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
+ @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<Modification> mods, String entityType,
YangInstanceIdentifier entityId, String candidateName) throws Exception {
assertEquals("BatchedModifications size", 1, mods.size());
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<EntityOwnershipShard> shard, String entityType, YangInstanceIdentifier entityId,
Code Review / controller.git / blobdiff