2 * Copyright (c) 2015 Brocade Communications Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
8 package org.opendaylight.controller.cluster.datastore.entityownership;
10 import static org.junit.Assert.assertEquals;
11 import static org.mockito.AdditionalMatchers.or;
12 import static org.mockito.Matchers.any;
13 import static org.mockito.Mockito.mock;
14 import static org.mockito.Mockito.never;
15 import static org.mockito.Mockito.reset;
16 import static org.mockito.Mockito.timeout;
17 import static org.mockito.Mockito.verify;
18 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.ENTITY_OWNERS_PATH;
19 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.candidatePath;
20 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.entityOwnersWithCandidate;
21 import akka.actor.ActorRef;
22 import akka.actor.PoisonPill;
23 import akka.actor.Props;
24 import akka.actor.Terminated;
25 import akka.actor.UntypedActor;
26 import akka.dispatch.Dispatchers;
27 import akka.testkit.JavaTestKit;
28 import akka.testkit.TestActorRef;
29 import com.google.common.base.Function;
30 import com.google.common.collect.ImmutableMap;
31 import com.google.common.util.concurrent.Uninterruptibles;
32 import java.util.ArrayList;
33 import java.util.Collections;
34 import java.util.List;
36 import java.util.concurrent.CountDownLatch;
37 import java.util.concurrent.TimeUnit;
38 import java.util.concurrent.atomic.AtomicInteger;
39 import org.junit.After;
40 import org.junit.Test;
41 import org.opendaylight.controller.cluster.datastore.AbstractShardTest;
42 import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
43 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
44 import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
45 import org.opendaylight.controller.cluster.datastore.ShardDataTree;
46 import org.opendaylight.controller.cluster.datastore.ShardTestKit;
47 import org.opendaylight.controller.cluster.datastore.entityownership.messages.RegisterCandidateLocal;
48 import org.opendaylight.controller.cluster.datastore.entityownership.messages.RegisterListenerLocal;
49 import org.opendaylight.controller.cluster.datastore.entityownership.messages.UnregisterCandidateLocal;
50 import org.opendaylight.controller.cluster.datastore.entityownership.messages.UnregisterListenerLocal;
51 import org.opendaylight.controller.cluster.datastore.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;
77 * Unit tests for EntityOwnershipShard.
79 * @author Thomas Pantelis
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";
97 private final Builder dataStoreContextBuilder = DatastoreContext.newBuilder();
98 private final TestActorFactory actorFactory = new TestActorFactory(getSystem());
101 public void tearDown() {
102 actorFactory.close();
106 public void testOnRegisterCandidateLocal() throws Exception {
107 ShardTestKit kit = new ShardTestKit(getSystem());
109 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
111 kit.waitUntilLeader(shard);
113 YangInstanceIdentifier entityId = ENTITY_ID1;
114 Entity entity = new Entity(ENTITY_TYPE, entityId);
116 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
117 kit.expectMsgClass(SuccessReply.class);
119 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
120 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
124 public void testOnRegisterCandidateLocalWithNoInitialLeader() throws Exception {
125 ShardTestKit kit = new ShardTestKit(getSystem());
127 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2);
129 String peerId = newShardId("follower").toString();
130 TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId, false).
131 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
133 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
134 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
135 withDispatcher(Dispatchers.DefaultDispatcherId()));
137 YangInstanceIdentifier entityId = ENTITY_ID1;
138 Entity entity = new Entity(ENTITY_TYPE, entityId);
140 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
141 kit.expectMsgClass(SuccessReply.class);
143 // Now grant the vote so the shard becomes the leader. This should retry the commit.
144 peer.underlyingActor().grantVote = true;
146 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
147 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
151 public void testOnRegisterCandidateLocalWithNoInitialConsensus() throws Exception {
152 ShardTestKit kit = new ShardTestKit(getSystem());
154 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
155 shardTransactionCommitTimeoutInSeconds(1);
157 String peerId = newShardId("follower").toString();
158 TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId).
159 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
161 MockFollower follower = peer.underlyingActor();
163 // Drop AppendEntries so consensus isn't reached.
164 follower.dropAppendEntries = true;
166 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
167 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
168 withDispatcher(Dispatchers.DefaultDispatcherId()));
170 kit.waitUntilLeader(shard);
172 YangInstanceIdentifier entityId = ENTITY_ID1;
173 Entity entity = new Entity(ENTITY_TYPE, entityId);
175 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
176 kit.expectMsgClass(SuccessReply.class);
178 // Wait enough time for the commit to timeout.
179 Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
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;
185 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
186 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
190 public void testOnRegisterCandidateLocalWithIsolatedLeader() throws Exception {
191 ShardTestKit kit = new ShardTestKit(getSystem());
193 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
194 shardIsolatedLeaderCheckIntervalInMillis(50);
196 String peerId = newShardId("follower").toString();
197 TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId).
198 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
200 MockFollower follower = peer.underlyingActor();
202 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
203 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
204 withDispatcher(Dispatchers.DefaultDispatcherId()));
206 kit.waitUntilLeader(shard);
208 // Drop AppendEntries and wait enough time for the shard to switch to IsolatedLeader.
209 follower.dropAppendEntries = true;
210 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
212 YangInstanceIdentifier entityId = ENTITY_ID1;
213 Entity entity = new Entity(ENTITY_TYPE, entityId);
215 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
216 kit.expectMsgClass(SuccessReply.class);
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);
225 public void testOnRegisterCandidateLocalWithRemoteLeader() throws Exception {
226 ShardTestKit kit = new ShardTestKit(getSystem());
228 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
229 shardBatchedModificationCount(5);
231 String peerId = newShardId("leader").toString();
232 TestActorRef<MockLeader> peer = actorFactory.createTestActor(Props.create(MockLeader.class).
233 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
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()));
240 shard.tell(new AppendEntries(1L, peerId, -1L, -1L, Collections.<ReplicatedLogEntry>emptyList(), -1L, -1L,
241 DataStoreVersions.CURRENT_VERSION), peer);
243 shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef());
244 kit.expectMsgClass(SuccessReply.class);
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,
252 // Test with initial commit timeout and subsequent retry.
254 leader.modificationsReceived = new CountDownLatch(1);
255 leader.sendReply = false;
257 shard.tell(dataStoreContextBuilder.shardTransactionCommitTimeoutInSeconds(1).build(), ActorRef.noSender());
259 shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID2)), kit.getRef());
260 kit.expectMsgClass(SuccessReply.class);
262 assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
263 leader.modificationsReceived, 5, TimeUnit.SECONDS));
264 verifyBatchedEntityCandidate(leader.getAndClearReceivedModifications(), ENTITY_TYPE, ENTITY_ID2,
267 // Send a bunch of registration messages quickly and verify.
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));
276 shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, id)), kit.getRef());
279 assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
280 leader.modificationsReceived, 10, TimeUnit.SECONDS));
282 // Sleep a little to ensure no additional BatchedModifications are received.
284 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
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);
291 assertEquals("# modifications received", max, receivedMods.size());
295 public void testOnUnregisterCandidateLocal() throws Exception {
296 ShardTestKit kit = new ShardTestKit(getSystem());
297 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
298 kit.waitUntilLeader(shard);
300 Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
304 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
305 kit.expectMsgClass(SuccessReply.class);
307 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
308 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
312 shard.tell(new UnregisterCandidateLocal(entity), kit.getRef());
313 kit.expectMsgClass(SuccessReply.class);
315 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, "");
319 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
320 kit.expectMsgClass(SuccessReply.class);
322 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
323 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
327 public void testOwnershipChanges() throws Exception {
328 ShardTestKit kit = new ShardTestKit(getSystem());
329 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
330 kit.waitUntilLeader(shard);
332 Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
333 ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
335 // Add a remote candidate
337 String remoteMemberName1 = "remoteMember1";
338 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
342 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
343 kit.expectMsgClass(SuccessReply.class);
345 // Verify the remote candidate becomes owner
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);
351 // Add another remote candidate and verify ownership doesn't change
353 String remoteMemberName2 = "remoteMember2";
354 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
356 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
357 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
358 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
360 // Remove the second remote candidate and verify ownership doesn't change
362 deleteNode(candidatePath(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
364 verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
365 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
366 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
368 // Remove the first remote candidate and verify the local candidate becomes owner
370 deleteNode(candidatePath(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
372 verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
373 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
375 // Add the second remote candidate back and verify ownership doesn't change
377 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
379 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
380 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
381 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
383 // Unregister the local candidate and verify the second remote candidate becomes owner
385 shard.tell(new UnregisterCandidateLocal(entity), kit.getRef());
386 kit.expectMsgClass(SuccessReply.class);
388 verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
389 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
393 public void testOwnerChangesOnPeerAvailabilityChanges() throws Exception {
394 ShardTestKit kit = new ShardTestKit(getSystem());
396 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(500).shardElectionTimeoutFactor(10000);
398 String peerMemberName1 = "peerMember1";
399 String peerMemberName2 = "peerMember2";
401 ShardIdentifier leaderId = newShardId(LOCAL_MEMBER_NAME);
402 ShardIdentifier peerId1 = newShardId(peerMemberName1);
403 ShardIdentifier peerId2 = newShardId(peerMemberName2);
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());
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());
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);
419 kit.waitUntilLeader(leader);
421 // Send PeerDown and PeerUp with no entities
423 leader.tell(new PeerDown(peerMemberName2, peerId2.toString()), ActorRef.noSender());
424 leader.tell(new PeerUp(peerMemberName2, peerId2.toString()), ActorRef.noSender());
426 // Add candidates for entity1 with the local leader as the owner
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);
432 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, peerMemberName2), kit);
433 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
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);
439 // Add candidates for entity2 with peerMember2 as the owner
441 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName2), kit);
442 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
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);
448 // Add candidates for entity3 with peerMember2 as the owner.
450 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName2), kit);
451 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
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);
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);
461 // Add only candidate peerMember2 for entity4.
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);
467 // Add only candidate peerMember1 for entity5.
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);
473 // Kill peerMember2 and send PeerDown - the entities (2, 3, 4) owned by peerMember2 should get a new
477 peer2.tell(PoisonPill.getInstance(), ActorRef.noSender());
478 kit.expectMsgClass(JavaTestKit.duration("5 seconds"), Terminated.class);
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());
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);
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);
496 // Reinstate peerMember2 - no owners should change
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());
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, "");
511 // Add back candidate peerMember2 for entities 1, 2, & 3.
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);
523 // Kill peerMember1 and send PeerDown - entity 2 should get a new owner selected
525 peer1.tell(PoisonPill.getInstance(), ActorRef.noSender());
526 leader.tell(new PeerDown(peerMemberName1, peerId1.toString()), ActorRef.noSender());
528 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
530 // Verify the reinstated peerMember2 is fully synced.
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);
537 // Reinstate peerMember1 and verify no owner changes
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());
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);
549 // Verify the reinstated peerMember1 is fully synced.
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);
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.
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());
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);
567 kit.waitUntilLeader(peer2);
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);
576 public void testLocalCandidateRemovedWithCandidateRegistered() throws Exception {
577 ShardTestKit kit = new ShardTestKit(getSystem());
579 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(10000);
580 ShardIdentifier leaderId = newShardId("leader");
581 ShardIdentifier localId = newShardId(LOCAL_MEMBER_NAME);
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()));
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);
593 kit.waitUntilLeader(leader);
595 shard.tell(new PeerAddressResolved(leaderId.toString(), leader.path().toString()), ActorRef.noSender());
597 Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
598 EntityOwnershipListener listener = mock(EntityOwnershipListener.class);
600 shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
601 kit.expectMsgClass(SuccessReply.class);
603 // Register local candidate
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));
611 // Simulate a replicated commit from the leader to remove the local candidate that would occur after a
612 // network partition is healed.
614 leader.tell(new PeerDown(LOCAL_MEMBER_NAME, localId.toString()), ActorRef.noSender());
616 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, true, false, false));
618 // Since the the shard has a local candidate registered, it should re-add its candidate to the entity.
620 verifyCommittedEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME);
621 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true));
623 // Unregister the local candidate and verify it's removed and no re-added.
625 shard.tell(new UnregisterCandidateLocal(entity), kit.getRef());
626 kit.expectMsgClass(SuccessReply.class);
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);
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();
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);
649 shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
650 kit.expectMsgClass(SuccessReply.class);
652 // Register a couple candidates for the desired entity type and verify listener is notified.
654 shard.tell(new RegisterCandidateLocal(entity1), kit.getRef());
655 kit.expectMsgClass(SuccessReply.class);
657 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, false, true, true));
659 shard.tell(new RegisterCandidateLocal(entity2), kit.getRef());
660 kit.expectMsgClass(SuccessReply.class);
662 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity2, false, true, true));
665 // Register another candidate for another entity type and verify listener is not notified.
667 shard.tell(new RegisterCandidateLocal(entity4), kit.getRef());
668 kit.expectMsgClass(SuccessReply.class);
670 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
671 verify(listener, never()).ownershipChanged(ownershipChange(entity4));
673 // Register remote candidate for entity1
675 String remoteMemberName = "remoteMember";
676 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, entity1.getId(), remoteMemberName),
678 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entity1.getId(), remoteMemberName);
680 // Unregister the local candidate for entity1 and verify listener is notified
682 shard.tell(new UnregisterCandidateLocal(entity1), kit.getRef());
683 kit.expectMsgClass(SuccessReply.class);
685 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, true, false, true));
688 // Unregister the listener, add a candidate for entity3 and verify listener isn't notified
690 shard.tell(new UnregisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
691 kit.expectMsgClass(SuccessReply.class);
693 shard.tell(new RegisterCandidateLocal(entity3), kit.getRef());
694 kit.expectMsgClass(SuccessReply.class);
696 verifyOwner(shard, ENTITY_TYPE, entity3.getId(), LOCAL_MEMBER_NAME);
697 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
698 verify(listener, never()).ownershipChanged(any(EntityOwnershipChange.class));
700 // Re-register the listener and verify it gets notified of currently owned entities
704 shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
705 kit.expectMsgClass(SuccessReply.class);
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));
714 private static void commitModification(TestActorRef<EntityOwnershipShard> shard, NormalizedNode<?, ?> node,
715 JavaTestKit sender) {
716 BatchedModifications modifications = newBatchedModifications();
717 modifications.addModification(new MergeModification(ENTITY_OWNERS_PATH, node));
719 shard.tell(modifications, sender.getRef());
720 sender.expectMsgClass(CommitTransactionReply.SERIALIZABLE_CLASS);
723 private static 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;
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<?,?>>() {
736 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
738 return AbstractShardTest.readStore(shard, path);
739 } catch(Exception e) {
740 throw new AssertionError("Failed to read " + path, e);
746 private void verifyCommittedEntityCandidate(final TestActorRef<EntityOwnershipShard> shard, String entityType,
747 YangInstanceIdentifier entityId, String candidateName) {
748 verifyEntityCandidate(entityType, entityId, candidateName, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
750 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
752 return AbstractShardTest.readStore(shard, path);
753 } catch(Exception e) {
754 throw new AssertionError("Failed to read " + path, e);
760 private void verifyNoEntityCandidate(final TestActorRef<EntityOwnershipShard> shard, String entityType,
761 YangInstanceIdentifier entityId, String candidateName) {
762 verifyEntityCandidate(entityType, entityId, candidateName, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
764 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
766 return AbstractShardTest.readStore(shard, path);
767 } catch(Exception e) {
768 throw new AssertionError("Failed to read " + path, e);
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);
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);
787 private static void verifyOwner(final TestActorRef<EntityOwnershipShard> shard, String entityType,
788 YangInstanceIdentifier entityId, String localMemberName) {
789 verifyOwner(localMemberName, entityType, entityId, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
791 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
793 return AbstractShardTest.readStore(shard, path);
794 } catch(Exception e) {
801 private Props newShardProps() {
802 return newShardProps(Collections.<String,String>emptyMap());
805 private Props newShardProps(Map<String,String> peers) {
806 return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME);
809 private Props newShardProps(ShardIdentifier shardId, Map<String,String> peers, String memberName) {
810 return EntityOwnershipShard.props(shardId, peers, dataStoreContextBuilder.build(),
811 SCHEMA_CONTEXT, memberName);
814 private static ShardIdentifier newShardId(String memberName) {
815 return ShardIdentifier.builder().memberName(memberName).shardName("entity-ownership").
816 type("operational" + NEXT_SHARD_NUM.getAndIncrement()).build();
819 public static class TestEntityOwnershipShard extends EntityOwnershipShard {
821 TestEntityOwnershipShard(ShardIdentifier name, Map<String, String> peerAddresses,
822 DatastoreContext datastoreContext) {
823 super(name, peerAddresses, datastoreContext, SCHEMA_CONTEXT, LOCAL_MEMBER_NAME);
827 public void onReceiveCommand(Object message) throws Exception {
828 if(!(message instanceof ElectionTimeout)) {
829 super.onReceiveCommand(message);
836 public static class MockFollower extends UntypedActor {
837 volatile boolean grantVote;
838 volatile boolean dropAppendEntries;
839 private final String myId;
841 public MockFollower(String myId) {
845 public MockFollower(String myId, boolean grantVote) {
847 this.grantVote = grantVote;
851 public void onReceive(Object message) {
852 if(message instanceof RequestVote) {
854 getSender().tell(new RequestVoteReply(((RequestVote)message).getTerm(), true), getSelf());
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();
866 getSender().tell(new AppendEntriesReply(myId, req.getTerm(), true, lastIndex, req.getTerm(),
867 DataStoreVersions.CURRENT_VERSION), getSelf());
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;
880 public void onReceive(Object message) {
881 if(message instanceof BatchedModifications) {
883 Uninterruptibles.sleepUninterruptibly(delay, TimeUnit.MILLISECONDS);
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();
895 getSender().tell(CommitTransactionReply.INSTANCE.toSerializable(), getSelf());
902 List<Modification> getAndClearReceivedModifications() {
903 synchronized (receivedModifications) {
904 List<Modification> ret = new ArrayList<>(receivedModifications);
905 receivedModifications.clear();