opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShardTest.java

   1 /*
   2  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.datastore.entityownership;
   9
  10 import static org.junit.Assert.assertEquals;
  11 import static org.mockito.AdditionalMatchers.or;
  12 import static org.mockito.Matchers.any;
  13 import static org.mockito.Mockito.mock;
  14 import static org.mockito.Mockito.never;
  15 import static org.mockito.Mockito.reset;
  16 import static org.mockito.Mockito.timeout;
  17 import static org.mockito.Mockito.times;
  18 import static org.mockito.Mockito.verify;
  19 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.ENTITY_OWNERS_PATH;
  20 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.candidatePath;
  21 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.entityOwnersWithCandidate;
  22 import akka.actor.ActorRef;
  23 import akka.actor.PoisonPill;
  24 import akka.actor.Props;
  25 import akka.actor.Terminated;
  26 import akka.actor.UntypedActor;
  27 import akka.dispatch.Dispatchers;
  28 import akka.testkit.JavaTestKit;
  29 import akka.testkit.TestActorRef;
  30 import com.google.common.base.Function;
  31 import com.google.common.collect.ImmutableMap;
  32 import com.google.common.util.concurrent.Uninterruptibles;
  33 import java.util.ArrayList;
  34 import java.util.Collections;
  35 import java.util.List;
  36 import java.util.Map;
  37 import java.util.concurrent.CountDownLatch;
  38 import java.util.concurrent.TimeUnit;
  39 import java.util.concurrent.atomic.AtomicInteger;
  40 import org.junit.After;
  41 import org.junit.Test;
  42 import org.opendaylight.controller.cluster.access.concepts.MemberName;
  43 import org.opendaylight.controller.cluster.datastore.AbstractShardTest;
  44 import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
  45 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
  46 import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
  47 import org.opendaylight.controller.cluster.datastore.ShardDataTree;
  48 import org.opendaylight.controller.cluster.datastore.ShardTestKit;
  49 import org.opendaylight.controller.cluster.datastore.entityownership.messages.RegisterCandidateLocal;
  50 import org.opendaylight.controller.cluster.datastore.entityownership.messages.RegisterListenerLocal;
  51 import org.opendaylight.controller.cluster.datastore.entityownership.messages.UnregisterCandidateLocal;
  52 import org.opendaylight.controller.cluster.datastore.entityownership.messages.UnregisterListenerLocal;
  53 import org.opendaylight.controller.cluster.datastore.entityownership.selectionstrategy.EntityOwnerSelectionStrategyConfig;
  54 import org.opendaylight.controller.cluster.datastore.entityownership.selectionstrategy.LastCandidateSelectionStrategy;
  55 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
  56 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
  57 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
  58 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
  59 import org.opendaylight.controller.cluster.datastore.messages.PeerDown;
  60 import org.opendaylight.controller.cluster.datastore.messages.PeerUp;
  61 import org.opendaylight.controller.cluster.datastore.messages.SuccessReply;
  62 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
  63 import org.opendaylight.controller.cluster.datastore.modification.Modification;
  64 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
  65 import org.opendaylight.controller.cluster.raft.TestActorFactory;
  66 import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
  67 import org.opendaylight.controller.cluster.raft.base.messages.TimeoutNow;
  68 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
  69 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
  70 import org.opendaylight.controller.cluster.raft.messages.RequestVote;
  71 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
  72 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  73 import org.opendaylight.mdsal.eos.dom.api.DOMEntity;
  74 import org.opendaylight.mdsal.eos.dom.api.DOMEntityOwnershipChange;
  75 import org.opendaylight.mdsal.eos.dom.api.DOMEntityOwnershipListener;
  76 import org.opendaylight.yangtools.yang.common.QName;
  77 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  78 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  79 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  80
  81 /**
  82  * Unit tests for EntityOwnershipShard.
  83  *
  84  * @author Thomas Pantelis
  85  */
  86 public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest {
  87     private static final String ENTITY_TYPE = "test type";
  88     private static final YangInstanceIdentifier ENTITY_ID1 =
  89             YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity1"));
  90     private static final YangInstanceIdentifier ENTITY_ID2 =
  91             YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity2"));
  92     private static final YangInstanceIdentifier ENTITY_ID3 =
  93             YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity3"));
  94     private static final YangInstanceIdentifier ENTITY_ID4 =
  95             YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity4"));
  96     private static final YangInstanceIdentifier ENTITY_ID5 =
  97             YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity5"));
  98     private static final SchemaContext SCHEMA_CONTEXT = SchemaContextHelper.entityOwners();
  99     private static final AtomicInteger NEXT_SHARD_NUM = new AtomicInteger();
 100     private static final String LOCAL_MEMBER_NAME = "member-1";
 101
 102     private final Builder dataStoreContextBuilder = DatastoreContext.newBuilder();
 103     private final TestActorFactory actorFactory = new TestActorFactory(getSystem());
 104
 105     @After
 106     public void tearDown() {
 107         actorFactory.close();
 108     }
 109
 110     @Test
 111     public void testOnRegisterCandidateLocal() throws Exception {
 112         ShardTestKit kit = new ShardTestKit(getSystem());
 113
 114         TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
 115
 116         ShardTestKit.waitUntilLeader(shard);
 117
 118         YangInstanceIdentifier entityId = ENTITY_ID1;
 119         DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId);
 120
 121         shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
 122         kit.expectMsgClass(SuccessReply.class);
 123
 124         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
 125         verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
 126     }
 127
 128     @Test
 129     public void testOnRegisterCandidateLocalWithNoInitialLeader() throws Exception {
 130         ShardTestKit kit = new ShardTestKit(getSystem());
 131
 132         dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2);
 133
 134         String peerId = newShardId("follower").toString();
 135         TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId, false).
 136                 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
 137
 138         TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
 139                 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
 140                 withDispatcher(Dispatchers.DefaultDispatcherId()));
 141
 142         YangInstanceIdentifier entityId = ENTITY_ID1;
 143         DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId);
 144
 145         shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
 146         kit.expectMsgClass(SuccessReply.class);
 147
 148         // Now grant the vote so the shard becomes the leader. This should retry the commit.
 149         peer.underlyingActor().grantVote = true;
 150
 151         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
 152         verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
 153     }
 154
 155     @Test
 156     public void testOnRegisterCandidateLocalWithNoInitialConsensus() throws Exception {
 157         ShardTestKit kit = new ShardTestKit(getSystem());
 158
 159         dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
 160                 shardTransactionCommitTimeoutInSeconds(1);
 161
 162         String peerId = newShardId("follower").toString();
 163         TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId).
 164                 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
 165
 166         MockFollower follower = peer.underlyingActor();
 167
 168         // Drop AppendEntries so consensus isn't reached.
 169         follower.dropAppendEntries = true;
 170
 171         TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
 172                 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
 173                 withDispatcher(Dispatchers.DefaultDispatcherId()));
 174
 175         ShardTestKit.waitUntilLeader(shard);
 176
 177         YangInstanceIdentifier entityId = ENTITY_ID1;
 178         DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId);
 179
 180         shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
 181         kit.expectMsgClass(SuccessReply.class);
 182
 183         // Wait enough time for the commit to timeout.
 184         Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
 185
 186         // Resume AppendEntries - the follower should ack the commit which should then result in the candidate
 187         // write being applied to the state.
 188         follower.dropAppendEntries = false;
 189
 190         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
 191         verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
 192     }
 193
 194     @Test
 195     public void testOnRegisterCandidateLocalWithIsolatedLeader() throws Exception {
 196         ShardTestKit kit = new ShardTestKit(getSystem());
 197
 198         dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
 199                 shardIsolatedLeaderCheckIntervalInMillis(50);
 200
 201         String peerId = newShardId("follower").toString();
 202         TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId).
 203                 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
 204
 205         MockFollower follower = peer.underlyingActor();
 206
 207         TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
 208                 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
 209                 withDispatcher(Dispatchers.DefaultDispatcherId()));
 210
 211         ShardTestKit.waitUntilLeader(shard);
 212
 213         // Drop AppendEntries and wait enough time for the shard to switch to IsolatedLeader.
 214         follower.dropAppendEntries = true;
 215         Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
 216
 217         YangInstanceIdentifier entityId = ENTITY_ID1;
 218         DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId);
 219
 220         shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
 221         kit.expectMsgClass(SuccessReply.class);
 222
 223         // Resume AppendEntries - the candidate write should now be committed.
 224         follower.dropAppendEntries = false;
 225         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
 226         verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
 227     }
 228
 229     @Test
 230     public void testOnRegisterCandidateLocalWithRemoteLeader() throws Exception {
 231         ShardTestKit kit = new ShardTestKit(getSystem());
 232
 233         dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
 234                 shardBatchedModificationCount(5);
 235
 236         String peerId = newShardId("leader").toString();
 237         TestActorRef<MockLeader> peer = actorFactory.createTestActor(Props.create(MockLeader.class).
 238                 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
 239
 240         TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(Props.create(
 241                 TestEntityOwnershipShard.class, newShardId(LOCAL_MEMBER_NAME),
 242                         ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build(),
 243                         dataStoreContextBuilder.build()).withDispatcher(Dispatchers.DefaultDispatcherId()));
 244
 245         shard.tell(new AppendEntries(1L, peerId, -1L, -1L, Collections.<ReplicatedLogEntry>emptyList(), -1L, -1L,
 246                 DataStoreVersions.CURRENT_VERSION), peer);
 247
 248         shard.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef());
 249         kit.expectMsgClass(SuccessReply.class);
 250
 251         MockLeader leader = peer.underlyingActor();
 252         assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
 253                 leader.modificationsReceived, 5, TimeUnit.SECONDS));
 254         verifyBatchedEntityCandidate(leader.getAndClearReceivedModifications(), ENTITY_TYPE, ENTITY_ID1,
 255                 LOCAL_MEMBER_NAME);
 256
 257         // Test with initial commit timeout and subsequent retry.
 258
 259         leader.modificationsReceived = new CountDownLatch(1);
 260         leader.sendReply = false;
 261
 262         shard.tell(dataStoreContextBuilder.shardTransactionCommitTimeoutInSeconds(1).build(), ActorRef.noSender());
 263
 264         shard.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, ENTITY_ID2)), kit.getRef());
 265         kit.expectMsgClass(SuccessReply.class);
 266
 267         assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
 268                 leader.modificationsReceived, 5, TimeUnit.SECONDS));
 269         verifyBatchedEntityCandidate(leader.getAndClearReceivedModifications(), ENTITY_TYPE, ENTITY_ID2,
 270                 LOCAL_MEMBER_NAME);
 271
 272         // Send a bunch of registration messages quickly and verify.
 273
 274         int max = 100;
 275         leader.delay = 4;
 276         leader.modificationsReceived = new CountDownLatch(max);
 277         List<YangInstanceIdentifier> entityIds = new ArrayList<>();
 278         for(int i = 1; i <= max; i++) {
 279             YangInstanceIdentifier id = YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "test" + i));
 280             entityIds.add(id);
 281             shard.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, id)), kit.getRef());
 282         }
 283
 284         assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
 285                 leader.modificationsReceived, 10, TimeUnit.SECONDS));
 286
 287         // Sleep a little to ensure no additional BatchedModifications are received.
 288
 289         Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
 290
 291         List<Modification> receivedMods = leader.getAndClearReceivedModifications();
 292         for(int i = 0; i < max; i++) {
 293             verifyBatchedEntityCandidate(receivedMods.get(i), ENTITY_TYPE, entityIds.get(i), LOCAL_MEMBER_NAME);
 294         }
 295
 296         assertEquals("# modifications received", max, receivedMods.size());
 297     }
 298
 299     @Test
 300     public void testOnUnregisterCandidateLocal() throws Exception {
 301         ShardTestKit kit = new ShardTestKit(getSystem());
 302         TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
 303         ShardTestKit.waitUntilLeader(shard);
 304
 305         DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
 306
 307         // Register
 308
 309         shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
 310         kit.expectMsgClass(SuccessReply.class);
 311
 312         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 313         verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 314
 315         // Unregister
 316
 317         shard.tell(new UnregisterCandidateLocal(entity), kit.getRef());
 318         kit.expectMsgClass(SuccessReply.class);
 319
 320         verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, "");
 321
 322         // Register again
 323
 324         shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
 325         kit.expectMsgClass(SuccessReply.class);
 326
 327         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 328         verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 329     }
 330
 331     @Test
 332     public void testOwnershipChanges() throws Exception {
 333         ShardTestKit kit = new ShardTestKit(getSystem());
 334         TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
 335         ShardTestKit.waitUntilLeader(shard);
 336
 337         DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
 338         ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
 339
 340         // Add a remote candidate
 341
 342         String remoteMemberName1 = "remoteMember1";
 343         writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
 344
 345         // Register local
 346
 347         shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
 348         kit.expectMsgClass(SuccessReply.class);
 349
 350         // Verify the remote candidate becomes owner
 351
 352         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
 353         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 354         verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
 355
 356         // Add another remote candidate and verify ownership doesn't change
 357
 358         String remoteMemberName2 = "remoteMember2";
 359         writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
 360
 361         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
 362         Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
 363         verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
 364
 365         // Remove the second remote candidate and verify ownership doesn't change
 366
 367         deleteNode(candidatePath(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
 368
 369         verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
 370         Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
 371         verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
 372
 373         // Remove the first remote candidate and verify the local candidate becomes owner
 374
 375         deleteNode(candidatePath(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
 376
 377         verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
 378         verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 379
 380         // Add the second remote candidate back and verify ownership doesn't change
 381
 382         writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
 383
 384         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
 385         Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
 386         verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 387
 388         // Unregister the local candidate and verify the second remote candidate becomes owner
 389
 390         shard.tell(new UnregisterCandidateLocal(entity), kit.getRef());
 391         kit.expectMsgClass(SuccessReply.class);
 392
 393         verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 394         verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
 395     }
 396
 397     @Test
 398     public void testOwnerChangesOnPeerAvailabilityChanges() throws Exception {
 399         ShardTestKit kit = new ShardTestKit(getSystem());
 400
 401         dataStoreContextBuilder.shardHeartbeatIntervalInMillis(500).shardElectionTimeoutFactor(10000);
 402
 403         String peerMemberName1 = "peerMember1";
 404         String peerMemberName2 = "peerMember2";
 405
 406         ShardIdentifier leaderId = newShardId(LOCAL_MEMBER_NAME);
 407         ShardIdentifier peerId1 = newShardId(peerMemberName1);
 408         ShardIdentifier peerId2 = newShardId(peerMemberName2);
 409
 410         TestActorRef<EntityOwnershipShard> peer1 = actorFactory.createTestActor(newShardProps(peerId1,
 411                 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId2.toString(), "").build(),
 412                         peerMemberName1, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
 413
 414         TestActorRef<EntityOwnershipShard> peer2 = actorFactory.createTestActor(newShardProps(peerId2,
 415                 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId1.toString(), "").build(),
 416                         peerMemberName2, EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString());
 417
 418         TestActorRef<EntityOwnershipShard> leader = actorFactory.createTestActor(newShardProps(leaderId,
 419                 ImmutableMap.<String, String>builder().put(peerId1.toString(), peer1.path().toString()).
 420                         put(peerId2.toString(), peer2.path().toString()).build(), LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build()).
 421                 withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString());
 422         leader.tell(TimeoutNow.INSTANCE, leader);
 423
 424         ShardTestKit.waitUntilLeader(leader);
 425
 426         // Send PeerDown and PeerUp with no entities
 427
 428         leader.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
 429         leader.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
 430
 431         // Add candidates for entity1 with the local leader as the owner
 432
 433         leader.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef());
 434         kit.expectMsgClass(SuccessReply.class);
 435         verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 436
 437         commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, peerMemberName2), kit);
 438         verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
 439
 440         commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, peerMemberName1), kit);
 441         verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName1);
 442         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 443
 444         // Add candidates for entity2 with peerMember2 as the owner
 445
 446         commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName2), kit);
 447         verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
 448
 449         commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName1), kit);
 450         verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
 451         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
 452
 453         // Add candidates for entity3 with peerMember2 as the owner.
 454
 455         commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName2), kit);
 456         verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
 457
 458         leader.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, ENTITY_ID3)), kit.getRef());
 459         kit.expectMsgClass(SuccessReply.class);
 460         verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
 461
 462         commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName1), kit);
 463         verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName1);
 464         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
 465
 466         // Add only candidate peerMember2 for entity4.
 467
 468         commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID4, peerMemberName2), kit);
 469         verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
 470         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
 471
 472         // Add only candidate peerMember1 for entity5.
 473
 474         commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID5, peerMemberName1), kit);
 475         verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID5, peerMemberName1);
 476         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID5, peerMemberName1);
 477
 478         // Kill peerMember2 and send PeerDown - the entities (2, 3, 4) owned by peerMember2 should get a new
 479         // owner selected
 480
 481         kit.watch(peer2);
 482         peer2.tell(PoisonPill.getInstance(), ActorRef.noSender());
 483         kit.expectMsgClass(JavaTestKit.duration("5 seconds"), Terminated.class);
 484         kit.unwatch(peer2);
 485
 486         leader.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
 487         // Send PeerDown again - should be noop
 488         leader.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
 489         peer1.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
 490
 491         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, ""); // no other candidates so should clear
 492         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
 493         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
 494         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 495
 496         verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
 497         verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
 498         verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
 499         verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
 500
 501         // Reinstate peerMember2 - no owners should change
 502
 503         peer2 = actorFactory.createTestActor(newShardProps(peerId2,
 504                 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId1.toString(), "").build(),
 505                         peerMemberName2, EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString());
 506         leader.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
 507         // Send PeerUp again - should be noop
 508         leader.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
 509         peer1.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
 510
 511         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
 512         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
 513         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 514         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, "");
 515
 516         // Add back candidate peerMember2 for entities 1, 2, & 3.
 517
 518         commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, peerMemberName2), kit);
 519         commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName2), kit);
 520         commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName2), kit);
 521         verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
 522         verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
 523         verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
 524         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 525         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
 526         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
 527
 528         // Kill peerMember1 and send PeerDown - entity 2 should get a new owner selected
 529
 530         peer1.tell(PoisonPill.getInstance(), ActorRef.noSender());
 531         leader.tell(new PeerDown(peerId1.getMemberName(), peerId1.toString()), ActorRef.noSender());
 532
 533         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
 534
 535         // Verify the reinstated peerMember2 is fully synced.
 536
 537         verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, "");
 538         verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
 539         verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
 540         verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 541
 542         // Reinstate peerMember1 and verify no owner changes
 543
 544         peer1 = actorFactory.createTestActor(newShardProps(peerId1,
 545                 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId2.toString(), "").build(),
 546                         peerMemberName1, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
 547         leader.tell(new PeerUp(peerId1.getMemberName(), peerId1.toString()), ActorRef.noSender());
 548
 549         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, "");
 550         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
 551         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
 552         verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 553
 554         // Verify the reinstated peerMember1 is fully synced.
 555
 556         verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID4, "");
 557         verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
 558         verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
 559         verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 560
 561         // Kill the local leader and elect peer2 the leader. This should cause a new owner to be selected for
 562         // the entities (1 and 3) previously owned by the local leader member.
 563
 564         peer2.tell(new PeerAddressResolved(peerId1.toString(), peer1.path().toString()), ActorRef.noSender());
 565         peer2.tell(new PeerUp(leaderId.getMemberName(), leaderId.toString()), ActorRef.noSender());
 566         peer2.tell(new PeerUp(peerId1.getMemberName(), peerId1.toString()), ActorRef.noSender());
 567
 568         leader.tell(PoisonPill.getInstance(), ActorRef.noSender());
 569         peer2.tell(new PeerDown(leaderId.getMemberName(), leaderId.toString()), ActorRef.noSender());
 570         peer2.tell(TimeoutNow.INSTANCE, peer2);
 571
 572         ShardTestKit.waitUntilLeader(peer2);
 573
 574         verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, "");
 575         verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
 576         verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
 577         verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
 578     }
 579
 580     @Test
 581     public void testLocalCandidateRemovedWithCandidateRegistered() throws Exception {
 582         ShardTestKit kit = new ShardTestKit(getSystem());
 583
 584         dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(10000);
 585         ShardIdentifier leaderId = newShardId("leader");
 586         ShardIdentifier localId = newShardId(LOCAL_MEMBER_NAME);
 587
 588         TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(Props.create(
 589                 TestEntityOwnershipShard.class, localId,
 590                 ImmutableMap.<String, String>builder().put(leaderId.toString(), "".toString()).build(),
 591                 dataStoreContextBuilder.build()).withDispatcher(Dispatchers.DefaultDispatcherId()));
 592
 593         TestActorRef<EntityOwnershipShard> leader = actorFactory.createTestActor(newShardProps(leaderId,
 594                 ImmutableMap.<String, String>builder().put(localId.toString(), shard.path().toString()).build(),
 595                     LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString());
 596         leader.tell(TimeoutNow.INSTANCE, leader);
 597
 598         ShardTestKit.waitUntilLeader(leader);
 599
 600         shard.tell(new PeerAddressResolved(leaderId.toString(), leader.path().toString()), ActorRef.noSender());
 601
 602         DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
 603         DOMEntityOwnershipListener listener = mock(DOMEntityOwnershipListener.class);
 604
 605         shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
 606         kit.expectMsgClass(SuccessReply.class);
 607
 608         // Register local candidate
 609
 610         shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
 611         kit.expectMsgClass(SuccessReply.class);
 612         verifyCommittedEntityCandidate(shard, entity.getType(), entity.getIdentifier(), LOCAL_MEMBER_NAME);
 613         verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true));
 614         reset(listener);
 615
 616         // Simulate a replicated commit from the leader to remove the local candidate that would occur after a
 617         // network partition is healed.
 618
 619         leader.tell(new PeerDown(localId.getMemberName(), localId.toString()), ActorRef.noSender());
 620
 621         verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, true, false, false));
 622
 623         // Since the the shard has a local candidate registered, it should re-add its candidate to the entity.
 624
 625         verifyCommittedEntityCandidate(shard, entity.getType(), entity.getIdentifier(), LOCAL_MEMBER_NAME);
 626         verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true));
 627
 628         // Unregister the local candidate and verify it's removed and no re-added.
 629
 630         shard.tell(new UnregisterCandidateLocal(entity), kit.getRef());
 631         kit.expectMsgClass(SuccessReply.class);
 632
 633         verifyNoEntityCandidate(shard, entity.getType(), entity.getIdentifier(), LOCAL_MEMBER_NAME);
 634         Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
 635         verifyNoEntityCandidate(shard, entity.getType(), entity.getIdentifier(), LOCAL_MEMBER_NAME);
 636     }
 637
 638     @Test
 639     public void testListenerRegistration() throws Exception {
 640         ShardTestKit kit = new ShardTestKit(getSystem());
 641         TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
 642         ShardTestKit.waitUntilLeader(shard);
 643         ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
 644
 645         String otherEntityType = "otherEntityType";
 646         DOMEntity entity1 = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
 647         DOMEntity entity2 = new DOMEntity(ENTITY_TYPE, ENTITY_ID2);
 648         DOMEntity entity3 = new DOMEntity(ENTITY_TYPE, ENTITY_ID3);
 649         DOMEntity entity4 = new DOMEntity(otherEntityType, ENTITY_ID3);
 650         DOMEntityOwnershipListener listener = mock(DOMEntityOwnershipListener.class);
 651
 652         // Register listener
 653
 654         shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
 655         kit.expectMsgClass(SuccessReply.class);
 656
 657         // Register a couple candidates for the desired entity type and verify listener is notified.
 658
 659         shard.tell(new RegisterCandidateLocal(entity1), kit.getRef());
 660         kit.expectMsgClass(SuccessReply.class);
 661
 662         verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, false, true, true));
 663
 664         shard.tell(new RegisterCandidateLocal(entity2), kit.getRef());
 665         kit.expectMsgClass(SuccessReply.class);
 666
 667         verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity2, false, true, true));
 668         reset(listener);
 669
 670         // Register another candidate for another entity type and verify listener is not notified.
 671
 672         shard.tell(new RegisterCandidateLocal(entity4), kit.getRef());
 673         kit.expectMsgClass(SuccessReply.class);
 674
 675         Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
 676         verify(listener, never()).ownershipChanged(ownershipChange(entity4));
 677
 678         // Register remote candidate for entity1
 679
 680         String remoteMemberName = "remoteMember";
 681         writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, entity1.getIdentifier(), remoteMemberName),
 682                 shardDataTree);
 683         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entity1.getIdentifier(), remoteMemberName);
 684
 685         // Unregister the local candidate for entity1 and verify listener is notified
 686
 687         shard.tell(new UnregisterCandidateLocal(entity1), kit.getRef());
 688         kit.expectMsgClass(SuccessReply.class);
 689
 690         verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, true, false, true));
 691         reset(listener);
 692
 693         // Unregister the listener, add a candidate for entity3 and verify listener isn't notified
 694
 695         shard.tell(new UnregisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
 696         kit.expectMsgClass(SuccessReply.class);
 697
 698         shard.tell(new RegisterCandidateLocal(entity3), kit.getRef());
 699         kit.expectMsgClass(SuccessReply.class);
 700
 701         verifyOwner(shard, ENTITY_TYPE, entity3.getIdentifier(), LOCAL_MEMBER_NAME);
 702         Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
 703         verify(listener, never()).ownershipChanged(any(DOMEntityOwnershipChange.class));
 704
 705         // Re-register the listener and verify it gets notified of currently owned entities
 706
 707         reset(listener);
 708
 709         shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
 710         kit.expectMsgClass(SuccessReply.class);
 711
 712         verify(listener, timeout(5000).times(2)).ownershipChanged(or(ownershipChange(entity2, false, true, true),
 713                 ownershipChange(entity3, false, true, true)));
 714         Uninterruptibles.sleepUninterruptibly(300, TimeUnit.MILLISECONDS);
 715         verify(listener, never()).ownershipChanged(ownershipChange(entity4));
 716         verify(listener, times(1)).ownershipChanged(ownershipChange(entity1));
 717     }
 718
 719     private static void commitModification(TestActorRef<EntityOwnershipShard> shard, NormalizedNode<?, ?> node,
 720             JavaTestKit sender) {
 721         BatchedModifications modifications = newBatchedModifications();
 722         modifications.addModification(new MergeModification(ENTITY_OWNERS_PATH, node));
 723
 724         shard.tell(modifications, sender.getRef());
 725         sender.expectMsgClass(CommitTransactionReply.class);
 726     }
 727
 728     private static BatchedModifications newBatchedModifications() {
 729         BatchedModifications modifications = new BatchedModifications(nextTransactionId(), DataStoreVersions.CURRENT_VERSION);
 730         modifications.setDoCommitOnReady(true);
 731         modifications.setReady(true);
 732         modifications.setTotalMessagesSent(1);
 733         return modifications;
 734     }
 735
 736     private void verifyEntityCandidateRemoved(final TestActorRef<EntityOwnershipShard> shard, String entityType,
 737             YangInstanceIdentifier entityId, String candidateName) {
 738         verifyNodeRemoved(candidatePath(entityType, entityId, candidateName),
 739                 new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
 740                     @Override
 741                     public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
 742                         try {
 743                             return AbstractShardTest.readStore(shard, path);
 744                         } catch(Exception e) {
 745                             throw new AssertionError("Failed to read " + path, e);
 746                         }
 747                 }
 748         });
 749     }
 750
 751     private void verifyCommittedEntityCandidate(final TestActorRef<EntityOwnershipShard> shard, String entityType,
 752             YangInstanceIdentifier entityId, String candidateName) {
 753         verifyEntityCandidate(entityType, entityId, candidateName, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
 754             @Override
 755             public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
 756                 try {
 757                     return AbstractShardTest.readStore(shard, path);
 758                 } catch(Exception e) {
 759                     throw new AssertionError("Failed to read " + path, e);
 760                 }
 761             }
 762         });
 763     }
 764
 765     private void verifyNoEntityCandidate(final TestActorRef<EntityOwnershipShard> shard, String entityType,
 766             YangInstanceIdentifier entityId, String candidateName) {
 767         verifyEntityCandidate(entityType, entityId, candidateName, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
 768             @Override
 769             public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
 770                 try {
 771                     return AbstractShardTest.readStore(shard, path);
 772                 } catch(Exception e) {
 773                     throw new AssertionError("Failed to read " + path, e);
 774                 }
 775             }
 776         }, false);
 777     }
 778
 779     private void verifyBatchedEntityCandidate(List<Modification> mods, String entityType,
 780             YangInstanceIdentifier entityId, String candidateName) throws Exception {
 781         assertEquals("BatchedModifications size", 1, mods.size());
 782         verifyBatchedEntityCandidate(mods.get(0), entityType, entityId, candidateName);
 783     }
 784
 785     private void verifyBatchedEntityCandidate(Modification mod, String entityType,
 786             YangInstanceIdentifier entityId, String candidateName) throws Exception {
 787         assertEquals("Modification type", MergeModification.class, mod.getClass());
 788         verifyEntityCandidate(((MergeModification)mod).getData(), entityType,
 789                 entityId, candidateName, true);
 790     }
 791
 792     private static void verifyOwner(final TestActorRef<EntityOwnershipShard> shard, String entityType,
 793             YangInstanceIdentifier entityId, String localMemberName) {
 794         verifyOwner(localMemberName, entityType, entityId, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
 795             @Override
 796             public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
 797                 try {
 798                     return AbstractShardTest.readStore(shard, path);
 799                 } catch(Exception e) {
 800                     return null;
 801                 }
 802             }
 803         });
 804     }
 805
 806     private Props newShardProps() {
 807         return newShardProps(Collections.<String,String>emptyMap());
 808     }
 809
 810     private Props newShardProps(EntityOwnerSelectionStrategyConfig strategyConfig, Map<String, String> peers) {
 811         return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, strategyConfig);
 812     }
 813
 814     private Props newShardProps(Map<String, String> peers) {
 815         return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build());
 816     }
 817
 818     private Props newShardProps(ShardIdentifier shardId, Map<String,String> peers, String memberName,
 819                                 EntityOwnerSelectionStrategyConfig config) {
 820         return EntityOwnershipShard.newBuilder().id(shardId).peerAddresses(peers).
 821                 datastoreContext(dataStoreContextBuilder.build()).schemaContext(SCHEMA_CONTEXT).
 822                 localMemberName(MemberName.forName(memberName)).ownerSelectionStrategyConfig(config).props()
 823                 .withDispatcher(Dispatchers.DefaultDispatcherId());
 824     }
 825
 826     private static ShardIdentifier newShardId(String memberName) {
 827         return ShardIdentifier.create("entity-ownership", MemberName.forName(memberName),
 828             "operational" + NEXT_SHARD_NUM.getAndIncrement());
 829     }
 830
 831     private static class TestEntityOwnershipShard extends EntityOwnershipShard {
 832
 833         TestEntityOwnershipShard(ShardIdentifier name, Map<String, String> peerAddresses,
 834                 DatastoreContext datastoreContext) {
 835             super(newBuilder().id(name).peerAddresses(peerAddresses).datastoreContext(datastoreContext).
 836                     schemaContext(SCHEMA_CONTEXT).localMemberName(MemberName.forName(LOCAL_MEMBER_NAME)));
 837         }
 838
 839         @Override
 840         public void handleCommand(Object message) {
 841             if(!(message instanceof ElectionTimeout)) {
 842                 super.handleCommand(message);
 843             }
 844         }
 845     }
 846
 847     private static class MockFollower extends UntypedActor {
 848         volatile boolean grantVote;
 849         volatile boolean dropAppendEntries;
 850         private final String myId;
 851
 852         @SuppressWarnings("unused")
 853         public MockFollower(String myId) {
 854             this(myId, true);
 855         }
 856
 857         public MockFollower(String myId, boolean grantVote) {
 858             this.myId = myId;
 859             this.grantVote = grantVote;
 860         }
 861
 862         @Override
 863         public void onReceive(Object message) {
 864             if(message instanceof RequestVote) {
 865                 if(grantVote) {
 866                     getSender().tell(new RequestVoteReply(((RequestVote)message).getTerm(), true), getSelf());
 867                 }
 868             } else if(message instanceof AppendEntries) {
 869                 if(!dropAppendEntries) {
 870                     AppendEntries req = (AppendEntries) message;
 871                     long lastIndex = req.getLeaderCommit();
 872                     if (req.getEntries().size() > 0) {
 873                         for(ReplicatedLogEntry entry : req.getEntries()) {
 874                             lastIndex = entry.getIndex();
 875                         }
 876                     }
 877
 878                     getSender().tell(new AppendEntriesReply(myId, req.getTerm(), true, lastIndex, req.getTerm(),
 879                             DataStoreVersions.CURRENT_VERSION), getSelf());
 880                 }
 881             }
 882         }
 883     }
 884
 885
 886     @Test
 887     public void testDelayedEntityOwnerSelectionWhenMaxPeerRequestsReceived() throws Exception {
 888         ShardTestKit kit = new ShardTestKit(getSystem());
 889         EntityOwnerSelectionStrategyConfig.Builder builder
 890                 = EntityOwnerSelectionStrategyConfig.newBuilder().addStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500);
 891
 892         String peerId = newShardId("follower").toString();
 893         TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId, false).
 894                 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
 895
 896         peer.underlyingActor().grantVote = true;
 897
 898         TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(builder.build(),
 899                 ImmutableMap.of(peerId.toString(), peer.path().toString())));
 900         ShardTestKit.waitUntilLeader(shard);
 901
 902         DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
 903         ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
 904
 905         // Add a remote candidate
 906
 907         String remoteMemberName1 = "follower";
 908         writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
 909
 910         // Register local
 911
 912         shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
 913         kit.expectMsgClass(SuccessReply.class);
 914
 915         // Verify the local candidate becomes owner
 916
 917         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
 918         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 919         verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 920     }
 921
 922     @Test
 923     public void testDelayedEntityOwnerSelection() throws Exception {
 924         ShardTestKit kit = new ShardTestKit(getSystem());
 925         EntityOwnerSelectionStrategyConfig.Builder builder
 926                 = EntityOwnerSelectionStrategyConfig.newBuilder().addStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500);
 927
 928         String follower1Id = newShardId("follower1").toString();
 929         TestActorRef<MockFollower> follower1 = actorFactory.createTestActor(Props.create(MockFollower.class, follower1Id, false).
 930                 withDispatcher(Dispatchers.DefaultDispatcherId()), follower1Id);
 931
 932         follower1.underlyingActor().grantVote = true;
 933
 934         String follower2Id = newShardId("follower").toString();
 935         TestActorRef<MockFollower> follower2 = actorFactory.createTestActor(Props.create(MockFollower.class, follower2Id, false).
 936                 withDispatcher(Dispatchers.DefaultDispatcherId()), follower2Id);
 937
 938         follower2.underlyingActor().grantVote = true;
 939
 940
 941         TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(builder.build(),
 942                 ImmutableMap.of(follower1Id.toString(), follower2.path().toString(), follower2Id.toString(), follower2.path().toString())));
 943         ShardTestKit.waitUntilLeader(shard);
 944
 945         DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
 946         ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
 947
 948         // Add a remote candidate
 949
 950         String remoteMemberName1 = "follower";
 951         writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
 952
 953         // Register local
 954
 955         shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
 956         kit.expectMsgClass(SuccessReply.class);
 957
 958         // Verify the local candidate becomes owner
 959
 960         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
 961         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 962         verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 963     }
 964
 965     private static class MockLeader extends UntypedActor {
 966         volatile CountDownLatch modificationsReceived = new CountDownLatch(1);
 967         List<Modification> receivedModifications = new ArrayList<>();
 968         volatile boolean sendReply = true;
 969         volatile long delay;
 970
 971         @Override
 972         public void onReceive(Object message) {
 973             if(message instanceof BatchedModifications) {
 974                 if(delay > 0) {
 975                     Uninterruptibles.sleepUninterruptibly(delay, TimeUnit.MILLISECONDS);
 976                 }
 977
 978                 if(sendReply) {
 979                     BatchedModifications mods = (BatchedModifications) message;
 980                     synchronized (receivedModifications) {
 981                         for(int i = 0; i < mods.getModifications().size(); i++) {
 982                             receivedModifications.add(mods.getModifications().get(i));
 983                             modificationsReceived.countDown();
 984                         }
 985                     }
 986
 987                     getSender().tell(CommitTransactionReply.instance(DataStoreVersions.CURRENT_VERSION).
 988                             toSerializable(), getSelf());
 989                 } else {
 990                     sendReply = true;
 991                 }
 992             }
 993         }
 994
 995         List<Modification> getAndClearReceivedModifications() {
 996             synchronized (receivedModifications) {
 997                 List<Modification> ret = new ArrayList<>(receivedModifications);
 998                 receivedModifications.clear();
 999                 return ret;
1000             }
1001         }
1002     }
1003 }