2 * Copyright (c) 2015 Brocade Communications Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
8 package org.opendaylight.controller.cluster.datastore.entityownership;
10 import static org.hamcrest.CoreMatchers.either;
11 import static org.hamcrest.CoreMatchers.equalTo;
12 import static org.junit.Assert.assertEquals;
13 import static org.mockito.Matchers.any;
14 import static org.mockito.Matchers.anyBoolean;
15 import static org.mockito.Matchers.argThat;
16 import static org.mockito.Matchers.eq;
17 import static org.mockito.Mockito.mock;
18 import static org.mockito.Mockito.never;
19 import static org.mockito.Mockito.reset;
20 import static org.mockito.Mockito.timeout;
21 import static org.mockito.Mockito.verify;
22 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.ENTITY_OWNERS_PATH;
23 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.candidatePath;
24 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.entityOwnersWithCandidate;
25 import akka.actor.ActorRef;
26 import akka.actor.PoisonPill;
27 import akka.actor.Props;
28 import akka.actor.Terminated;
29 import akka.actor.UntypedActor;
30 import akka.dispatch.Dispatchers;
31 import akka.testkit.JavaTestKit;
32 import akka.testkit.TestActorRef;
33 import com.google.common.base.Function;
34 import com.google.common.collect.ImmutableMap;
35 import com.google.common.util.concurrent.Uninterruptibles;
36 import java.util.ArrayList;
37 import java.util.Collections;
38 import java.util.List;
40 import java.util.concurrent.CountDownLatch;
41 import java.util.concurrent.TimeUnit;
42 import java.util.concurrent.atomic.AtomicInteger;
43 import org.hamcrest.Matcher;
44 import org.junit.After;
45 import org.junit.Test;
46 import org.opendaylight.controller.cluster.datastore.AbstractShardTest;
47 import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
48 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
49 import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
50 import org.opendaylight.controller.cluster.datastore.ShardDataTree;
51 import org.opendaylight.controller.cluster.datastore.ShardTestKit;
52 import org.opendaylight.controller.cluster.datastore.entityownership.messages.RegisterCandidateLocal;
53 import org.opendaylight.controller.cluster.datastore.entityownership.messages.RegisterListenerLocal;
54 import org.opendaylight.controller.cluster.datastore.entityownership.messages.UnregisterCandidateLocal;
55 import org.opendaylight.controller.cluster.datastore.entityownership.messages.UnregisterListenerLocal;
56 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
57 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
58 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
59 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
60 import org.opendaylight.controller.cluster.datastore.messages.PeerDown;
61 import org.opendaylight.controller.cluster.datastore.messages.PeerUp;
62 import org.opendaylight.controller.cluster.datastore.messages.SuccessReply;
63 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
64 import org.opendaylight.controller.cluster.datastore.modification.Modification;
65 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
66 import org.opendaylight.controller.cluster.raft.TestActorFactory;
67 import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
68 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
69 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
70 import org.opendaylight.controller.cluster.raft.messages.RequestVote;
71 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
72 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
73 import org.opendaylight.controller.md.sal.common.api.clustering.Entity;
74 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipCandidate;
75 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListener;
76 import org.opendaylight.yangtools.yang.common.QName;
77 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
78 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
79 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
82 * Unit tests for EntityOwnershipShard.
84 * @author Thomas Pantelis
86 public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest {
87 private static final String ENTITY_TYPE = "test type";
88 private static final YangInstanceIdentifier ENTITY_ID1 =
89 YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity1"));
90 private static final YangInstanceIdentifier ENTITY_ID2 =
91 YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity2"));
92 private static final YangInstanceIdentifier ENTITY_ID3 =
93 YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity3"));
94 private static final YangInstanceIdentifier ENTITY_ID4 =
95 YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity4"));
96 private static final SchemaContext SCHEMA_CONTEXT = SchemaContextHelper.entityOwners();
97 private static final AtomicInteger NEXT_SHARD_NUM = new AtomicInteger();
98 private static final String LOCAL_MEMBER_NAME = "member-1";
100 private final Builder dataStoreContextBuilder = DatastoreContext.newBuilder();
101 private final TestActorFactory actorFactory = new TestActorFactory(getSystem());
104 public void tearDown() {
105 actorFactory.close();
109 public void testOnRegisterCandidateLocal() throws Exception {
110 ShardTestKit kit = new ShardTestKit(getSystem());
112 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
114 kit.waitUntilLeader(shard);
116 YangInstanceIdentifier entityId = ENTITY_ID1;
117 Entity entity = new Entity(ENTITY_TYPE, entityId);
118 EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class);
120 shard.tell(new RegisterCandidateLocal(candidate, entity), kit.getRef());
121 kit.expectMsgClass(SuccessReply.class);
123 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
124 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
125 verify(candidate, timeout(5000)).ownershipChanged(entity, false, true);
129 public void testOnRegisterCandidateLocalWithNoInitialLeader() throws Exception {
130 ShardTestKit kit = new ShardTestKit(getSystem());
132 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2);
134 String peerId = newShardId("follower").toString();
135 TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId, false).
136 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
138 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
139 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
140 withDispatcher(Dispatchers.DefaultDispatcherId()));
142 YangInstanceIdentifier entityId = ENTITY_ID1;
143 Entity entity = new Entity(ENTITY_TYPE, entityId);
144 EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class);
146 shard.tell(new RegisterCandidateLocal(candidate, entity), kit.getRef());
147 kit.expectMsgClass(SuccessReply.class);
149 // Now grant the vote so the shard becomes the leader. This should retry the commit.
150 peer.underlyingActor().grantVote = true;
152 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
154 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
156 verify(candidate, timeout(5000)).ownershipChanged(entity, false, true);
160 public void testOnRegisterCandidateLocalWithNoInitialConsensus() throws Exception {
161 ShardTestKit kit = new ShardTestKit(getSystem());
163 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
164 shardTransactionCommitTimeoutInSeconds(1);
166 String peerId = newShardId("follower").toString();
167 TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId).
168 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
170 MockFollower follower = peer.underlyingActor();
172 // Drop AppendEntries so consensus isn't reached.
173 follower.dropAppendEntries = true;
175 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
176 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
177 withDispatcher(Dispatchers.DefaultDispatcherId()));
179 kit.waitUntilLeader(shard);
181 YangInstanceIdentifier entityId = ENTITY_ID1;
182 Entity entity = new Entity(ENTITY_TYPE, entityId);
183 EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class);
185 shard.tell(new RegisterCandidateLocal(candidate, entity), kit.getRef());
186 kit.expectMsgClass(SuccessReply.class);
188 // Wait enough time for the commit to timeout.
189 Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
191 // Resume AppendEntries - the follower should ack the commit which should then result in the candidate
192 // write being applied to the state.
193 follower.dropAppendEntries = false;
195 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
197 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
199 verify(candidate, timeout(5000)).ownershipChanged(entity, false, true);
203 public void testOnRegisterCandidateLocalWithIsolatedLeader() throws Exception {
204 ShardTestKit kit = new ShardTestKit(getSystem());
206 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
207 shardIsolatedLeaderCheckIntervalInMillis(50);
209 String peerId = newShardId("follower").toString();
210 TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId).
211 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
213 MockFollower follower = peer.underlyingActor();
215 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
216 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
217 withDispatcher(Dispatchers.DefaultDispatcherId()));
219 kit.waitUntilLeader(shard);
221 // Drop AppendEntries and wait enough time for the shard to switch to IsolatedLeader.
222 follower.dropAppendEntries = true;
223 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
225 YangInstanceIdentifier entityId = ENTITY_ID1;
226 Entity entity = new Entity(ENTITY_TYPE, entityId);
227 EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class);
229 shard.tell(new RegisterCandidateLocal(candidate, entity), kit.getRef());
230 kit.expectMsgClass(SuccessReply.class);
232 // Resume AppendEntries - the candidate write should now be committed.
233 follower.dropAppendEntries = false;
234 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
236 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
238 verify(candidate, timeout(5000)).ownershipChanged(entity, false, true);
242 public void testOnRegisterCandidateLocalWithRemoteLeader() throws Exception {
243 ShardTestKit kit = new ShardTestKit(getSystem());
245 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
246 shardBatchedModificationCount(5);
248 String peerId = newShardId("leader").toString();
249 TestActorRef<MockLeader> peer = actorFactory.createTestActor(Props.create(MockLeader.class).
250 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
252 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(Props.create(
253 TestEntityOwnershipShard.class, newShardId(LOCAL_MEMBER_NAME),
254 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build(),
255 dataStoreContextBuilder.build()).withDispatcher(Dispatchers.DefaultDispatcherId()));
257 shard.tell(new AppendEntries(1L, peerId, -1L, -1L, Collections.<ReplicatedLogEntry>emptyList(), -1L, -1L,
258 DataStoreVersions.CURRENT_VERSION), peer);
260 EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class);
262 shard.tell(new RegisterCandidateLocal(candidate, new Entity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef());
263 kit.expectMsgClass(SuccessReply.class);
265 MockLeader leader = peer.underlyingActor();
266 assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
267 leader.modificationsReceived, 5, TimeUnit.SECONDS));
268 verifyBatchedEntityCandidate(leader.getAndClearReceivedModifications(), ENTITY_TYPE, ENTITY_ID1,
271 // Test with initial commit timeout and subsequent retry.
273 leader.modificationsReceived = new CountDownLatch(1);
274 leader.sendReply = false;
276 shard.tell(dataStoreContextBuilder.shardTransactionCommitTimeoutInSeconds(1).build(), ActorRef.noSender());
278 shard.tell(new RegisterCandidateLocal(candidate, new Entity(ENTITY_TYPE, ENTITY_ID2)), kit.getRef());
279 kit.expectMsgClass(SuccessReply.class);
281 assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
282 leader.modificationsReceived, 5, TimeUnit.SECONDS));
283 verifyBatchedEntityCandidate(leader.getAndClearReceivedModifications(), ENTITY_TYPE, ENTITY_ID2,
286 // Send a bunch of registration messages quickly and verify.
290 leader.modificationsReceived = new CountDownLatch(max);
291 List<YangInstanceIdentifier> entityIds = new ArrayList<>();
292 for(int i = 1; i <= max; i++) {
293 YangInstanceIdentifier id = YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "test" + i));
295 shard.tell(new RegisterCandidateLocal(candidate, new Entity(ENTITY_TYPE, id)), kit.getRef());
298 assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
299 leader.modificationsReceived, 10, TimeUnit.SECONDS));
301 // Sleep a little to ensure no additional BatchedModifications are received.
303 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
305 List<Modification> receivedMods = leader.getAndClearReceivedModifications();
306 for(int i = 0; i < max; i++) {
307 verifyBatchedEntityCandidate(receivedMods.get(i), ENTITY_TYPE, entityIds.get(i), LOCAL_MEMBER_NAME);
310 assertEquals("# modifications received", max, receivedMods.size());
314 public void testOnUnregisterCandidateLocal() throws Exception {
315 ShardTestKit kit = new ShardTestKit(getSystem());
316 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
317 kit.waitUntilLeader(shard);
319 Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
320 EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class);
324 shard.tell(new RegisterCandidateLocal(candidate, entity), kit.getRef());
325 kit.expectMsgClass(SuccessReply.class);
327 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
328 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
329 verify(candidate, timeout(5000)).ownershipChanged(entity, false, true);
335 shard.tell(new UnregisterCandidateLocal(candidate, entity), kit.getRef());
336 kit.expectMsgClass(SuccessReply.class);
338 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, "");
339 verify(candidate, never()).ownershipChanged(any(Entity.class), anyBoolean(), anyBoolean());
343 shard.tell(new RegisterCandidateLocal(candidate, entity), kit.getRef());
344 kit.expectMsgClass(SuccessReply.class);
346 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
347 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
348 verify(candidate, timeout(5000)).ownershipChanged(entity, false, true);
352 public void testOwnershipChanges() throws Exception {
353 ShardTestKit kit = new ShardTestKit(getSystem());
354 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
355 kit.waitUntilLeader(shard);
357 Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
358 EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class);
359 ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
361 // Add a remote candidate
363 String remoteMemberName1 = "remoteMember1";
364 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
368 shard.tell(new RegisterCandidateLocal(candidate, entity), kit.getRef());
369 kit.expectMsgClass(SuccessReply.class);
371 // Verify the remote candidate becomes owner
373 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
374 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
375 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
376 verify(candidate, never()).ownershipChanged(any(Entity.class), anyBoolean(), anyBoolean());
378 // Add another remote candidate and verify ownership doesn't change
381 String remoteMemberName2 = "remoteMember2";
382 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
384 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
385 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
386 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
387 verify(candidate, never()).ownershipChanged(any(Entity.class), anyBoolean(), anyBoolean());
389 // Remove the second remote candidate and verify ownership doesn't change
392 deleteNode(candidatePath(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
394 verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
395 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
396 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
397 verify(candidate, never()).ownershipChanged(any(Entity.class), anyBoolean(), anyBoolean());
399 // Remove the first remote candidate and verify the local candidate becomes owner
402 deleteNode(candidatePath(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
404 verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
405 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
406 verify(candidate, timeout(5000)).ownershipChanged(entity, false, true);
408 // Add the second remote candidate back and verify ownership doesn't change
411 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
413 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
414 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
415 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
416 verify(candidate, never()).ownershipChanged(any(Entity.class), anyBoolean(), anyBoolean());
418 // Unregister the local candidate and verify the second remote candidate becomes owner
420 shard.tell(new UnregisterCandidateLocal(candidate, entity), kit.getRef());
421 kit.expectMsgClass(SuccessReply.class);
423 verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
424 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
428 public void testOwnerChangesOnPeerAvailabilityChanges() throws Exception {
429 ShardTestKit kit = new ShardTestKit(getSystem());
431 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(500).shardElectionTimeoutFactor(10000);
433 String peerMemberName1 = "peerMember1";
434 String peerMemberName2 = "peerMember2";
436 ShardIdentifier leaderId = newShardId(LOCAL_MEMBER_NAME);
437 ShardIdentifier peerId1 = newShardId(peerMemberName1);
438 ShardIdentifier peerId2 = newShardId(peerMemberName2);
440 TestActorRef<EntityOwnershipShard> peer1 = actorFactory.createTestActor(newShardProps(peerId1,
441 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId2.toString(), "").build(),
442 peerMemberName1).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
444 TestActorRef<EntityOwnershipShard> peer2 = actorFactory.createTestActor(newShardProps(peerId2,
445 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId1.toString(), "").build(),
446 peerMemberName2). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString());
448 TestActorRef<EntityOwnershipShard> leader = actorFactory.createTestActor(newShardProps(leaderId,
449 ImmutableMap.<String, String>builder().put(peerId1.toString(), peer1.path().toString()).
450 put(peerId2.toString(), peer2.path().toString()).build(), LOCAL_MEMBER_NAME).
451 withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString());
452 leader.tell(new ElectionTimeout(), leader);
454 kit.waitUntilLeader(leader);
456 EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class);
458 // Send PeerDown and PeerUp with no entities
460 leader.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender());
461 leader.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender());
463 // Add candidates for entity1 with the local leader as the owner
465 leader.tell(new RegisterCandidateLocal(candidate, new Entity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef());
466 kit.expectMsgClass(SuccessReply.class);
467 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
469 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, peerMemberName2), kit);
470 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
472 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, peerMemberName1), kit);
473 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName1);
474 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
476 // Add candidates for entity2 with peerMember2 as the owner
478 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName2), kit);
479 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
481 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName1), kit);
482 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
484 leader.tell(new RegisterCandidateLocal(candidate, new Entity(ENTITY_TYPE, ENTITY_ID2)), kit.getRef());
485 kit.expectMsgClass(SuccessReply.class);
486 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, LOCAL_MEMBER_NAME);
487 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
489 // Add candidates for entity3 with peerMember2 as the owner.
491 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName2), kit);
492 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
494 leader.tell(new RegisterCandidateLocal(candidate, new Entity(ENTITY_TYPE, ENTITY_ID3)), kit.getRef());
495 kit.expectMsgClass(SuccessReply.class);
496 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
498 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName1), kit);
499 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName1);
500 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
502 // Add only candidate peerMember2 for entity4.
504 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID4, peerMemberName2), kit);
505 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
506 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
508 // Kill peerMember2 and send PeerDown - the entities (2, 3, 4) owned by peerMember2 should get a new
512 peer2.tell(PoisonPill.getInstance(), ActorRef.noSender());
513 kit.expectMsgClass(JavaTestKit.duration("5 seconds"), Terminated.class);
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());
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);
526 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
527 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
528 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
529 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
531 // Reinstate peerMember2 - should become owner again for entity 4
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());
541 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
542 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
543 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
544 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
546 // Kill peerMember1 and send PeerDown - entity 2 should get a new owner selected
548 peer1.tell(PoisonPill.getInstance(), ActorRef.noSender());
549 leader.tell(new PeerDown(peerMemberName1, peerId1.toString()), ActorRef.noSender());
551 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
553 // Verify the reinstated peerMember2 is fully synced.
555 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
556 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
557 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
558 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
560 // Reinstate peerMember1 and verify no owner changes
562 peer1 = actorFactory.createTestActor(newShardProps(peerId1,
563 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId2.toString(), "").build(),
564 peerMemberName1).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
565 leader.tell(new PeerUp(peerMemberName1, peerId1.toString()), ActorRef.noSender());
567 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
568 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
569 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
570 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
572 // Verify the reinstated peerMember1 is fully synced.
574 verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
575 verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
576 verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
577 verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
579 // Kill the local leader and elect peer2 the leader. This should cause a new owner to be selected for
580 // the entities (1 and 3) previously owned by the local leader member.
582 peer2.tell(new PeerAddressResolved(peerId1.toString(), peer1.path().toString()), ActorRef.noSender());
583 peer2.tell(new PeerUp(LOCAL_MEMBER_NAME, leaderId.toString()), ActorRef.noSender());
584 peer2.tell(new PeerUp(peerMemberName1, peerId1.toString()), ActorRef.noSender());
586 leader.tell(PoisonPill.getInstance(), ActorRef.noSender());
587 peer2.tell(new PeerDown(LOCAL_MEMBER_NAME, leaderId.toString()), ActorRef.noSender());
588 peer2.tell(new ElectionTimeout(), peer2);
590 kit.waitUntilLeader(peer2);
592 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
593 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
594 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
595 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
599 public void testListenerRegistration() throws Exception {
600 ShardTestKit kit = new ShardTestKit(getSystem());
601 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
602 kit.waitUntilLeader(shard);
603 ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
605 String otherEntityType = "otherEntityType";
606 Entity entity1 = new Entity(ENTITY_TYPE, ENTITY_ID1);
607 Entity entity2 = new Entity(ENTITY_TYPE, ENTITY_ID2);
608 Entity entity3 = new Entity(ENTITY_TYPE, ENTITY_ID3);
609 Entity entity4 = new Entity(otherEntityType, ENTITY_ID3);
610 EntityOwnershipListener listener = mock(EntityOwnershipListener.class);
611 EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class);
615 shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
616 kit.expectMsgClass(SuccessReply.class);
618 // Register a couple candidates for the desired entity type and verify listener is notified.
620 shard.tell(new RegisterCandidateLocal(candidate, entity1), kit.getRef());
621 kit.expectMsgClass(SuccessReply.class);
623 verify(listener, timeout(5000)).ownershipChanged(entity1, false, true);
625 shard.tell(new RegisterCandidateLocal(candidate, entity2), kit.getRef());
626 kit.expectMsgClass(SuccessReply.class);
628 verify(listener, timeout(5000)).ownershipChanged(entity2, false, true);
631 // Register another candidate for another entity type and verify listener is not notified.
633 shard.tell(new RegisterCandidateLocal(candidate, entity4), kit.getRef());
634 kit.expectMsgClass(SuccessReply.class);
636 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
637 verify(listener, never()).ownershipChanged(eq(entity4), anyBoolean(), anyBoolean());
639 // Register remote candidate for entity1
641 String remoteMemberName = "remoteMember";
642 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, entity1.getId(), remoteMemberName),
644 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entity1.getId(), remoteMemberName);
646 // Unregister the local candidate for entity1 and verify listener is notified
648 shard.tell(new UnregisterCandidateLocal(candidate, entity1), kit.getRef());
649 kit.expectMsgClass(SuccessReply.class);
651 verify(listener, timeout(5000)).ownershipChanged(entity1, true, false);
654 // Unregister the listener, add a candidate for entity3 and verify listener isn't notified
656 shard.tell(new UnregisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
657 kit.expectMsgClass(SuccessReply.class);
659 shard.tell(new RegisterCandidateLocal(candidate, entity3), kit.getRef());
660 kit.expectMsgClass(SuccessReply.class);
662 verifyOwner(shard, ENTITY_TYPE, entity3.getId(), LOCAL_MEMBER_NAME);
663 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
664 verify(listener, never()).ownershipChanged(any(Entity.class), anyBoolean(), anyBoolean());
666 // Re-register the listener and verify it gets notified of current locally owned entities
668 reset(listener, candidate);
670 shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
671 kit.expectMsgClass(SuccessReply.class);
673 Matcher<Entity> entityMatcher = either(equalTo(entity2)).or(equalTo(entity3));
674 verify(listener, timeout(5000).times(2)).ownershipChanged(argThat(entityMatcher), eq(false), eq(true));
675 Uninterruptibles.sleepUninterruptibly(300, TimeUnit.MILLISECONDS);
676 verify(listener, never()).ownershipChanged(eq(entity4), anyBoolean(), anyBoolean());
677 verify(listener, never()).ownershipChanged(eq(entity1), anyBoolean(), anyBoolean());
678 verify(candidate, never()).ownershipChanged(eq(entity2), anyBoolean(), anyBoolean());
679 verify(candidate, never()).ownershipChanged(eq(entity3), anyBoolean(), anyBoolean());
682 private void commitModification(TestActorRef<EntityOwnershipShard> shard, NormalizedNode<?, ?> node,
683 JavaTestKit sender) {
684 BatchedModifications modifications = new BatchedModifications("tnx", DataStoreVersions.CURRENT_VERSION, "");
685 modifications.setDoCommitOnReady(true);
686 modifications.setReady(true);
687 modifications.setTotalMessagesSent(1);
688 modifications.addModification(new MergeModification(ENTITY_OWNERS_PATH, node));
690 shard.tell(modifications, sender.getRef());
691 sender.expectMsgClass(CommitTransactionReply.SERIALIZABLE_CLASS);
694 private void verifyEntityCandidateRemoved(final TestActorRef<EntityOwnershipShard> shard, String entityType,
695 YangInstanceIdentifier entityId, String candidateName) {
696 verifyNodeRemoved(candidatePath(entityType, entityId, candidateName),
697 new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
699 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
701 return AbstractShardTest.readStore(shard, path);
702 } catch(Exception e) {
703 throw new AssertionError("Failed to read " + path, e);
709 private void verifyCommittedEntityCandidate(final TestActorRef<EntityOwnershipShard> shard, String entityType,
710 YangInstanceIdentifier entityId, String candidateName) {
711 verifyEntityCandidate(entityType, entityId, candidateName, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
713 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
715 return AbstractShardTest.readStore(shard, path);
716 } catch(Exception e) {
717 throw new AssertionError("Failed to read " + path, e);
723 private void verifyBatchedEntityCandidate(List<Modification> mods, String entityType,
724 YangInstanceIdentifier entityId, String candidateName) throws Exception {
725 assertEquals("BatchedModifications size", 1, mods.size());
726 verifyBatchedEntityCandidate(mods.get(0), entityType, entityId, candidateName);
729 private void verifyBatchedEntityCandidate(Modification mod, String entityType,
730 YangInstanceIdentifier entityId, String candidateName) throws Exception {
731 assertEquals("Modification type", MergeModification.class, mod.getClass());
732 verifyEntityCandidate(((MergeModification)mod).getData(), entityType,
733 entityId, candidateName);
736 private void verifyOwner(final TestActorRef<EntityOwnershipShard> shard, String entityType, YangInstanceIdentifier entityId,
737 String localMemberName) {
738 verifyOwner(localMemberName, entityType, entityId, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
740 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
742 return AbstractShardTest.readStore(shard, path);
743 } catch(Exception e) {
750 private Props newShardProps() {
751 return newShardProps(Collections.<String,String>emptyMap());
754 private Props newShardProps(Map<String,String> peers) {
755 return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME);
758 private Props newShardProps(ShardIdentifier shardId, Map<String,String> peers, String memberName) {
759 return EntityOwnershipShard.props(shardId, peers, dataStoreContextBuilder.build(),
760 SCHEMA_CONTEXT, memberName);
763 private ShardIdentifier newShardId(String memberName) {
764 return ShardIdentifier.builder().memberName(memberName).shardName("entity-ownership").
765 type("operational" + NEXT_SHARD_NUM.getAndIncrement()).build();
768 public static class TestEntityOwnershipShard extends EntityOwnershipShard {
770 TestEntityOwnershipShard(ShardIdentifier name, Map<String, String> peerAddresses,
771 DatastoreContext datastoreContext) {
772 super(name, peerAddresses, datastoreContext, SCHEMA_CONTEXT, LOCAL_MEMBER_NAME);
776 public void onReceiveCommand(Object message) throws Exception {
777 if(!(message instanceof ElectionTimeout)) {
778 super.onReceiveCommand(message);
785 public static class MockFollower extends UntypedActor {
786 volatile boolean grantVote;
787 volatile boolean dropAppendEntries;
788 private final String myId;
790 public MockFollower(String myId) {
794 public MockFollower(String myId, boolean grantVote) {
796 this.grantVote = grantVote;
800 public void onReceive(Object message) {
801 if(message instanceof RequestVote) {
803 getSender().tell(new RequestVoteReply(((RequestVote)message).getTerm(), true), getSelf());
805 } else if(message instanceof AppendEntries) {
806 if(!dropAppendEntries) {
807 AppendEntries req = (AppendEntries) message;
808 long lastIndex = req.getLeaderCommit();
809 if (req.getEntries().size() > 0) {
810 for(ReplicatedLogEntry entry : req.getEntries()) {
811 lastIndex = entry.getIndex();
815 getSender().tell(new AppendEntriesReply(myId, req.getTerm(), true, lastIndex, req.getTerm(),
816 DataStoreVersions.CURRENT_VERSION), getSelf());
822 public static class MockLeader extends UntypedActor {
823 volatile CountDownLatch modificationsReceived = new CountDownLatch(1);
824 List<Modification> receivedModifications = new ArrayList<>();
825 volatile boolean sendReply = true;
829 public void onReceive(Object message) {
830 if(message instanceof BatchedModifications) {
832 Uninterruptibles.sleepUninterruptibly(delay, TimeUnit.MILLISECONDS);
836 BatchedModifications mods = (BatchedModifications) message;
837 synchronized (receivedModifications) {
838 for(int i = 0; i < mods.getModifications().size(); i++) {
839 receivedModifications.add(mods.getModifications().get(i));
840 modificationsReceived.countDown();
844 getSender().tell(CommitTransactionReply.INSTANCE.toSerializable(), getSelf());
851 List<Modification> getAndClearReceivedModifications() {
852 synchronized (receivedModifications) {
853 List<Modification> ret = new ArrayList<>(receivedModifications);
854 receivedModifications.clear();