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