Bug 4105: Remove EntityOwnershipCandidate
[controller.git] / 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 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.