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