import static org.mockito.Mockito.never;
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 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;
TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
- kit.waitUntilLeader(shard);
+ ShardTestKit.waitUntilLeader(shard);
YangInstanceIdentifier entityId = ENTITY_ID1;
Entity entity = new Entity(ENTITY_TYPE, entityId);
ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
withDispatcher(Dispatchers.DefaultDispatcherId()));
- kit.waitUntilLeader(shard);
+ ShardTestKit.waitUntilLeader(shard);
YangInstanceIdentifier entityId = ENTITY_ID1;
Entity entity = new Entity(ENTITY_TYPE, entityId);
ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
withDispatcher(Dispatchers.DefaultDispatcherId()));
- kit.waitUntilLeader(shard);
+ ShardTestKit.waitUntilLeader(shard);
// Drop AppendEntries and wait enough time for the shard to switch to IsolatedLeader.
follower.dropAppendEntries = true;
public void testOnUnregisterCandidateLocal() throws Exception {
ShardTestKit kit = new ShardTestKit(getSystem());
TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
- kit.waitUntilLeader(shard);
+ ShardTestKit.waitUntilLeader(shard);
Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
public void testOwnershipChanges() throws Exception {
ShardTestKit kit = new ShardTestKit(getSystem());
TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
- kit.waitUntilLeader(shard);
+ ShardTestKit.waitUntilLeader(shard);
Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
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(new ElectionTimeout(), leader);
+ leader.tell(ElectionTimeout.INSTANCE, leader);
- kit.waitUntilLeader(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
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);
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());
+ 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());
verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
// 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);
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());
+ 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);
// 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(new ElectionTimeout(), peer2);
+ peer2.tell(new PeerDown(leaderId.getMemberName(), leaderId.toString()), ActorRef.noSender());
+ peer2.tell(ElectionTimeout.INSTANCE, peer2);
- kit.waitUntilLeader(peer2);
+ ShardTestKit.waitUntilLeader(peer2);
verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, "");
verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
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(new ElectionTimeout(), leader);
+ leader.tell(ElectionTimeout.INSTANCE, leader);
- kit.waitUntilLeader(leader);
+ ShardTestKit.waitUntilLeader(leader);
shard.tell(new PeerAddressResolved(leaderId.toString(), leader.path().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());
+ leader.tell(new PeerDown(localId.getMemberName(), localId.toString()), ActorRef.noSender());
verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, true, false, false));
public void testListenerRegistration() throws Exception {
ShardTestKit kit = new ShardTestKit(getSystem());
TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
- kit.waitUntilLeader(shard);
+ ShardTestKit.waitUntilLeader(shard);
ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
String otherEntityType = "otherEntityType";
ownershipChange(entity3, false, true, true)));
Uninterruptibles.sleepUninterruptibly(300, TimeUnit.MILLISECONDS);
verify(listener, never()).ownershipChanged(ownershipChange(entity4));
- verify(listener, never()).ownershipChanged(ownershipChange(entity1));
+ verify(listener, times(1)).ownershipChanged(ownershipChange(entity1));
}
private static void commitModification(TestActorRef<EntityOwnershipShard> shard, NormalizedNode<?, ?> node,
modifications.addModification(new MergeModification(ENTITY_OWNERS_PATH, node));
shard.tell(modifications, sender.getRef());
- sender.expectMsgClass(CommitTransactionReply.SERIALIZABLE_CLASS);
+ sender.expectMsgClass(CommitTransactionReply.class);
}
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);
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(MemberName.forName(LOCAL_MEMBER_NAME)));
}
@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);
}
@Test
- public void testDelayedEntityOwnerSelection() throws Exception {
+ public void testDelayedEntityOwnerSelectionWhenMaxPeerRequestsReceived() throws Exception {
ShardTestKit kit = new ShardTestKit(getSystem());
EntityOwnerSelectionStrategyConfig.Builder builder
= EntityOwnerSelectionStrategyConfig.newBuilder().addStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500);
- TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(builder.build()));
- kit.waitUntilLeader(shard);
+
+ String peerId = newShardId("follower").toString();
+ TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId, false).
+ withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
+
+ peer.underlyingActor().grantVote = true;
+
+ TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(builder.build(),
+ ImmutableMap.of(peerId.toString(), peer.path().toString())));
+ ShardTestKit.waitUntilLeader(shard);
Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
// Add a remote candidate
- String remoteMemberName1 = "remoteMember1";
+ String remoteMemberName1 = "follower";
writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
+ // Register local
+
+ shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
+ kit.expectMsgClass(SuccessReply.class);
+
+ // Verify the local candidate becomes owner
+
+ verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
+ verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
+ verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
+ }
+
+ @Test
+ public void testDelayedEntityOwnerSelection() throws Exception {
+ ShardTestKit kit = new ShardTestKit(getSystem());
+ EntityOwnerSelectionStrategyConfig.Builder builder
+ = EntityOwnerSelectionStrategyConfig.newBuilder().addStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500);
+
+ String follower1Id = newShardId("follower1").toString();
+ TestActorRef<MockFollower> follower1 = actorFactory.createTestActor(Props.create(MockFollower.class, follower1Id, false).
+ withDispatcher(Dispatchers.DefaultDispatcherId()), follower1Id);
+
+ follower1.underlyingActor().grantVote = true;
+
+ String follower2Id = newShardId("follower").toString();
+ TestActorRef<MockFollower> follower2 = actorFactory.createTestActor(Props.create(MockFollower.class, follower2Id, false).
+ withDispatcher(Dispatchers.DefaultDispatcherId()), follower2Id);
+
+ follower2.underlyingActor().grantVote = true;
+
+
+ TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(builder.build(),
+ ImmutableMap.of(follower1Id.toString(), follower2.path().toString(), follower2Id.toString(), follower2.path().toString())));
+ ShardTestKit.waitUntilLeader(shard);
+
+ Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
+ ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
+
+ // Add a remote candidate
+
+ String remoteMemberName1 = "follower";
+ writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
// Register local
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;
}
}
- getSender().tell(CommitTransactionReply.INSTANCE.toSerializable(), getSelf());
+ getSender().tell(CommitTransactionReply.instance(DataStoreVersions.CURRENT_VERSION).
+ toSerializable(), getSelf());
} else {
sendReply = true;
}
Code Review / controller.git / blobdiff