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                     leaderId.getMemberName().getName(), EntityOwnerSelectionStrategyConfig.newBuilder().build())
 596                 .withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString());
 597         leader.tell(TimeoutNow.INSTANCE, leader);
 598
 599         ShardTestKit.waitUntilLeader(leader);
 600
 601         shard.tell(new PeerAddressResolved(leaderId.toString(), leader.path().toString()), ActorRef.noSender());
 602
 603         DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
 604         DOMEntityOwnershipListener listener = mock(DOMEntityOwnershipListener.class);
 605
 606         shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
 607         kit.expectMsgClass(SuccessReply.class);
 608
 609         // Register local candidate
 610
 611         shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
 612         kit.expectMsgClass(SuccessReply.class);
 613         verifyCommittedEntityCandidate(shard, entity.getType(), entity.getIdentifier(), LOCAL_MEMBER_NAME);
 614         verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true));
 615         reset(listener);
 616
 617         // Simulate a replicated commit from the leader to remove the local candidate that would occur after a
 618         // network partition is healed.
 619
 620         leader.tell(new PeerDown(localId.getMemberName(), localId.toString()), ActorRef.noSender());
 621
 622         verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, true, false, false));
 623
 624         // Since the the shard has a local candidate registered, it should re-add its candidate to the entity.
 625
 626         verifyCommittedEntityCandidate(shard, entity.getType(), entity.getIdentifier(), LOCAL_MEMBER_NAME);
 627         verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true));
 628
 629         // Unregister the local candidate and verify it's removed and no re-added.
 630
 631         shard.tell(new UnregisterCandidateLocal(entity), kit.getRef());
 632         kit.expectMsgClass(SuccessReply.class);
 633
 634         verifyNoEntityCandidate(shard, entity.getType(), entity.getIdentifier(), LOCAL_MEMBER_NAME);
 635         Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
 636         verifyNoEntityCandidate(shard, entity.getType(), entity.getIdentifier(), LOCAL_MEMBER_NAME);
 637     }
 638
 639     @Test
 640     public void testListenerRegistration() throws Exception {
 641         ShardTestKit kit = new ShardTestKit(getSystem());
 642         TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
 643         ShardTestKit.waitUntilLeader(shard);
 644         ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
 645
 646         String otherEntityType = "otherEntityType";
 647         DOMEntity entity1 = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
 648         DOMEntity entity2 = new DOMEntity(ENTITY_TYPE, ENTITY_ID2);
 649         DOMEntity entity3 = new DOMEntity(ENTITY_TYPE, ENTITY_ID3);
 650         DOMEntity entity4 = new DOMEntity(otherEntityType, ENTITY_ID3);
 651         DOMEntityOwnershipListener listener = mock(DOMEntityOwnershipListener.class);
 652
 653         // Register listener
 654
 655         shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
 656         kit.expectMsgClass(SuccessReply.class);
 657
 658         // Register a couple candidates for the desired entity type and verify listener is notified.
 659
 660         shard.tell(new RegisterCandidateLocal(entity1), kit.getRef());
 661         kit.expectMsgClass(SuccessReply.class);
 662
 663         verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, false, true, true));
 664
 665         shard.tell(new RegisterCandidateLocal(entity2), kit.getRef());
 666         kit.expectMsgClass(SuccessReply.class);
 667
 668         verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity2, false, true, true));
 669         reset(listener);
 670
 671         // Register another candidate for another entity type and verify listener is not notified.
 672
 673         shard.tell(new RegisterCandidateLocal(entity4), kit.getRef());
 674         kit.expectMsgClass(SuccessReply.class);
 675
 676         Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
 677         verify(listener, never()).ownershipChanged(ownershipChange(entity4));
 678
 679         // Register remote candidate for entity1
 680
 681         String remoteMemberName = "remoteMember";
 682         writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, entity1.getIdentifier(), remoteMemberName),
 683                 shardDataTree);
 684         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entity1.getIdentifier(), remoteMemberName);
 685
 686         // Unregister the local candidate for entity1 and verify listener is notified
 687
 688         shard.tell(new UnregisterCandidateLocal(entity1), kit.getRef());
 689         kit.expectMsgClass(SuccessReply.class);
 690
 691         verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, true, false, true));
 692         reset(listener);
 693
 694         // Unregister the listener, add a candidate for entity3 and verify listener isn't notified
 695
 696         shard.tell(new UnregisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
 697         kit.expectMsgClass(SuccessReply.class);
 698
 699         shard.tell(new RegisterCandidateLocal(entity3), kit.getRef());
 700         kit.expectMsgClass(SuccessReply.class);
 701
 702         verifyOwner(shard, ENTITY_TYPE, entity3.getIdentifier(), LOCAL_MEMBER_NAME);
 703         Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
 704         verify(listener, never()).ownershipChanged(any(DOMEntityOwnershipChange.class));
 705
 706         // Re-register the listener and verify it gets notified of currently owned entities
 707
 708         reset(listener);
 709
 710         shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
 711         kit.expectMsgClass(SuccessReply.class);
 712
 713         verify(listener, timeout(5000).times(2)).ownershipChanged(or(ownershipChange(entity2, false, true, true),
 714                 ownershipChange(entity3, false, true, true)));
 715         Uninterruptibles.sleepUninterruptibly(300, TimeUnit.MILLISECONDS);
 716         verify(listener, never()).ownershipChanged(ownershipChange(entity4));
 717         verify(listener, times(1)).ownershipChanged(ownershipChange(entity1));
 718     }
 719
 720     private static void commitModification(TestActorRef<EntityOwnershipShard> shard, NormalizedNode<?, ?> node,
 721             JavaTestKit sender) {
 722         BatchedModifications modifications = newBatchedModifications();
 723         modifications.addModification(new MergeModification(ENTITY_OWNERS_PATH, node));
 724
 725         shard.tell(modifications, sender.getRef());
 726         sender.expectMsgClass(CommitTransactionReply.class);
 727     }
 728
 729     private static BatchedModifications newBatchedModifications() {
 730         BatchedModifications modifications = new BatchedModifications(nextTransactionId(), DataStoreVersions.CURRENT_VERSION);
 731         modifications.setDoCommitOnReady(true);
 732         modifications.setReady(true);
 733         modifications.setTotalMessagesSent(1);
 734         return modifications;
 735     }
 736
 737     private void verifyEntityCandidateRemoved(final TestActorRef<EntityOwnershipShard> shard, String entityType,
 738             YangInstanceIdentifier entityId, String candidateName) {
 739         verifyNodeRemoved(candidatePath(entityType, entityId, candidateName),
 740                 new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
 741                     @Override
 742                     public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
 743                         try {
 744                             return AbstractShardTest.readStore(shard, path);
 745                         } catch(Exception e) {
 746                             throw new AssertionError("Failed to read " + path, e);
 747                         }
 748                 }
 749         });
 750     }
 751
 752     private void verifyCommittedEntityCandidate(final TestActorRef<EntityOwnershipShard> shard, String entityType,
 753             YangInstanceIdentifier entityId, String candidateName) {
 754         verifyEntityCandidate(entityType, entityId, candidateName, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
 755             @Override
 756             public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
 757                 try {
 758                     return AbstractShardTest.readStore(shard, path);
 759                 } catch(Exception e) {
 760                     throw new AssertionError("Failed to read " + path, e);
 761                 }
 762             }
 763         });
 764     }
 765
 766     private void verifyNoEntityCandidate(final TestActorRef<EntityOwnershipShard> shard, String entityType,
 767             YangInstanceIdentifier entityId, String candidateName) {
 768         verifyEntityCandidate(entityType, entityId, candidateName, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
 769             @Override
 770             public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
 771                 try {
 772                     return AbstractShardTest.readStore(shard, path);
 773                 } catch(Exception e) {
 774                     throw new AssertionError("Failed to read " + path, e);
 775                 }
 776             }
 777         }, false);
 778     }
 779
 780     private void verifyBatchedEntityCandidate(List<Modification> mods, String entityType,
 781             YangInstanceIdentifier entityId, String candidateName) throws Exception {
 782         assertEquals("BatchedModifications size", 1, mods.size());
 783         verifyBatchedEntityCandidate(mods.get(0), entityType, entityId, candidateName);
 784     }
 785
 786     private void verifyBatchedEntityCandidate(Modification mod, String entityType,
 787             YangInstanceIdentifier entityId, String candidateName) throws Exception {
 788         assertEquals("Modification type", MergeModification.class, mod.getClass());
 789         verifyEntityCandidate(((MergeModification)mod).getData(), entityType,
 790                 entityId, candidateName, true);
 791     }
 792
 793     private static void verifyOwner(final TestActorRef<EntityOwnershipShard> shard, String entityType,
 794             YangInstanceIdentifier entityId, String localMemberName) {
 795         verifyOwner(localMemberName, entityType, entityId, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
 796             @Override
 797             public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
 798                 try {
 799                     return AbstractShardTest.readStore(shard, path);
 800                 } catch(Exception e) {
 801                     return null;
 802                 }
 803             }
 804         });
 805     }
 806
 807     private Props newShardProps() {
 808         return newShardProps(Collections.<String,String>emptyMap());
 809     }
 810
 811     private Props newShardProps(EntityOwnerSelectionStrategyConfig strategyConfig, Map<String, String> peers) {
 812         return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, strategyConfig);
 813     }
 814
 815     private Props newShardProps(Map<String, String> peers) {
 816         return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build());
 817     }
 818
 819     private Props newShardProps(ShardIdentifier shardId, Map<String,String> peers, String memberName,
 820                                 EntityOwnerSelectionStrategyConfig config) {
 821         return EntityOwnershipShard.newBuilder().id(shardId).peerAddresses(peers).
 822                 datastoreContext(dataStoreContextBuilder.build()).schemaContext(SCHEMA_CONTEXT).
 823                 localMemberName(MemberName.forName(memberName)).ownerSelectionStrategyConfig(config).props()
 824                 .withDispatcher(Dispatchers.DefaultDispatcherId());
 825     }
 826
 827     private static ShardIdentifier newShardId(String memberName) {
 828         return ShardIdentifier.create("entity-ownership", MemberName.forName(memberName),
 829             "operational" + NEXT_SHARD_NUM.getAndIncrement());
 830     }
 831
 832     private static class TestEntityOwnershipShard extends EntityOwnershipShard {
 833
 834         TestEntityOwnershipShard(ShardIdentifier name, Map<String, String> peerAddresses,
 835                 DatastoreContext datastoreContext) {
 836             super(newBuilder().id(name).peerAddresses(peerAddresses).datastoreContext(datastoreContext).
 837                     schemaContext(SCHEMA_CONTEXT).localMemberName(name.getMemberName()));
 838         }
 839
 840         @Override
 841         public void handleCommand(Object message) {
 842             if(!(message instanceof ElectionTimeout)) {
 843                 super.handleCommand(message);
 844             }
 845         }
 846     }
 847
 848     private static class MockFollower extends UntypedActor {
 849         volatile boolean grantVote;
 850         volatile boolean dropAppendEntries;
 851         private final String myId;
 852
 853         @SuppressWarnings("unused")
 854         public MockFollower(String myId) {
 855             this(myId, true);
 856         }
 857
 858         public MockFollower(String myId, boolean grantVote) {
 859             this.myId = myId;
 860             this.grantVote = grantVote;
 861         }
 862
 863         @Override
 864         public void onReceive(Object message) {
 865             if(message instanceof RequestVote) {
 866                 if(grantVote) {
 867                     getSender().tell(new RequestVoteReply(((RequestVote)message).getTerm(), true), getSelf());
 868                 }
 869             } else if(message instanceof AppendEntries) {
 870                 if(!dropAppendEntries) {
 871                     AppendEntries req = (AppendEntries) message;
 872                     long lastIndex = req.getLeaderCommit();
 873                     if (req.getEntries().size() > 0) {
 874                         for(ReplicatedLogEntry entry : req.getEntries()) {
 875                             lastIndex = entry.getIndex();
 876                         }
 877                     }
 878
 879                     getSender().tell(new AppendEntriesReply(myId, req.getTerm(), true, lastIndex, req.getTerm(),
 880                             DataStoreVersions.CURRENT_VERSION), getSelf());
 881                 }
 882             }
 883         }
 884     }
 885
 886
 887     @Test
 888     public void testDelayedEntityOwnerSelectionWhenMaxPeerRequestsReceived() throws Exception {
 889         ShardTestKit kit = new ShardTestKit(getSystem());
 890         EntityOwnerSelectionStrategyConfig.Builder builder
 891                 = EntityOwnerSelectionStrategyConfig.newBuilder().addStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500);
 892
 893         String peerId = newShardId("follower").toString();
 894         TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId, false).
 895                 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
 896
 897         peer.underlyingActor().grantVote = true;
 898
 899         TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(builder.build(),
 900                 ImmutableMap.of(peerId.toString(), peer.path().toString())));
 901         ShardTestKit.waitUntilLeader(shard);
 902
 903         DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
 904         ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
 905
 906         // Add a remote candidate
 907
 908         String remoteMemberName1 = "follower";
 909         writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
 910
 911         // Register local
 912
 913         shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
 914         kit.expectMsgClass(SuccessReply.class);
 915
 916         // Verify the local candidate becomes owner
 917
 918         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
 919         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 920         verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 921     }
 922
 923     @Test
 924     public void testDelayedEntityOwnerSelection() throws Exception {
 925         ShardTestKit kit = new ShardTestKit(getSystem());
 926         EntityOwnerSelectionStrategyConfig.Builder builder
 927                 = EntityOwnerSelectionStrategyConfig.newBuilder().addStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500);
 928
 929         String follower1Id = newShardId("follower1").toString();
 930         TestActorRef<MockFollower> follower1 = actorFactory.createTestActor(Props.create(MockFollower.class, follower1Id, false).
 931                 withDispatcher(Dispatchers.DefaultDispatcherId()), follower1Id);
 932
 933         follower1.underlyingActor().grantVote = true;
 934
 935         String follower2Id = newShardId("follower").toString();
 936         TestActorRef<MockFollower> follower2 = actorFactory.createTestActor(Props.create(MockFollower.class, follower2Id, false).
 937                 withDispatcher(Dispatchers.DefaultDispatcherId()), follower2Id);
 938
 939         follower2.underlyingActor().grantVote = true;
 940
 941
 942         TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(builder.build(),
 943                 ImmutableMap.of(follower1Id.toString(), follower2.path().toString(), follower2Id.toString(), follower2.path().toString())));
 944         ShardTestKit.waitUntilLeader(shard);
 945
 946         DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
 947         ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
 948
 949         // Add a remote candidate
 950
 951         String remoteMemberName1 = "follower";
 952         writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
 953
 954         // Register local
 955
 956         shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
 957         kit.expectMsgClass(SuccessReply.class);
 958
 959         // Verify the local candidate becomes owner
 960
 961         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
 962         verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 963         verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
 964     }
 965
 966     private static class MockLeader extends UntypedActor {
 967         volatile CountDownLatch modificationsReceived = new CountDownLatch(1);
 968         List<Modification> receivedModifications = new ArrayList<>();
 969         volatile boolean sendReply = true;
 970         volatile long delay;
 971
 972         @Override
 973         public void onReceive(Object message) {
 974             if(message instanceof BatchedModifications) {
 975                 if(delay > 0) {
 976                     Uninterruptibles.sleepUninterruptibly(delay, TimeUnit.MILLISECONDS);
 977                 }
 978
 979                 if(sendReply) {
 980                     BatchedModifications mods = (BatchedModifications) message;
 981                     synchronized (receivedModifications) {
 982                         for(int i = 0; i < mods.getModifications().size(); i++) {
 983                             receivedModifications.add(mods.getModifications().get(i));
 984                             modificationsReceived.countDown();
 985                         }
 986                     }
 987
 988                     getSender().tell(CommitTransactionReply.instance(DataStoreVersions.CURRENT_VERSION).
 989                             toSerializable(), getSelf());
 990                 } else {
 991                     sendReply = true;
 992                 }
 993             }
 994         }
 995
 996         List<Modification> getAndClearReceivedModifications() {
 997             synchronized (receivedModifications) {
 998                 List<Modification> ret = new ArrayList<>(receivedModifications);
 999                 receivedModifications.clear();
1000                 return ret;
1001             }
1002         }
1003     }
1004 }