import org.opendaylight.controller.cluster.datastore.entityownership.messages.RegisterListenerLocal;
import org.opendaylight.controller.cluster.datastore.entityownership.messages.UnregisterCandidateLocal;
import org.opendaylight.controller.cluster.datastore.entityownership.messages.UnregisterListenerLocal;
+import org.opendaylight.controller.cluster.datastore.entityownership.selectionstrategy.EntityOwnerSelectionStrategyConfig;
+import org.opendaylight.controller.cluster.datastore.entityownership.selectionstrategy.LastCandidateSelectionStrategy;
import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
TestActorRef<EntityOwnershipShard> peer1 = actorFactory.createTestActor(newShardProps(peerId1,
ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId2.toString(), "").build(),
- peerMemberName1).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
+ 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). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString());
+ 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).
+ put(peerId2.toString(), peer2.path().toString()).build(), LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build()).
withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString());
leader.tell(new ElectionTimeout(), leader);
peer2 = actorFactory.createTestActor(newShardProps(peerId2,
ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId1.toString(), "").build(),
- peerMemberName2). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString());
+ peerMemberName2, EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString());
leader.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender());
// Send PeerUp again - should be noop
leader.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender());
peer1 = actorFactory.createTestActor(newShardProps(peerId1,
ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId2.toString(), "").build(),
- peerMemberName1).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
+ peerMemberName1, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
leader.tell(new PeerUp(peerMemberName1, peerId1.toString()), ActorRef.noSender());
verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, "");
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());
+ LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString());
leader.tell(new ElectionTimeout(), leader);
kit.waitUntilLeader(leader);
verify(listener, never()).ownershipChanged(ownershipChange(entity1));
}
- private void commitModification(TestActorRef<EntityOwnershipShard> shard, NormalizedNode<?, ?> node,
+ private static void commitModification(TestActorRef<EntityOwnershipShard> shard, NormalizedNode<?, ?> node,
JavaTestKit sender) {
BatchedModifications modifications = newBatchedModifications();
modifications.addModification(new MergeModification(ENTITY_OWNERS_PATH, node));
sender.expectMsgClass(CommitTransactionReply.SERIALIZABLE_CLASS);
}
- private BatchedModifications newBatchedModifications() {
+ private static BatchedModifications newBatchedModifications() {
BatchedModifications modifications = new BatchedModifications("tnx", DataStoreVersions.CURRENT_VERSION, "");
modifications.setDoCommitOnReady(true);
modifications.setReady(true);
entityId, candidateName, true);
}
- private void verifyOwner(final TestActorRef<EntityOwnershipShard> shard, String entityType, YangInstanceIdentifier entityId,
- String localMemberName) {
+ 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) {
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(Map<String,String> peers) {
- return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME);
+ return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build());
}
- private Props newShardProps(ShardIdentifier shardId, Map<String,String> peers, String memberName) {
- return EntityOwnershipShard.props(shardId, peers, dataStoreContextBuilder.build(),
- SCHEMA_CONTEXT, memberName);
+ private Props newShardProps(ShardIdentifier shardId, Map<String,String> peers, String memberName,
+ EntityOwnerSelectionStrategyConfig config) {
+ return EntityOwnershipShard.newBuilder().id(shardId).peerAddresses(peers).
+ datastoreContext(dataStoreContextBuilder.build()).schemaContext(SCHEMA_CONTEXT).
+ localMemberName(memberName).ownerSelectionStrategyConfig(config).props().withDispatcher(Dispatchers.DefaultDispatcherId());
}
- private ShardIdentifier newShardId(String memberName) {
+ private static ShardIdentifier newShardId(String memberName) {
return ShardIdentifier.builder().memberName(memberName).shardName("entity-ownership").
type("operational" + NEXT_SHARD_NUM.getAndIncrement()).build();
}
TestEntityOwnershipShard(ShardIdentifier name, Map<String, String> peerAddresses,
DatastoreContext datastoreContext) {
- super(name, peerAddresses, datastoreContext, SCHEMA_CONTEXT, LOCAL_MEMBER_NAME);
+ super(newBuilder().id(name).peerAddresses(peerAddresses).datastoreContext(datastoreContext).
+ schemaContext(SCHEMA_CONTEXT).localMemberName(LOCAL_MEMBER_NAME));
}
@Override
}
}
+
+ @Test
+ public void testDelayedEntityOwnerSelection() 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);
+
+ Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
+ ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
+
+ // Add a remote candidate
+
+ String remoteMemberName1 = "remoteMember1";
+ 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);
+ }
+
public static class MockLeader extends UntypedActor {
volatile CountDownLatch modificationsReceived = new CountDownLatch(1);
List<Modification> receivedModifications = new ArrayList<>();