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

©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.