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.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.yangtools.yang.common.QName;
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;
YangInstanceIdentifier entityId = ENTITY_ID1;
Entity entity = new Entity(ENTITY_TYPE, entityId);
YangInstanceIdentifier entityId = ENTITY_ID1;
Entity entity = new Entity(ENTITY_TYPE, entityId);
kit.expectMsgClass(SuccessReply.class);
verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
kit.expectMsgClass(SuccessReply.class);
verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
YangInstanceIdentifier entityId = ENTITY_ID1;
Entity entity = new Entity(ENTITY_TYPE, entityId);
YangInstanceIdentifier entityId = ENTITY_ID1;
Entity entity = new Entity(ENTITY_TYPE, entityId);
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);
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);
verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
YangInstanceIdentifier entityId = ENTITY_ID1;
Entity entity = new Entity(ENTITY_TYPE, entityId);
YangInstanceIdentifier entityId = ENTITY_ID1;
Entity entity = new Entity(ENTITY_TYPE, entityId);
follower.dropAppendEntries = false;
verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
follower.dropAppendEntries = false;
verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
YangInstanceIdentifier entityId = ENTITY_ID1;
Entity entity = new Entity(ENTITY_TYPE, entityId);
YangInstanceIdentifier entityId = ENTITY_ID1;
Entity entity = new Entity(ENTITY_TYPE, entityId);
kit.expectMsgClass(SuccessReply.class);
// Resume AppendEntries - the candidate write should now be committed.
follower.dropAppendEntries = false;
verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
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);
verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
shard.tell(new AppendEntries(1L, peerId, -1L, -1L, Collections.<ReplicatedLogEntry>emptyList(), -1L, -1L,
DataStoreVersions.CURRENT_VERSION), peer);
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());
- 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(
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);
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());
kit.waitUntilLeader(shard);
Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
kit.waitUntilLeader(shard);
Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
kit.expectMsgClass(SuccessReply.class);
verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
kit.expectMsgClass(SuccessReply.class);
verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
kit.expectMsgClass(SuccessReply.class);
verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, "");
kit.expectMsgClass(SuccessReply.class);
verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, "");
kit.expectMsgClass(SuccessReply.class);
verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
kit.expectMsgClass(SuccessReply.class);
verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
kit.waitUntilLeader(shard);
Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
kit.waitUntilLeader(shard);
Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
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);
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);
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);
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);
deleteNode(candidatePath(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
deleteNode(candidatePath(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
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);
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);
kit.expectMsgClass(SuccessReply.class);
verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
kit.expectMsgClass(SuccessReply.class);
verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
// Send PeerDown and PeerUp with no entities
leader.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender());
// Send PeerDown and PeerUp with no entities
leader.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender());
- 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);
kit.expectMsgClass(SuccessReply.class);
verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName2), kit);
verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
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);
kit.expectMsgClass(SuccessReply.class);
verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
shard.tell(new PeerAddressResolved(leaderId.toString(), leader.path().toString()), ActorRef.noSender());
Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
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);
kit.expectMsgClass(SuccessReply.class);
verifyCommittedEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME);
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());
// 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);
// 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));
kit.expectMsgClass(SuccessReply.class);
verifyNoEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME);
kit.expectMsgClass(SuccessReply.class);
verifyNoEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME);
Entity entity3 = new Entity(ENTITY_TYPE, ENTITY_ID3);
Entity entity4 = new Entity(otherEntityType, ENTITY_ID3);
EntityOwnershipListener listener = mock(EntityOwnershipListener.class);
Entity entity3 = new Entity(ENTITY_TYPE, ENTITY_ID3);
Entity entity4 = new Entity(otherEntityType, ENTITY_ID3);
EntityOwnershipListener listener = mock(EntityOwnershipListener.class);
kit.expectMsgClass(SuccessReply.class);
verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, false, true, true));
kit.expectMsgClass(SuccessReply.class);
verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, false, true, true));
kit.expectMsgClass(SuccessReply.class);
verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity2, false, true, true));
kit.expectMsgClass(SuccessReply.class);
verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity2, false, true, true));
kit.expectMsgClass(SuccessReply.class);
Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
kit.expectMsgClass(SuccessReply.class);
Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
kit.expectMsgClass(SuccessReply.class);
verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, true, false, true));
kit.expectMsgClass(SuccessReply.class);
verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, true, false, true));
shard.tell(new UnregisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
kit.expectMsgClass(SuccessReply.class);
shard.tell(new UnregisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
kit.expectMsgClass(SuccessReply.class);
kit.expectMsgClass(SuccessReply.class);
verifyOwner(shard, ENTITY_TYPE, entity3.getId(), LOCAL_MEMBER_NAME);
kit.expectMsgClass(SuccessReply.class);
verifyOwner(shard, ENTITY_TYPE, entity3.getId(), LOCAL_MEMBER_NAME);
shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
kit.expectMsgClass(SuccessReply.class);
shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
kit.expectMsgClass(SuccessReply.class);
Uninterruptibles.sleepUninterruptibly(300, TimeUnit.MILLISECONDS);
verify(listener, never()).ownershipChanged(ownershipChange(entity4));
verify(listener, never()).ownershipChanged(ownershipChange(entity1));
Uninterruptibles.sleepUninterruptibly(300, TimeUnit.MILLISECONDS);
verify(listener, never()).ownershipChanged(ownershipChange(entity4));
verify(listener, never()).ownershipChanged(ownershipChange(entity1));
}
private void commitModification(TestActorRef<EntityOwnershipShard> shard, NormalizedNode<?, ?> node,
}
private void commitModification(TestActorRef<EntityOwnershipShard> shard, NormalizedNode<?, ?> node,