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