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