2 * Copyright (c) 2015 Brocade Communications Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.controller.cluster.datastore.entityownership;
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;
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.entityownership.selectionstrategy.EntityOwnerSelectionStrategyConfig;
52 import org.opendaylight.controller.cluster.datastore.entityownership.selectionstrategy.LastCandidateSelectionStrategy;
53 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
54 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
55 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
56 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
57 import org.opendaylight.controller.cluster.datastore.messages.PeerDown;
58 import org.opendaylight.controller.cluster.datastore.messages.PeerUp;
59 import org.opendaylight.controller.cluster.datastore.messages.SuccessReply;
60 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
61 import org.opendaylight.controller.cluster.datastore.modification.Modification;
62 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
63 import org.opendaylight.controller.cluster.raft.TestActorFactory;
64 import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
65 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
66 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
67 import org.opendaylight.controller.cluster.raft.messages.RequestVote;
68 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
69 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
70 import org.opendaylight.controller.md.sal.common.api.clustering.Entity;
71 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipChange;
72 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListener;
73 import org.opendaylight.yangtools.yang.common.QName;
74 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
75 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
76 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
79 * Unit tests for EntityOwnershipShard.
81 * @author Thomas Pantelis
83 public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest {
84 private static final String ENTITY_TYPE = "test type";
85 private static final YangInstanceIdentifier ENTITY_ID1 =
86 YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity1"));
87 private static final YangInstanceIdentifier ENTITY_ID2 =
88 YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity2"));
89 private static final YangInstanceIdentifier ENTITY_ID3 =
90 YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity3"));
91 private static final YangInstanceIdentifier ENTITY_ID4 =
92 YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity4"));
93 private static final YangInstanceIdentifier ENTITY_ID5 =
94 YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity5"));
95 private static final SchemaContext SCHEMA_CONTEXT = SchemaContextHelper.entityOwners();
96 private static final AtomicInteger NEXT_SHARD_NUM = new AtomicInteger();
97 private static final String LOCAL_MEMBER_NAME = "member-1";
99 private final Builder dataStoreContextBuilder = DatastoreContext.newBuilder();
100 private final TestActorFactory actorFactory = new TestActorFactory(getSystem());
103 public void tearDown() {
104 actorFactory.close();
108 public void testOnRegisterCandidateLocal() throws Exception {
109 ShardTestKit kit = new ShardTestKit(getSystem());
111 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
113 kit.waitUntilLeader(shard);
115 YangInstanceIdentifier entityId = ENTITY_ID1;
116 Entity entity = new Entity(ENTITY_TYPE, entityId);
118 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
119 kit.expectMsgClass(SuccessReply.class);
121 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
122 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
126 public void testOnRegisterCandidateLocalWithNoInitialLeader() throws Exception {
127 ShardTestKit kit = new ShardTestKit(getSystem());
129 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2);
131 String peerId = newShardId("follower").toString();
132 TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId, false).
133 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
135 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
136 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
137 withDispatcher(Dispatchers.DefaultDispatcherId()));
139 YangInstanceIdentifier entityId = ENTITY_ID1;
140 Entity entity = new Entity(ENTITY_TYPE, entityId);
142 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
143 kit.expectMsgClass(SuccessReply.class);
145 // Now grant the vote so the shard becomes the leader. This should retry the commit.
146 peer.underlyingActor().grantVote = true;
148 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
149 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
153 public void testOnRegisterCandidateLocalWithNoInitialConsensus() throws Exception {
154 ShardTestKit kit = new ShardTestKit(getSystem());
156 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
157 shardTransactionCommitTimeoutInSeconds(1);
159 String peerId = newShardId("follower").toString();
160 TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId).
161 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
163 MockFollower follower = peer.underlyingActor();
165 // Drop AppendEntries so consensus isn't reached.
166 follower.dropAppendEntries = true;
168 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
169 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
170 withDispatcher(Dispatchers.DefaultDispatcherId()));
172 kit.waitUntilLeader(shard);
174 YangInstanceIdentifier entityId = ENTITY_ID1;
175 Entity entity = new Entity(ENTITY_TYPE, entityId);
177 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
178 kit.expectMsgClass(SuccessReply.class);
180 // Wait enough time for the commit to timeout.
181 Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
183 // Resume AppendEntries - the follower should ack the commit which should then result in the candidate
184 // write being applied to the state.
185 follower.dropAppendEntries = false;
187 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
188 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
192 public void testOnRegisterCandidateLocalWithIsolatedLeader() throws Exception {
193 ShardTestKit kit = new ShardTestKit(getSystem());
195 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
196 shardIsolatedLeaderCheckIntervalInMillis(50);
198 String peerId = newShardId("follower").toString();
199 TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId).
200 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
202 MockFollower follower = peer.underlyingActor();
204 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
205 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
206 withDispatcher(Dispatchers.DefaultDispatcherId()));
208 kit.waitUntilLeader(shard);
210 // Drop AppendEntries and wait enough time for the shard to switch to IsolatedLeader.
211 follower.dropAppendEntries = true;
212 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
214 YangInstanceIdentifier entityId = ENTITY_ID1;
215 Entity entity = new Entity(ENTITY_TYPE, entityId);
217 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
218 kit.expectMsgClass(SuccessReply.class);
220 // Resume AppendEntries - the candidate write should now be committed.
221 follower.dropAppendEntries = false;
222 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
223 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
227 public void testOnRegisterCandidateLocalWithRemoteLeader() throws Exception {
228 ShardTestKit kit = new ShardTestKit(getSystem());
230 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
231 shardBatchedModificationCount(5);
233 String peerId = newShardId("leader").toString();
234 TestActorRef<MockLeader> peer = actorFactory.createTestActor(Props.create(MockLeader.class).
235 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
237 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(Props.create(
238 TestEntityOwnershipShard.class, newShardId(LOCAL_MEMBER_NAME),
239 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build(),
240 dataStoreContextBuilder.build()).withDispatcher(Dispatchers.DefaultDispatcherId()));
242 shard.tell(new AppendEntries(1L, peerId, -1L, -1L, Collections.<ReplicatedLogEntry>emptyList(), -1L, -1L,
243 DataStoreVersions.CURRENT_VERSION), peer);
245 shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef());
246 kit.expectMsgClass(SuccessReply.class);
248 MockLeader leader = peer.underlyingActor();
249 assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
250 leader.modificationsReceived, 5, TimeUnit.SECONDS));
251 verifyBatchedEntityCandidate(leader.getAndClearReceivedModifications(), ENTITY_TYPE, ENTITY_ID1,
254 // Test with initial commit timeout and subsequent retry.
256 leader.modificationsReceived = new CountDownLatch(1);
257 leader.sendReply = false;
259 shard.tell(dataStoreContextBuilder.shardTransactionCommitTimeoutInSeconds(1).build(), ActorRef.noSender());
261 shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID2)), kit.getRef());
262 kit.expectMsgClass(SuccessReply.class);
264 assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
265 leader.modificationsReceived, 5, TimeUnit.SECONDS));
266 verifyBatchedEntityCandidate(leader.getAndClearReceivedModifications(), ENTITY_TYPE, ENTITY_ID2,
269 // Send a bunch of registration messages quickly and verify.
273 leader.modificationsReceived = new CountDownLatch(max);
274 List<YangInstanceIdentifier> entityIds = new ArrayList<>();
275 for(int i = 1; i <= max; i++) {
276 YangInstanceIdentifier id = YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "test" + i));
278 shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, id)), kit.getRef());
281 assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
282 leader.modificationsReceived, 10, TimeUnit.SECONDS));
284 // Sleep a little to ensure no additional BatchedModifications are received.
286 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
288 List<Modification> receivedMods = leader.getAndClearReceivedModifications();
289 for(int i = 0; i < max; i++) {
290 verifyBatchedEntityCandidate(receivedMods.get(i), ENTITY_TYPE, entityIds.get(i), LOCAL_MEMBER_NAME);
293 assertEquals("# modifications received", max, receivedMods.size());
297 public void testOnUnregisterCandidateLocal() throws Exception {
298 ShardTestKit kit = new ShardTestKit(getSystem());
299 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
300 kit.waitUntilLeader(shard);
302 Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
306 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
307 kit.expectMsgClass(SuccessReply.class);
309 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
310 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
314 shard.tell(new UnregisterCandidateLocal(entity), kit.getRef());
315 kit.expectMsgClass(SuccessReply.class);
317 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, "");
321 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
322 kit.expectMsgClass(SuccessReply.class);
324 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
325 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
329 public void testOwnershipChanges() throws Exception {
330 ShardTestKit kit = new ShardTestKit(getSystem());
331 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
332 kit.waitUntilLeader(shard);
334 Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
335 ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
337 // Add a remote candidate
339 String remoteMemberName1 = "remoteMember1";
340 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
344 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
345 kit.expectMsgClass(SuccessReply.class);
347 // Verify the remote candidate becomes owner
349 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
350 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
351 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
353 // Add another remote candidate and verify ownership doesn't change
355 String remoteMemberName2 = "remoteMember2";
356 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
358 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
359 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
360 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
362 // Remove the second remote candidate and verify ownership doesn't change
364 deleteNode(candidatePath(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
366 verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
367 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
368 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
370 // Remove the first remote candidate and verify the local candidate becomes owner
372 deleteNode(candidatePath(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
374 verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
375 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
377 // Add the second remote candidate back and verify ownership doesn't change
379 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
381 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
382 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
383 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
385 // Unregister the local candidate and verify the second remote candidate becomes owner
387 shard.tell(new UnregisterCandidateLocal(entity), kit.getRef());
388 kit.expectMsgClass(SuccessReply.class);
390 verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
391 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
395 public void testOwnerChangesOnPeerAvailabilityChanges() throws Exception {
396 ShardTestKit kit = new ShardTestKit(getSystem());
398 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(500).shardElectionTimeoutFactor(10000);
400 String peerMemberName1 = "peerMember1";
401 String peerMemberName2 = "peerMember2";
403 ShardIdentifier leaderId = newShardId(LOCAL_MEMBER_NAME);
404 ShardIdentifier peerId1 = newShardId(peerMemberName1);
405 ShardIdentifier peerId2 = newShardId(peerMemberName2);
407 TestActorRef<EntityOwnershipShard> peer1 = actorFactory.createTestActor(newShardProps(peerId1,
408 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId2.toString(), "").build(),
409 peerMemberName1, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
411 TestActorRef<EntityOwnershipShard> peer2 = actorFactory.createTestActor(newShardProps(peerId2,
412 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId1.toString(), "").build(),
413 peerMemberName2, EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString());
415 TestActorRef<EntityOwnershipShard> leader = actorFactory.createTestActor(newShardProps(leaderId,
416 ImmutableMap.<String, String>builder().put(peerId1.toString(), peer1.path().toString()).
417 put(peerId2.toString(), peer2.path().toString()).build(), LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build()).
418 withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString());
419 leader.tell(new ElectionTimeout(), leader);
421 kit.waitUntilLeader(leader);
423 // Send PeerDown and PeerUp with no entities
425 leader.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender());
426 leader.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender());
428 // Add candidates for entity1 with the local leader as the owner
430 leader.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef());
431 kit.expectMsgClass(SuccessReply.class);
432 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
434 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, peerMemberName2), kit);
435 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
437 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, peerMemberName1), kit);
438 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName1);
439 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
441 // Add candidates for entity2 with peerMember2 as the owner
443 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName2), kit);
444 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
446 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName1), kit);
447 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
448 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
450 // Add candidates for entity3 with peerMember2 as the owner.
452 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName2), kit);
453 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
455 leader.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID3)), kit.getRef());
456 kit.expectMsgClass(SuccessReply.class);
457 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
459 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName1), kit);
460 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName1);
461 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
463 // Add only candidate peerMember2 for entity4.
465 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID4, peerMemberName2), kit);
466 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
467 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
469 // Add only candidate peerMember1 for entity5.
471 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID5, peerMemberName1), kit);
472 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID5, peerMemberName1);
473 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID5, peerMemberName1);
475 // Kill peerMember2 and send PeerDown - the entities (2, 3, 4) owned by peerMember2 should get a new
479 peer2.tell(PoisonPill.getInstance(), ActorRef.noSender());
480 kit.expectMsgClass(JavaTestKit.duration("5 seconds"), Terminated.class);
483 leader.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender());
484 // Send PeerDown again - should be noop
485 leader.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender());
486 peer1.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender());
488 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, ""); // no other candidates so should clear
489 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
490 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
491 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
493 verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
494 verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
495 verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
496 verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
498 // Reinstate peerMember2 - no owners should change
500 peer2 = actorFactory.createTestActor(newShardProps(peerId2,
501 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId1.toString(), "").build(),
502 peerMemberName2, EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString());
503 leader.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender());
504 // Send PeerUp again - should be noop
505 leader.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender());
506 peer1.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender());
508 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
509 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
510 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
511 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, "");
513 // Add back candidate peerMember2 for entities 1, 2, & 3.
515 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, peerMemberName2), kit);
516 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName2), kit);
517 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName2), kit);
518 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
519 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
520 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
521 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
522 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
523 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
525 // Kill peerMember1 and send PeerDown - entity 2 should get a new owner selected
527 peer1.tell(PoisonPill.getInstance(), ActorRef.noSender());
528 leader.tell(new PeerDown(peerMemberName1, peerId1.toString()), ActorRef.noSender());
530 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
532 // Verify the reinstated peerMember2 is fully synced.
534 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, "");
535 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
536 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
537 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
539 // Reinstate peerMember1 and verify no owner changes
541 peer1 = actorFactory.createTestActor(newShardProps(peerId1,
542 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId2.toString(), "").build(),
543 peerMemberName1, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
544 leader.tell(new PeerUp(peerMemberName1, peerId1.toString()), ActorRef.noSender());
546 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, "");
547 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
548 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
549 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
551 // Verify the reinstated peerMember1 is fully synced.
553 verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID4, "");
554 verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
555 verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
556 verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
558 // Kill the local leader and elect peer2 the leader. This should cause a new owner to be selected for
559 // the entities (1 and 3) previously owned by the local leader member.
561 peer2.tell(new PeerAddressResolved(peerId1.toString(), peer1.path().toString()), ActorRef.noSender());
562 peer2.tell(new PeerUp(LOCAL_MEMBER_NAME, leaderId.toString()), ActorRef.noSender());
563 peer2.tell(new PeerUp(peerMemberName1, peerId1.toString()), ActorRef.noSender());
565 leader.tell(PoisonPill.getInstance(), ActorRef.noSender());
566 peer2.tell(new PeerDown(LOCAL_MEMBER_NAME, leaderId.toString()), ActorRef.noSender());
567 peer2.tell(new ElectionTimeout(), peer2);
569 kit.waitUntilLeader(peer2);
571 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, "");
572 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
573 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
574 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
578 public void testLocalCandidateRemovedWithCandidateRegistered() throws Exception {
579 ShardTestKit kit = new ShardTestKit(getSystem());
581 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(10000);
582 ShardIdentifier leaderId = newShardId("leader");
583 ShardIdentifier localId = newShardId(LOCAL_MEMBER_NAME);
585 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(Props.create(
586 TestEntityOwnershipShard.class, localId,
587 ImmutableMap.<String, String>builder().put(leaderId.toString(), "".toString()).build(),
588 dataStoreContextBuilder.build()).withDispatcher(Dispatchers.DefaultDispatcherId()));
590 TestActorRef<EntityOwnershipShard> leader = actorFactory.createTestActor(newShardProps(leaderId,
591 ImmutableMap.<String, String>builder().put(localId.toString(), shard.path().toString()).build(),
592 LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString());
593 leader.tell(new ElectionTimeout(), leader);
595 kit.waitUntilLeader(leader);
597 shard.tell(new PeerAddressResolved(leaderId.toString(), leader.path().toString()), ActorRef.noSender());
599 Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
600 EntityOwnershipListener listener = mock(EntityOwnershipListener.class);
602 shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
603 kit.expectMsgClass(SuccessReply.class);
605 // Register local candidate
607 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
608 kit.expectMsgClass(SuccessReply.class);
609 verifyCommittedEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME);
610 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true));
613 // Simulate a replicated commit from the leader to remove the local candidate that would occur after a
614 // network partition is healed.
616 leader.tell(new PeerDown(LOCAL_MEMBER_NAME, localId.toString()), ActorRef.noSender());
618 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, true, false, false));
620 // Since the the shard has a local candidate registered, it should re-add its candidate to the entity.
622 verifyCommittedEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME);
623 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true));
625 // Unregister the local candidate and verify it's removed and no re-added.
627 shard.tell(new UnregisterCandidateLocal(entity), kit.getRef());
628 kit.expectMsgClass(SuccessReply.class);
630 verifyNoEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME);
631 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
632 verifyNoEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME);
636 public void testListenerRegistration() throws Exception {
637 ShardTestKit kit = new ShardTestKit(getSystem());
638 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
639 kit.waitUntilLeader(shard);
640 ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
642 String otherEntityType = "otherEntityType";
643 Entity entity1 = new Entity(ENTITY_TYPE, ENTITY_ID1);
644 Entity entity2 = new Entity(ENTITY_TYPE, ENTITY_ID2);
645 Entity entity3 = new Entity(ENTITY_TYPE, ENTITY_ID3);
646 Entity entity4 = new Entity(otherEntityType, ENTITY_ID3);
647 EntityOwnershipListener listener = mock(EntityOwnershipListener.class);
651 shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
652 kit.expectMsgClass(SuccessReply.class);
654 // Register a couple candidates for the desired entity type and verify listener is notified.
656 shard.tell(new RegisterCandidateLocal(entity1), kit.getRef());
657 kit.expectMsgClass(SuccessReply.class);
659 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, false, true, true));
661 shard.tell(new RegisterCandidateLocal(entity2), kit.getRef());
662 kit.expectMsgClass(SuccessReply.class);
664 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity2, false, true, true));
667 // Register another candidate for another entity type and verify listener is not notified.
669 shard.tell(new RegisterCandidateLocal(entity4), kit.getRef());
670 kit.expectMsgClass(SuccessReply.class);
672 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
673 verify(listener, never()).ownershipChanged(ownershipChange(entity4));
675 // Register remote candidate for entity1
677 String remoteMemberName = "remoteMember";
678 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, entity1.getId(), remoteMemberName),
680 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entity1.getId(), remoteMemberName);
682 // Unregister the local candidate for entity1 and verify listener is notified
684 shard.tell(new UnregisterCandidateLocal(entity1), kit.getRef());
685 kit.expectMsgClass(SuccessReply.class);
687 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, true, false, true));
690 // Unregister the listener, add a candidate for entity3 and verify listener isn't notified
692 shard.tell(new UnregisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
693 kit.expectMsgClass(SuccessReply.class);
695 shard.tell(new RegisterCandidateLocal(entity3), kit.getRef());
696 kit.expectMsgClass(SuccessReply.class);
698 verifyOwner(shard, ENTITY_TYPE, entity3.getId(), LOCAL_MEMBER_NAME);
699 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
700 verify(listener, never()).ownershipChanged(any(EntityOwnershipChange.class));
702 // Re-register the listener and verify it gets notified of currently owned entities
706 shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
707 kit.expectMsgClass(SuccessReply.class);
709 verify(listener, timeout(5000).times(2)).ownershipChanged(or(ownershipChange(entity2, false, true, true),
710 ownershipChange(entity3, false, true, true)));
711 Uninterruptibles.sleepUninterruptibly(300, TimeUnit.MILLISECONDS);
712 verify(listener, never()).ownershipChanged(ownershipChange(entity4));
713 verify(listener, never()).ownershipChanged(ownershipChange(entity1));
716 private static void commitModification(TestActorRef<EntityOwnershipShard> shard, NormalizedNode<?, ?> node,
717 JavaTestKit sender) {
718 BatchedModifications modifications = newBatchedModifications();
719 modifications.addModification(new MergeModification(ENTITY_OWNERS_PATH, node));
721 shard.tell(modifications, sender.getRef());
722 sender.expectMsgClass(CommitTransactionReply.SERIALIZABLE_CLASS);
725 private static BatchedModifications newBatchedModifications() {
726 BatchedModifications modifications = new BatchedModifications("tnx", DataStoreVersions.CURRENT_VERSION, "");
727 modifications.setDoCommitOnReady(true);
728 modifications.setReady(true);
729 modifications.setTotalMessagesSent(1);
730 return modifications;
733 private void verifyEntityCandidateRemoved(final TestActorRef<EntityOwnershipShard> shard, String entityType,
734 YangInstanceIdentifier entityId, String candidateName) {
735 verifyNodeRemoved(candidatePath(entityType, entityId, candidateName),
736 new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
738 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
740 return AbstractShardTest.readStore(shard, path);
741 } catch(Exception e) {
742 throw new AssertionError("Failed to read " + path, e);
748 private void verifyCommittedEntityCandidate(final TestActorRef<EntityOwnershipShard> shard, String entityType,
749 YangInstanceIdentifier entityId, String candidateName) {
750 verifyEntityCandidate(entityType, entityId, candidateName, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
752 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
754 return AbstractShardTest.readStore(shard, path);
755 } catch(Exception e) {
756 throw new AssertionError("Failed to read " + path, e);
762 private void verifyNoEntityCandidate(final TestActorRef<EntityOwnershipShard> shard, String entityType,
763 YangInstanceIdentifier entityId, String candidateName) {
764 verifyEntityCandidate(entityType, entityId, candidateName, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
766 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
768 return AbstractShardTest.readStore(shard, path);
769 } catch(Exception e) {
770 throw new AssertionError("Failed to read " + path, e);
776 private void verifyBatchedEntityCandidate(List<Modification> mods, String entityType,
777 YangInstanceIdentifier entityId, String candidateName) throws Exception {
778 assertEquals("BatchedModifications size", 1, mods.size());
779 verifyBatchedEntityCandidate(mods.get(0), entityType, entityId, candidateName);
782 private void verifyBatchedEntityCandidate(Modification mod, String entityType,
783 YangInstanceIdentifier entityId, String candidateName) throws Exception {
784 assertEquals("Modification type", MergeModification.class, mod.getClass());
785 verifyEntityCandidate(((MergeModification)mod).getData(), entityType,
786 entityId, candidateName, true);
789 private static void verifyOwner(final TestActorRef<EntityOwnershipShard> shard, String entityType,
790 YangInstanceIdentifier entityId, String localMemberName) {
791 verifyOwner(localMemberName, entityType, entityId, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
793 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
795 return AbstractShardTest.readStore(shard, path);
796 } catch(Exception e) {
803 private Props newShardProps() {
804 return newShardProps(Collections.<String,String>emptyMap());
807 private Props newShardProps(EntityOwnerSelectionStrategyConfig strategyConfig) {
808 return newShardProps(newShardId(LOCAL_MEMBER_NAME), Collections.<String,String>emptyMap(),
809 LOCAL_MEMBER_NAME, strategyConfig);
812 private Props newShardProps(Map<String,String> peers) {
813 return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build());
816 private Props newShardProps(ShardIdentifier shardId, Map<String,String> peers, String memberName,
817 EntityOwnerSelectionStrategyConfig config) {
818 return EntityOwnershipShard.newBuilder().id(shardId).peerAddresses(peers).
819 datastoreContext(dataStoreContextBuilder.build()).schemaContext(SCHEMA_CONTEXT).
820 localMemberName(memberName).ownerSelectionStrategyConfig(config).props().withDispatcher(Dispatchers.DefaultDispatcherId());
823 private static ShardIdentifier newShardId(String memberName) {
824 return ShardIdentifier.builder().memberName(memberName).shardName("entity-ownership").
825 type("operational" + NEXT_SHARD_NUM.getAndIncrement()).build();
828 public static class TestEntityOwnershipShard extends EntityOwnershipShard {
830 TestEntityOwnershipShard(ShardIdentifier name, Map<String, String> peerAddresses,
831 DatastoreContext datastoreContext) {
832 super(newBuilder().id(name).peerAddresses(peerAddresses).datastoreContext(datastoreContext).
833 schemaContext(SCHEMA_CONTEXT).localMemberName(LOCAL_MEMBER_NAME));
837 public void onReceiveCommand(Object message) throws Exception {
838 if(!(message instanceof ElectionTimeout)) {
839 super.onReceiveCommand(message);
846 public static class MockFollower extends UntypedActor {
847 volatile boolean grantVote;
848 volatile boolean dropAppendEntries;
849 private final String myId;
851 public MockFollower(String myId) {
855 public MockFollower(String myId, boolean grantVote) {
857 this.grantVote = grantVote;
861 public void onReceive(Object message) {
862 if(message instanceof RequestVote) {
864 getSender().tell(new RequestVoteReply(((RequestVote)message).getTerm(), true), getSelf());
866 } else if(message instanceof AppendEntries) {
867 if(!dropAppendEntries) {
868 AppendEntries req = (AppendEntries) message;
869 long lastIndex = req.getLeaderCommit();
870 if (req.getEntries().size() > 0) {
871 for(ReplicatedLogEntry entry : req.getEntries()) {
872 lastIndex = entry.getIndex();
876 getSender().tell(new AppendEntriesReply(myId, req.getTerm(), true, lastIndex, req.getTerm(),
877 DataStoreVersions.CURRENT_VERSION), getSelf());
885 public void testDelayedEntityOwnerSelection() throws Exception {
886 ShardTestKit kit = new ShardTestKit(getSystem());
887 EntityOwnerSelectionStrategyConfig.Builder builder
888 = EntityOwnerSelectionStrategyConfig.newBuilder().addStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500);
889 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(builder.build()));
890 kit.waitUntilLeader(shard);
892 Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
893 ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
895 // Add a remote candidate
897 String remoteMemberName1 = "remoteMember1";
898 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
903 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
904 kit.expectMsgClass(SuccessReply.class);
906 // Verify the local candidate becomes owner
908 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
909 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
910 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
913 public static class MockLeader extends UntypedActor {
914 volatile CountDownLatch modificationsReceived = new CountDownLatch(1);
915 List<Modification> receivedModifications = new ArrayList<>();
916 volatile boolean sendReply = true;
920 public void onReceive(Object message) {
921 if(message instanceof BatchedModifications) {
923 Uninterruptibles.sleepUninterruptibly(delay, TimeUnit.MILLISECONDS);
927 BatchedModifications mods = (BatchedModifications) message;
928 synchronized (receivedModifications) {
929 for(int i = 0; i < mods.getModifications().size(); i++) {
930 receivedModifications.add(mods.getModifications().get(i));
931 modificationsReceived.countDown();
935 getSender().tell(CommitTransactionReply.INSTANCE.toSerializable(), getSelf());
942 List<Modification> getAndClearReceivedModifications() {
943 synchronized (receivedModifications) {
944 List<Modification> ret = new ArrayList<>(receivedModifications);
945 receivedModifications.clear();