opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShardTest.java

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