import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.times;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.ENTITY_OWNERS_PATH;
import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.candidatePath;
import akka.dispatch.Dispatchers;
import akka.testkit.JavaTestKit;
import akka.testkit.TestActorRef;
-import com.google.common.base.Function;
import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.Uninterruptibles;
import java.util.ArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.After;
import org.junit.Test;
+import org.opendaylight.controller.cluster.access.concepts.MemberName;
import org.opendaylight.controller.cluster.datastore.AbstractShardTest;
import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
import org.opendaylight.controller.cluster.datastore.DatastoreContext;
import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
import org.opendaylight.controller.cluster.raft.TestActorFactory;
import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
+import org.opendaylight.controller.cluster.raft.base.messages.TimeoutNow;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
import org.opendaylight.controller.cluster.raft.messages.RequestVote;
import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
-import org.opendaylight.controller.md.sal.common.api.clustering.Entity;
-import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipChange;
-import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListener;
+import org.opendaylight.mdsal.eos.dom.api.DOMEntity;
+import org.opendaylight.mdsal.eos.dom.api.DOMEntityOwnershipChange;
+import org.opendaylight.mdsal.eos.dom.api.DOMEntityOwnershipListener;
import org.opendaylight.yangtools.yang.common.QName;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
ShardTestKit.waitUntilLeader(shard);
YangInstanceIdentifier entityId = ENTITY_ID1;
- Entity entity = new Entity(ENTITY_TYPE, entityId);
+ DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId);
shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
kit.expectMsgClass(SuccessReply.class);
withDispatcher(Dispatchers.DefaultDispatcherId()));
YangInstanceIdentifier entityId = ENTITY_ID1;
- Entity entity = new Entity(ENTITY_TYPE, entityId);
+ DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId);
shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
kit.expectMsgClass(SuccessReply.class);
ShardTestKit.waitUntilLeader(shard);
YangInstanceIdentifier entityId = ENTITY_ID1;
- Entity entity = new Entity(ENTITY_TYPE, entityId);
+ DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId);
shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
kit.expectMsgClass(SuccessReply.class);
Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
YangInstanceIdentifier entityId = ENTITY_ID1;
- Entity entity = new Entity(ENTITY_TYPE, entityId);
+ DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId);
shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
kit.expectMsgClass(SuccessReply.class);
shard.tell(new AppendEntries(1L, peerId, -1L, -1L, Collections.<ReplicatedLogEntry>emptyList(), -1L, -1L,
DataStoreVersions.CURRENT_VERSION), peer);
- shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef());
+ shard.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef());
kit.expectMsgClass(SuccessReply.class);
MockLeader leader = peer.underlyingActor();
shard.tell(dataStoreContextBuilder.shardTransactionCommitTimeoutInSeconds(1).build(), ActorRef.noSender());
- shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID2)), kit.getRef());
+ shard.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, ENTITY_ID2)), kit.getRef());
kit.expectMsgClass(SuccessReply.class);
assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
for(int i = 1; i <= max; i++) {
YangInstanceIdentifier id = YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "test" + i));
entityIds.add(id);
- shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, id)), kit.getRef());
+ shard.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, id)), kit.getRef());
}
assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
ShardTestKit.waitUntilLeader(shard);
- Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
+ DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
// Register
TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
ShardTestKit.waitUntilLeader(shard);
- Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
+ DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
// Add a remote candidate
ShardIdentifier peerId2 = newShardId(peerMemberName2);
TestActorRef<EntityOwnershipShard> peer1 = actorFactory.createTestActor(newShardProps(peerId1,
- ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId2.toString(), "").build(),
- peerMemberName1, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
+ ImmutableMap.<String, String>builder().put(leaderId.toString(), actorFactory.createTestActorPath(leaderId.toString())).
+ put(peerId2.toString(), actorFactory.createTestActorPath(peerId2.toString())).build(), peerMemberName1,
+ EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
TestActorRef<EntityOwnershipShard> peer2 = actorFactory.createTestActor(newShardProps(peerId2,
- ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId1.toString(), "").build(),
- peerMemberName2, EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString());
+ ImmutableMap.<String, String>builder().put(leaderId.toString(), actorFactory.createTestActorPath(leaderId.toString())).
+ put(peerId1.toString(), peer1.path().toString()).build(), peerMemberName2,
+ EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString());
TestActorRef<EntityOwnershipShard> leader = actorFactory.createTestActor(newShardProps(leaderId,
ImmutableMap.<String, String>builder().put(peerId1.toString(), peer1.path().toString()).
put(peerId2.toString(), peer2.path().toString()).build(), LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build()).
withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString());
- leader.tell(ElectionTimeout.INSTANCE, leader);
+ leader.tell(TimeoutNow.INSTANCE, leader);
ShardTestKit.waitUntilLeader(leader);
// Send PeerDown and PeerUp with no entities
- leader.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender());
- leader.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender());
+ leader.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
+ leader.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
// Add candidates for entity1 with the local leader as the owner
- leader.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef());
+ leader.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef());
kit.expectMsgClass(SuccessReply.class);
verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName2), kit);
verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
- leader.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID3)), kit.getRef());
+ leader.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, ENTITY_ID3)), kit.getRef());
kit.expectMsgClass(SuccessReply.class);
verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
kit.expectMsgClass(JavaTestKit.duration("5 seconds"), Terminated.class);
kit.unwatch(peer2);
- leader.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender());
+ leader.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
// Send PeerDown again - should be noop
- leader.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender());
- peer1.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender());
+ leader.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
+ peer1.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
- verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, ""); // no other candidates so should clear
- verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
- verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
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);
+ // no other candidates for entity4 so peerMember2 should remain owner.
+ verifyOwner(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);
+ 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);
- // Reinstate peerMember2 - no owners should change
+ // Reinstate peerMember2
peer2 = actorFactory.createTestActor(newShardProps(peerId2,
- ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId1.toString(), "").build(),
- peerMemberName2, EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString());
- leader.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender());
+ ImmutableMap.<String, String>builder().put(leaderId.toString(), leader.path().toString()).
+ put(peerId1.toString(), peer1.path().toString()).build(), peerMemberName2,
+ EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString());
+ leader.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
// Send PeerUp again - should be noop
- leader.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender());
- peer1.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender());
+ leader.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
+ peer1.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
+
+ // peerMember2's candidates should be removed on startup.
+ 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);
- 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_ID2, peerMemberName1);
+ verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, "");
// Add back candidate peerMember2 for entities 1, 2, & 3.
// Kill peerMember1 and send PeerDown - entity 2 should get a new owner selected
peer1.tell(PoisonPill.getInstance(), ActorRef.noSender());
- leader.tell(new PeerDown(peerMemberName1, peerId1.toString()), ActorRef.noSender());
+ leader.tell(new PeerDown(peerId1.getMemberName(), peerId1.toString()), ActorRef.noSender());
verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
// Verify the reinstated peerMember2 is fully synced.
- 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);
+ verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
+ verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
+ verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, "");
// Reinstate peerMember1 and verify no owner changes
peer1 = actorFactory.createTestActor(newShardProps(peerId1,
- ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId2.toString(), "").build(),
- peerMemberName1, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
- leader.tell(new PeerUp(peerMemberName1, peerId1.toString()), ActorRef.noSender());
+ ImmutableMap.<String, String>builder().put(leaderId.toString(), leader.path().toString()).
+ put(peerId2.toString(), peer2.path().toString()).build(), peerMemberName1,
+ EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
+ leader.tell(new PeerUp(peerId1.getMemberName(), peerId1.toString()), ActorRef.noSender());
- verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, "");
- verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
- verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
+ verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
+ verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
+ verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, "");
+
+ verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName1);
+ verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
+ verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName1);
// Verify the reinstated peerMember1 is fully synced.
- 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);
+ verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
+ verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
+ verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID4, "");
// Kill the local leader and elect peer2 the leader. This should cause a new owner to be selected for
// the entities (1 and 3) previously owned by the local leader member.
peer2.tell(new PeerAddressResolved(peerId1.toString(), peer1.path().toString()), ActorRef.noSender());
- peer2.tell(new PeerUp(LOCAL_MEMBER_NAME, leaderId.toString()), ActorRef.noSender());
- peer2.tell(new PeerUp(peerMemberName1, peerId1.toString()), ActorRef.noSender());
+ peer2.tell(new PeerUp(leaderId.getMemberName(), leaderId.toString()), ActorRef.noSender());
+ peer2.tell(new PeerUp(peerId1.getMemberName(), peerId1.toString()), ActorRef.noSender());
leader.tell(PoisonPill.getInstance(), ActorRef.noSender());
- peer2.tell(new PeerDown(LOCAL_MEMBER_NAME, leaderId.toString()), ActorRef.noSender());
- peer2.tell(ElectionTimeout.INSTANCE, peer2);
+ peer2.tell(new PeerDown(leaderId.getMemberName(), leaderId.toString()), ActorRef.noSender());
+ peer2.tell(TimeoutNow.INSTANCE, peer2);
ShardTestKit.waitUntilLeader(peer2);
- 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, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString());
- leader.tell(ElectionTimeout.INSTANCE, leader);
-
- ShardTestKit.waitUntilLeader(leader);
-
- shard.tell(new PeerAddressResolved(leaderId.toString(), leader.path().toString()), ActorRef.noSender());
-
- Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
- EntityOwnershipListener listener = mock(EntityOwnershipListener.class);
-
- 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);
+ verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
+ verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
+ verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, "");
}
@Test
ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
String otherEntityType = "otherEntityType";
- Entity entity1 = new Entity(ENTITY_TYPE, ENTITY_ID1);
- Entity entity2 = new Entity(ENTITY_TYPE, ENTITY_ID2);
- Entity entity3 = new Entity(ENTITY_TYPE, ENTITY_ID3);
- Entity entity4 = new Entity(otherEntityType, ENTITY_ID3);
- EntityOwnershipListener listener = mock(EntityOwnershipListener.class);
+ DOMEntity entity1 = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
+ DOMEntity entity2 = new DOMEntity(ENTITY_TYPE, ENTITY_ID2);
+ DOMEntity entity3 = new DOMEntity(ENTITY_TYPE, ENTITY_ID3);
+ DOMEntity entity4 = new DOMEntity(otherEntityType, ENTITY_ID3);
+ DOMEntityOwnershipListener listener = mock(DOMEntityOwnershipListener.class);
// Register listener
// Register remote candidate for entity1
String remoteMemberName = "remoteMember";
- writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, entity1.getId(), remoteMemberName),
+ writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, entity1.getIdentifier(), remoteMemberName),
shardDataTree);
- verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entity1.getId(), remoteMemberName);
+ verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entity1.getIdentifier(), remoteMemberName);
// Unregister the local candidate for entity1 and verify listener is notified
shard.tell(new RegisterCandidateLocal(entity3), kit.getRef());
kit.expectMsgClass(SuccessReply.class);
- verifyOwner(shard, ENTITY_TYPE, entity3.getId(), LOCAL_MEMBER_NAME);
+ verifyOwner(shard, ENTITY_TYPE, entity3.getIdentifier(), LOCAL_MEMBER_NAME);
Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
- verify(listener, never()).ownershipChanged(any(EntityOwnershipChange.class));
+ verify(listener, never()).ownershipChanged(any(DOMEntityOwnershipChange.class));
// Re-register the listener and verify it gets notified of currently owned entities
}
private static BatchedModifications newBatchedModifications() {
- BatchedModifications modifications = new BatchedModifications("tnx", DataStoreVersions.CURRENT_VERSION, "");
+ BatchedModifications modifications = new BatchedModifications(nextTransactionId(), DataStoreVersions.CURRENT_VERSION);
modifications.setDoCommitOnReady(true);
modifications.setReady(true);
modifications.setTotalMessagesSent(1);
private void verifyEntityCandidateRemoved(final TestActorRef<EntityOwnershipShard> shard, String entityType,
YangInstanceIdentifier entityId, String candidateName) {
verifyNodeRemoved(candidatePath(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);
- }
- }
- });
+ path -> {
+ try {
+ return AbstractShardTest.readStore(shard, path);
+ } catch(Exception e) {
+ throw new AssertionError("Failed to read " + path, e);
+ }
+ });
}
private void verifyCommittedEntityCandidate(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);
- }
+ verifyEntityCandidate(entityType, entityId, candidateName, path -> {
+ try {
+ return AbstractShardTest.readStore(shard, path);
+ } catch(Exception e) {
+ throw new AssertionError("Failed to read " + path, e);
}
});
}
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);
- }
+ verifyEntityCandidate(entityType, entityId, candidateName, path -> {
+ try {
+ return AbstractShardTest.readStore(shard, path);
+ } catch(Exception e) {
+ throw new AssertionError("Failed to read " + path, e);
}
}, false);
}
private static void verifyOwner(final TestActorRef<EntityOwnershipShard> shard, String entityType,
YangInstanceIdentifier entityId, String localMemberName) {
- verifyOwner(localMemberName, entityType, entityId, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
- @Override
- public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
- try {
- return AbstractShardTest.readStore(shard, path);
- } catch(Exception e) {
- return null;
- }
+ verifyOwner(localMemberName, entityType, entityId, path -> {
+ try {
+ return AbstractShardTest.readStore(shard, path);
+ } catch(Exception e) {
+ return null;
}
});
}
return newShardProps(Collections.<String,String>emptyMap());
}
- private Props newShardProps(EntityOwnerSelectionStrategyConfig strategyConfig) {
- return newShardProps(newShardId(LOCAL_MEMBER_NAME), Collections.<String,String>emptyMap(),
- LOCAL_MEMBER_NAME, strategyConfig);
- }
-
private Props newShardProps(EntityOwnerSelectionStrategyConfig strategyConfig, Map<String, String> peers) {
return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, strategyConfig);
}
-
- private Props newShardProps(Map<String,String> peers) {
+ private Props newShardProps(Map<String, String> peers) {
return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build());
}
EntityOwnerSelectionStrategyConfig config) {
return EntityOwnershipShard.newBuilder().id(shardId).peerAddresses(peers).
datastoreContext(dataStoreContextBuilder.build()).schemaContext(SCHEMA_CONTEXT).
- localMemberName(memberName).ownerSelectionStrategyConfig(config).props().withDispatcher(Dispatchers.DefaultDispatcherId());
+ localMemberName(MemberName.forName(memberName)).ownerSelectionStrategyConfig(config).props()
+ .withDispatcher(Dispatchers.DefaultDispatcherId());
}
private static ShardIdentifier newShardId(String memberName) {
- return ShardIdentifier.builder().memberName(memberName).shardName("entity-ownership").
- type("operational" + NEXT_SHARD_NUM.getAndIncrement()).build();
+ return ShardIdentifier.create("entity-ownership", MemberName.forName(memberName),
+ "operational" + NEXT_SHARD_NUM.getAndIncrement());
}
- public static class TestEntityOwnershipShard extends EntityOwnershipShard {
+ private static class TestEntityOwnershipShard extends EntityOwnershipShard {
TestEntityOwnershipShard(ShardIdentifier name, Map<String, String> peerAddresses,
DatastoreContext datastoreContext) {
super(newBuilder().id(name).peerAddresses(peerAddresses).datastoreContext(datastoreContext).
- schemaContext(SCHEMA_CONTEXT).localMemberName(LOCAL_MEMBER_NAME));
+ schemaContext(SCHEMA_CONTEXT).localMemberName(name.getMemberName()));
}
@Override
- public void onReceiveCommand(Object message) throws Exception {
+ public void handleCommand(Object message) {
if(!(message instanceof ElectionTimeout)) {
- super.onReceiveCommand(message);
+ super.handleCommand(message);
}
}
-
-
}
- public static class MockFollower extends UntypedActor {
+ private static class MockFollower extends UntypedActor {
volatile boolean grantVote;
volatile boolean dropAppendEntries;
private final String myId;
+ @SuppressWarnings("unused")
public MockFollower(String myId) {
this(myId, true);
}
ImmutableMap.of(peerId.toString(), peer.path().toString())));
ShardTestKit.waitUntilLeader(shard);
- Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
+ DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
// Add a remote candidate
ImmutableMap.of(follower1Id.toString(), follower2.path().toString(), follower2Id.toString(), follower2.path().toString())));
ShardTestKit.waitUntilLeader(shard);
- Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
+ DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
// Add a remote candidate
verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
}
- public static class MockLeader extends UntypedActor {
+ private static class MockLeader extends UntypedActor {
volatile CountDownLatch modificationsReceived = new CountDownLatch(1);
List<Modification> receivedModifications = new ArrayList<>();
volatile boolean sendReply = true;
Code Review / controller.git / blobdiff