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.times;
18 import static org.mockito.Mockito.verify;
19 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.ENTITY_OWNERS_PATH;
20 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.candidatePath;
21 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.entityOwnersWithCandidate;
22 import akka.actor.ActorRef;
23 import akka.actor.PoisonPill;
24 import akka.actor.Props;
25 import akka.actor.Terminated;
26 import akka.actor.UntypedActor;
27 import akka.dispatch.Dispatchers;
28 import akka.testkit.JavaTestKit;
29 import akka.testkit.TestActorRef;
30 import com.google.common.base.Function;
31 import com.google.common.collect.ImmutableMap;
32 import com.google.common.util.concurrent.Uninterruptibles;
33 import java.util.ArrayList;
34 import java.util.Collections;
35 import java.util.List;
37 import java.util.concurrent.CountDownLatch;
38 import java.util.concurrent.TimeUnit;
39 import java.util.concurrent.atomic.AtomicInteger;
40 import org.junit.After;
41 import org.junit.Test;
42 import org.opendaylight.controller.cluster.access.concepts.MemberName;
43 import org.opendaylight.controller.cluster.datastore.AbstractShardTest;
44 import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
45 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
46 import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
47 import org.opendaylight.controller.cluster.datastore.ShardDataTree;
48 import org.opendaylight.controller.cluster.datastore.ShardTestKit;
49 import org.opendaylight.controller.cluster.datastore.entityownership.messages.RegisterCandidateLocal;
50 import org.opendaylight.controller.cluster.datastore.entityownership.messages.RegisterListenerLocal;
51 import org.opendaylight.controller.cluster.datastore.entityownership.messages.UnregisterCandidateLocal;
52 import org.opendaylight.controller.cluster.datastore.entityownership.messages.UnregisterListenerLocal;
53 import org.opendaylight.controller.cluster.datastore.entityownership.selectionstrategy.EntityOwnerSelectionStrategyConfig;
54 import org.opendaylight.controller.cluster.datastore.entityownership.selectionstrategy.LastCandidateSelectionStrategy;
55 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
56 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
57 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
58 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
59 import org.opendaylight.controller.cluster.datastore.messages.PeerDown;
60 import org.opendaylight.controller.cluster.datastore.messages.PeerUp;
61 import org.opendaylight.controller.cluster.datastore.messages.SuccessReply;
62 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
63 import org.opendaylight.controller.cluster.datastore.modification.Modification;
64 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
65 import org.opendaylight.controller.cluster.raft.TestActorFactory;
66 import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
67 import org.opendaylight.controller.cluster.raft.base.messages.TimeoutNow;
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.mdsal.eos.dom.api.DOMEntity;
74 import org.opendaylight.mdsal.eos.dom.api.DOMEntityOwnershipChange;
75 import org.opendaylight.mdsal.eos.dom.api.DOMEntityOwnershipListener;
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 YangInstanceIdentifier ENTITY_ID5 =
97 YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity5"));
98 private static final SchemaContext SCHEMA_CONTEXT = SchemaContextHelper.entityOwners();
99 private static final AtomicInteger NEXT_SHARD_NUM = new AtomicInteger();
100 private static final String LOCAL_MEMBER_NAME = "member-1";
102 private final Builder dataStoreContextBuilder = DatastoreContext.newBuilder();
103 private final TestActorFactory actorFactory = new TestActorFactory(getSystem());
106 public void tearDown() {
107 actorFactory.close();
111 public void testOnRegisterCandidateLocal() throws Exception {
112 ShardTestKit kit = new ShardTestKit(getSystem());
114 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
116 ShardTestKit.waitUntilLeader(shard);
118 YangInstanceIdentifier entityId = ENTITY_ID1;
119 DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId);
121 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
122 kit.expectMsgClass(SuccessReply.class);
124 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
125 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
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 DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId);
145 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
146 kit.expectMsgClass(SuccessReply.class);
148 // Now grant the vote so the shard becomes the leader. This should retry the commit.
149 peer.underlyingActor().grantVote = true;
151 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
152 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
156 public void testOnRegisterCandidateLocalWithNoInitialConsensus() throws Exception {
157 ShardTestKit kit = new ShardTestKit(getSystem());
159 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
160 shardTransactionCommitTimeoutInSeconds(1);
162 String peerId = newShardId("follower").toString();
163 TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId).
164 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
166 MockFollower follower = peer.underlyingActor();
168 // Drop AppendEntries so consensus isn't reached.
169 follower.dropAppendEntries = true;
171 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
172 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
173 withDispatcher(Dispatchers.DefaultDispatcherId()));
175 ShardTestKit.waitUntilLeader(shard);
177 YangInstanceIdentifier entityId = ENTITY_ID1;
178 DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId);
180 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
181 kit.expectMsgClass(SuccessReply.class);
183 // Wait enough time for the commit to timeout.
184 Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
186 // Resume AppendEntries - the follower should ack the commit which should then result in the candidate
187 // write being applied to the state.
188 follower.dropAppendEntries = false;
190 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
191 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
195 public void testOnRegisterCandidateLocalWithIsolatedLeader() throws Exception {
196 ShardTestKit kit = new ShardTestKit(getSystem());
198 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
199 shardIsolatedLeaderCheckIntervalInMillis(50);
201 String peerId = newShardId("follower").toString();
202 TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId).
203 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
205 MockFollower follower = peer.underlyingActor();
207 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
208 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
209 withDispatcher(Dispatchers.DefaultDispatcherId()));
211 ShardTestKit.waitUntilLeader(shard);
213 // Drop AppendEntries and wait enough time for the shard to switch to IsolatedLeader.
214 follower.dropAppendEntries = true;
215 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
217 YangInstanceIdentifier entityId = ENTITY_ID1;
218 DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId);
220 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
221 kit.expectMsgClass(SuccessReply.class);
223 // Resume AppendEntries - the candidate write should now be committed.
224 follower.dropAppendEntries = false;
225 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
226 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
230 public void testOnRegisterCandidateLocalWithRemoteLeader() throws Exception {
231 ShardTestKit kit = new ShardTestKit(getSystem());
233 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
234 shardBatchedModificationCount(5);
236 String peerId = newShardId("leader").toString();
237 TestActorRef<MockLeader> peer = actorFactory.createTestActor(Props.create(MockLeader.class).
238 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
240 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(Props.create(
241 TestEntityOwnershipShard.class, newShardId(LOCAL_MEMBER_NAME),
242 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build(),
243 dataStoreContextBuilder.build()).withDispatcher(Dispatchers.DefaultDispatcherId()));
245 shard.tell(new AppendEntries(1L, peerId, -1L, -1L, Collections.<ReplicatedLogEntry>emptyList(), -1L, -1L,
246 DataStoreVersions.CURRENT_VERSION), peer);
248 shard.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef());
249 kit.expectMsgClass(SuccessReply.class);
251 MockLeader leader = peer.underlyingActor();
252 assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
253 leader.modificationsReceived, 5, TimeUnit.SECONDS));
254 verifyBatchedEntityCandidate(leader.getAndClearReceivedModifications(), ENTITY_TYPE, ENTITY_ID1,
257 // Test with initial commit timeout and subsequent retry.
259 leader.modificationsReceived = new CountDownLatch(1);
260 leader.sendReply = false;
262 shard.tell(dataStoreContextBuilder.shardTransactionCommitTimeoutInSeconds(1).build(), ActorRef.noSender());
264 shard.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, ENTITY_ID2)), kit.getRef());
265 kit.expectMsgClass(SuccessReply.class);
267 assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
268 leader.modificationsReceived, 5, TimeUnit.SECONDS));
269 verifyBatchedEntityCandidate(leader.getAndClearReceivedModifications(), ENTITY_TYPE, ENTITY_ID2,
272 // Send a bunch of registration messages quickly and verify.
276 leader.modificationsReceived = new CountDownLatch(max);
277 List<YangInstanceIdentifier> entityIds = new ArrayList<>();
278 for(int i = 1; i <= max; i++) {
279 YangInstanceIdentifier id = YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "test" + i));
281 shard.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, id)), kit.getRef());
284 assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
285 leader.modificationsReceived, 10, TimeUnit.SECONDS));
287 // Sleep a little to ensure no additional BatchedModifications are received.
289 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
291 List<Modification> receivedMods = leader.getAndClearReceivedModifications();
292 for(int i = 0; i < max; i++) {
293 verifyBatchedEntityCandidate(receivedMods.get(i), ENTITY_TYPE, entityIds.get(i), LOCAL_MEMBER_NAME);
296 assertEquals("# modifications received", max, receivedMods.size());
300 public void testOnUnregisterCandidateLocal() throws Exception {
301 ShardTestKit kit = new ShardTestKit(getSystem());
302 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
303 ShardTestKit.waitUntilLeader(shard);
305 DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
309 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
310 kit.expectMsgClass(SuccessReply.class);
312 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
313 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
317 shard.tell(new UnregisterCandidateLocal(entity), kit.getRef());
318 kit.expectMsgClass(SuccessReply.class);
320 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, "");
324 shard.tell(new RegisterCandidateLocal(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);
332 public void testOwnershipChanges() throws Exception {
333 ShardTestKit kit = new ShardTestKit(getSystem());
334 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
335 ShardTestKit.waitUntilLeader(shard);
337 DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
338 ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
340 // Add a remote candidate
342 String remoteMemberName1 = "remoteMember1";
343 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
347 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
348 kit.expectMsgClass(SuccessReply.class);
350 // Verify the remote candidate becomes owner
352 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
353 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
354 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
356 // Add another remote candidate and verify ownership doesn't change
358 String remoteMemberName2 = "remoteMember2";
359 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
361 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
362 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
363 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
365 // Remove the second remote candidate and verify ownership doesn't change
367 deleteNode(candidatePath(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
369 verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
370 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
371 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
373 // Remove the first remote candidate and verify the local candidate becomes owner
375 deleteNode(candidatePath(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
377 verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
378 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
380 // Add the second remote candidate back and verify ownership doesn't change
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, LOCAL_MEMBER_NAME);
388 // Unregister the local candidate and verify the second remote candidate becomes owner
390 shard.tell(new UnregisterCandidateLocal(entity), kit.getRef());
391 kit.expectMsgClass(SuccessReply.class);
393 verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
394 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
398 public void testOwnerChangesOnPeerAvailabilityChanges() throws Exception {
399 ShardTestKit kit = new ShardTestKit(getSystem());
401 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(500).shardElectionTimeoutFactor(10000);
403 String peerMemberName1 = "peerMember1";
404 String peerMemberName2 = "peerMember2";
406 ShardIdentifier leaderId = newShardId(LOCAL_MEMBER_NAME);
407 ShardIdentifier peerId1 = newShardId(peerMemberName1);
408 ShardIdentifier peerId2 = newShardId(peerMemberName2);
410 TestActorRef<EntityOwnershipShard> peer1 = actorFactory.createTestActor(newShardProps(peerId1,
411 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId2.toString(), "").build(),
412 peerMemberName1, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
414 TestActorRef<EntityOwnershipShard> peer2 = actorFactory.createTestActor(newShardProps(peerId2,
415 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId1.toString(), "").build(),
416 peerMemberName2, EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString());
418 TestActorRef<EntityOwnershipShard> leader = actorFactory.createTestActor(newShardProps(leaderId,
419 ImmutableMap.<String, String>builder().put(peerId1.toString(), peer1.path().toString()).
420 put(peerId2.toString(), peer2.path().toString()).build(), LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build()).
421 withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString());
422 leader.tell(TimeoutNow.INSTANCE, leader);
424 ShardTestKit.waitUntilLeader(leader);
426 // Send PeerDown and PeerUp with no entities
428 leader.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
429 leader.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
431 // Add candidates for entity1 with the local leader as the owner
433 leader.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef());
434 kit.expectMsgClass(SuccessReply.class);
435 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
437 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, peerMemberName2), kit);
438 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
440 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, peerMemberName1), kit);
441 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName1);
442 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
444 // Add candidates for entity2 with peerMember2 as the owner
446 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName2), kit);
447 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
449 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName1), kit);
450 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
451 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
453 // Add candidates for entity3 with peerMember2 as the owner.
455 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName2), kit);
456 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
458 leader.tell(new RegisterCandidateLocal(new DOMEntity(ENTITY_TYPE, ENTITY_ID3)), kit.getRef());
459 kit.expectMsgClass(SuccessReply.class);
460 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
462 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName1), kit);
463 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName1);
464 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
466 // Add only candidate peerMember2 for entity4.
468 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID4, peerMemberName2), kit);
469 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
470 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
472 // Add only candidate peerMember1 for entity5.
474 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID5, peerMemberName1), kit);
475 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID5, peerMemberName1);
476 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID5, peerMemberName1);
478 // Kill peerMember2 and send PeerDown - the entities (2, 3, 4) owned by peerMember2 should get a new
482 peer2.tell(PoisonPill.getInstance(), ActorRef.noSender());
483 kit.expectMsgClass(JavaTestKit.duration("5 seconds"), Terminated.class);
486 leader.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
487 // Send PeerDown again - should be noop
488 leader.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
489 peer1.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
491 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, ""); // no other candidates so should clear
492 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
493 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
494 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
496 verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
497 verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
498 verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
499 verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
501 // Reinstate peerMember2 - no owners should change
503 peer2 = actorFactory.createTestActor(newShardProps(peerId2,
504 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId1.toString(), "").build(),
505 peerMemberName2, EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString());
506 leader.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
507 // Send PeerUp again - should be noop
508 leader.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
509 peer1.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
511 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
512 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
513 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
514 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, "");
516 // Add back candidate peerMember2 for entities 1, 2, & 3.
518 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, peerMemberName2), kit);
519 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName2), kit);
520 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName2), kit);
521 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
522 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
523 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
524 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
525 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
526 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
528 // Kill peerMember1 and send PeerDown - entity 2 should get a new owner selected
530 peer1.tell(PoisonPill.getInstance(), ActorRef.noSender());
531 leader.tell(new PeerDown(peerId1.getMemberName(), peerId1.toString()), ActorRef.noSender());
533 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
535 // Verify the reinstated peerMember2 is fully synced.
537 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, "");
538 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
539 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
540 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
542 // Reinstate peerMember1 and verify no owner changes
544 peer1 = actorFactory.createTestActor(newShardProps(peerId1,
545 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId2.toString(), "").build(),
546 peerMemberName1, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
547 leader.tell(new PeerUp(peerId1.getMemberName(), peerId1.toString()), ActorRef.noSender());
549 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, "");
550 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
551 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
552 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
554 // Verify the reinstated peerMember1 is fully synced.
556 verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID4, "");
557 verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
558 verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
559 verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
561 // Kill the local leader and elect peer2 the leader. This should cause a new owner to be selected for
562 // the entities (1 and 3) previously owned by the local leader member.
564 peer2.tell(new PeerAddressResolved(peerId1.toString(), peer1.path().toString()), ActorRef.noSender());
565 peer2.tell(new PeerUp(leaderId.getMemberName(), leaderId.toString()), ActorRef.noSender());
566 peer2.tell(new PeerUp(peerId1.getMemberName(), peerId1.toString()), ActorRef.noSender());
568 leader.tell(PoisonPill.getInstance(), ActorRef.noSender());
569 peer2.tell(new PeerDown(leaderId.getMemberName(), leaderId.toString()), ActorRef.noSender());
570 peer2.tell(TimeoutNow.INSTANCE, peer2);
572 ShardTestKit.waitUntilLeader(peer2);
574 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, "");
575 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
576 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
577 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
581 public void testLocalCandidateRemovedWithCandidateRegistered() throws Exception {
582 ShardTestKit kit = new ShardTestKit(getSystem());
584 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(10000);
585 ShardIdentifier leaderId = newShardId("leader");
586 ShardIdentifier localId = newShardId(LOCAL_MEMBER_NAME);
588 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(Props.create(
589 TestEntityOwnershipShard.class, localId,
590 ImmutableMap.<String, String>builder().put(leaderId.toString(), "".toString()).build(),
591 dataStoreContextBuilder.build()).withDispatcher(Dispatchers.DefaultDispatcherId()));
593 TestActorRef<EntityOwnershipShard> leader = actorFactory.createTestActor(newShardProps(leaderId,
594 ImmutableMap.<String, String>builder().put(localId.toString(), shard.path().toString()).build(),
595 LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString());
596 leader.tell(TimeoutNow.INSTANCE, leader);
598 ShardTestKit.waitUntilLeader(leader);
600 shard.tell(new PeerAddressResolved(leaderId.toString(), leader.path().toString()), ActorRef.noSender());
602 DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
603 DOMEntityOwnershipListener listener = mock(DOMEntityOwnershipListener.class);
605 shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
606 kit.expectMsgClass(SuccessReply.class);
608 // Register local candidate
610 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
611 kit.expectMsgClass(SuccessReply.class);
612 verifyCommittedEntityCandidate(shard, entity.getType(), entity.getIdentifier(), LOCAL_MEMBER_NAME);
613 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true));
616 // Simulate a replicated commit from the leader to remove the local candidate that would occur after a
617 // network partition is healed.
619 leader.tell(new PeerDown(localId.getMemberName(), localId.toString()), ActorRef.noSender());
621 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, true, false, false));
623 // Since the the shard has a local candidate registered, it should re-add its candidate to the entity.
625 verifyCommittedEntityCandidate(shard, entity.getType(), entity.getIdentifier(), LOCAL_MEMBER_NAME);
626 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true));
628 // Unregister the local candidate and verify it's removed and no re-added.
630 shard.tell(new UnregisterCandidateLocal(entity), kit.getRef());
631 kit.expectMsgClass(SuccessReply.class);
633 verifyNoEntityCandidate(shard, entity.getType(), entity.getIdentifier(), LOCAL_MEMBER_NAME);
634 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
635 verifyNoEntityCandidate(shard, entity.getType(), entity.getIdentifier(), LOCAL_MEMBER_NAME);
639 public void testListenerRegistration() throws Exception {
640 ShardTestKit kit = new ShardTestKit(getSystem());
641 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
642 ShardTestKit.waitUntilLeader(shard);
643 ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
645 String otherEntityType = "otherEntityType";
646 DOMEntity entity1 = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
647 DOMEntity entity2 = new DOMEntity(ENTITY_TYPE, ENTITY_ID2);
648 DOMEntity entity3 = new DOMEntity(ENTITY_TYPE, ENTITY_ID3);
649 DOMEntity entity4 = new DOMEntity(otherEntityType, ENTITY_ID3);
650 DOMEntityOwnershipListener listener = mock(DOMEntityOwnershipListener.class);
654 shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
655 kit.expectMsgClass(SuccessReply.class);
657 // Register a couple candidates for the desired entity type and verify listener is notified.
659 shard.tell(new RegisterCandidateLocal(entity1), kit.getRef());
660 kit.expectMsgClass(SuccessReply.class);
662 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, false, true, true));
664 shard.tell(new RegisterCandidateLocal(entity2), kit.getRef());
665 kit.expectMsgClass(SuccessReply.class);
667 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity2, false, true, true));
670 // Register another candidate for another entity type and verify listener is not notified.
672 shard.tell(new RegisterCandidateLocal(entity4), kit.getRef());
673 kit.expectMsgClass(SuccessReply.class);
675 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
676 verify(listener, never()).ownershipChanged(ownershipChange(entity4));
678 // Register remote candidate for entity1
680 String remoteMemberName = "remoteMember";
681 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, entity1.getIdentifier(), remoteMemberName),
683 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entity1.getIdentifier(), remoteMemberName);
685 // Unregister the local candidate for entity1 and verify listener is notified
687 shard.tell(new UnregisterCandidateLocal(entity1), kit.getRef());
688 kit.expectMsgClass(SuccessReply.class);
690 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, true, false, true));
693 // Unregister the listener, add a candidate for entity3 and verify listener isn't notified
695 shard.tell(new UnregisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
696 kit.expectMsgClass(SuccessReply.class);
698 shard.tell(new RegisterCandidateLocal(entity3), kit.getRef());
699 kit.expectMsgClass(SuccessReply.class);
701 verifyOwner(shard, ENTITY_TYPE, entity3.getIdentifier(), LOCAL_MEMBER_NAME);
702 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
703 verify(listener, never()).ownershipChanged(any(DOMEntityOwnershipChange.class));
705 // Re-register the listener and verify it gets notified of currently owned entities
709 shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
710 kit.expectMsgClass(SuccessReply.class);
712 verify(listener, timeout(5000).times(2)).ownershipChanged(or(ownershipChange(entity2, false, true, true),
713 ownershipChange(entity3, false, true, true)));
714 Uninterruptibles.sleepUninterruptibly(300, TimeUnit.MILLISECONDS);
715 verify(listener, never()).ownershipChanged(ownershipChange(entity4));
716 verify(listener, times(1)).ownershipChanged(ownershipChange(entity1));
719 private static void commitModification(TestActorRef<EntityOwnershipShard> shard, NormalizedNode<?, ?> node,
720 JavaTestKit sender) {
721 BatchedModifications modifications = newBatchedModifications();
722 modifications.addModification(new MergeModification(ENTITY_OWNERS_PATH, node));
724 shard.tell(modifications, sender.getRef());
725 sender.expectMsgClass(CommitTransactionReply.class);
728 private static BatchedModifications newBatchedModifications() {
729 BatchedModifications modifications = new BatchedModifications(nextTransactionId(), DataStoreVersions.CURRENT_VERSION);
730 modifications.setDoCommitOnReady(true);
731 modifications.setReady(true);
732 modifications.setTotalMessagesSent(1);
733 return modifications;
736 private void verifyEntityCandidateRemoved(final TestActorRef<EntityOwnershipShard> shard, String entityType,
737 YangInstanceIdentifier entityId, String candidateName) {
738 verifyNodeRemoved(candidatePath(entityType, entityId, candidateName),
739 new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
741 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
743 return AbstractShardTest.readStore(shard, path);
744 } catch(Exception e) {
745 throw new AssertionError("Failed to read " + path, e);
751 private void verifyCommittedEntityCandidate(final TestActorRef<EntityOwnershipShard> shard, String entityType,
752 YangInstanceIdentifier entityId, String candidateName) {
753 verifyEntityCandidate(entityType, entityId, candidateName, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
755 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
757 return AbstractShardTest.readStore(shard, path);
758 } catch(Exception e) {
759 throw new AssertionError("Failed to read " + path, e);
765 private void verifyNoEntityCandidate(final TestActorRef<EntityOwnershipShard> shard, String entityType,
766 YangInstanceIdentifier entityId, String candidateName) {
767 verifyEntityCandidate(entityType, entityId, candidateName, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
769 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
771 return AbstractShardTest.readStore(shard, path);
772 } catch(Exception e) {
773 throw new AssertionError("Failed to read " + path, e);
779 private void verifyBatchedEntityCandidate(List<Modification> mods, String entityType,
780 YangInstanceIdentifier entityId, String candidateName) throws Exception {
781 assertEquals("BatchedModifications size", 1, mods.size());
782 verifyBatchedEntityCandidate(mods.get(0), entityType, entityId, candidateName);
785 private void verifyBatchedEntityCandidate(Modification mod, String entityType,
786 YangInstanceIdentifier entityId, String candidateName) throws Exception {
787 assertEquals("Modification type", MergeModification.class, mod.getClass());
788 verifyEntityCandidate(((MergeModification)mod).getData(), entityType,
789 entityId, candidateName, true);
792 private static void verifyOwner(final TestActorRef<EntityOwnershipShard> shard, String entityType,
793 YangInstanceIdentifier entityId, String localMemberName) {
794 verifyOwner(localMemberName, entityType, entityId, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
796 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
798 return AbstractShardTest.readStore(shard, path);
799 } catch(Exception e) {
806 private Props newShardProps() {
807 return newShardProps(Collections.<String,String>emptyMap());
810 private Props newShardProps(EntityOwnerSelectionStrategyConfig strategyConfig, Map<String, String> peers) {
811 return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, strategyConfig);
814 private Props newShardProps(Map<String, String> peers) {
815 return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build());
818 private Props newShardProps(ShardIdentifier shardId, Map<String,String> peers, String memberName,
819 EntityOwnerSelectionStrategyConfig config) {
820 return EntityOwnershipShard.newBuilder().id(shardId).peerAddresses(peers).
821 datastoreContext(dataStoreContextBuilder.build()).schemaContext(SCHEMA_CONTEXT).
822 localMemberName(MemberName.forName(memberName)).ownerSelectionStrategyConfig(config).props()
823 .withDispatcher(Dispatchers.DefaultDispatcherId());
826 private static ShardIdentifier newShardId(String memberName) {
827 return ShardIdentifier.create("entity-ownership", MemberName.forName(memberName),
828 "operational" + NEXT_SHARD_NUM.getAndIncrement());
831 private static class TestEntityOwnershipShard extends EntityOwnershipShard {
833 TestEntityOwnershipShard(ShardIdentifier name, Map<String, String> peerAddresses,
834 DatastoreContext datastoreContext) {
835 super(newBuilder().id(name).peerAddresses(peerAddresses).datastoreContext(datastoreContext).
836 schemaContext(SCHEMA_CONTEXT).localMemberName(MemberName.forName(LOCAL_MEMBER_NAME)));
840 public void handleCommand(Object message) {
841 if(!(message instanceof ElectionTimeout)) {
842 super.handleCommand(message);
847 private static class MockFollower extends UntypedActor {
848 volatile boolean grantVote;
849 volatile boolean dropAppendEntries;
850 private final String myId;
852 @SuppressWarnings("unused")
853 public MockFollower(String myId) {
857 public MockFollower(String myId, boolean grantVote) {
859 this.grantVote = grantVote;
863 public void onReceive(Object message) {
864 if(message instanceof RequestVote) {
866 getSender().tell(new RequestVoteReply(((RequestVote)message).getTerm(), true), getSelf());
868 } else if(message instanceof AppendEntries) {
869 if(!dropAppendEntries) {
870 AppendEntries req = (AppendEntries) message;
871 long lastIndex = req.getLeaderCommit();
872 if (req.getEntries().size() > 0) {
873 for(ReplicatedLogEntry entry : req.getEntries()) {
874 lastIndex = entry.getIndex();
878 getSender().tell(new AppendEntriesReply(myId, req.getTerm(), true, lastIndex, req.getTerm(),
879 DataStoreVersions.CURRENT_VERSION), getSelf());
887 public void testDelayedEntityOwnerSelectionWhenMaxPeerRequestsReceived() throws Exception {
888 ShardTestKit kit = new ShardTestKit(getSystem());
889 EntityOwnerSelectionStrategyConfig.Builder builder
890 = EntityOwnerSelectionStrategyConfig.newBuilder().addStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500);
892 String peerId = newShardId("follower").toString();
893 TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId, false).
894 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
896 peer.underlyingActor().grantVote = true;
898 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(builder.build(),
899 ImmutableMap.of(peerId.toString(), peer.path().toString())));
900 ShardTestKit.waitUntilLeader(shard);
902 DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
903 ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
905 // Add a remote candidate
907 String remoteMemberName1 = "follower";
908 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
912 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
913 kit.expectMsgClass(SuccessReply.class);
915 // Verify the local candidate becomes owner
917 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
918 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
919 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
923 public void testDelayedEntityOwnerSelection() throws Exception {
924 ShardTestKit kit = new ShardTestKit(getSystem());
925 EntityOwnerSelectionStrategyConfig.Builder builder
926 = EntityOwnerSelectionStrategyConfig.newBuilder().addStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500);
928 String follower1Id = newShardId("follower1").toString();
929 TestActorRef<MockFollower> follower1 = actorFactory.createTestActor(Props.create(MockFollower.class, follower1Id, false).
930 withDispatcher(Dispatchers.DefaultDispatcherId()), follower1Id);
932 follower1.underlyingActor().grantVote = true;
934 String follower2Id = newShardId("follower").toString();
935 TestActorRef<MockFollower> follower2 = actorFactory.createTestActor(Props.create(MockFollower.class, follower2Id, false).
936 withDispatcher(Dispatchers.DefaultDispatcherId()), follower2Id);
938 follower2.underlyingActor().grantVote = true;
941 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(builder.build(),
942 ImmutableMap.of(follower1Id.toString(), follower2.path().toString(), follower2Id.toString(), follower2.path().toString())));
943 ShardTestKit.waitUntilLeader(shard);
945 DOMEntity entity = new DOMEntity(ENTITY_TYPE, ENTITY_ID1);
946 ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
948 // Add a remote candidate
950 String remoteMemberName1 = "follower";
951 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
955 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
956 kit.expectMsgClass(SuccessReply.class);
958 // Verify the local candidate becomes owner
960 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
961 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
962 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
965 private static class MockLeader extends UntypedActor {
966 volatile CountDownLatch modificationsReceived = new CountDownLatch(1);
967 List<Modification> receivedModifications = new ArrayList<>();
968 volatile boolean sendReply = true;
972 public void onReceive(Object message) {
973 if(message instanceof BatchedModifications) {
975 Uninterruptibles.sleepUninterruptibly(delay, TimeUnit.MILLISECONDS);
979 BatchedModifications mods = (BatchedModifications) message;
980 synchronized (receivedModifications) {
981 for(int i = 0; i < mods.getModifications().size(); i++) {
982 receivedModifications.add(mods.getModifications().get(i));
983 modificationsReceived.countDown();
987 getSender().tell(CommitTransactionReply.instance(DataStoreVersions.CURRENT_VERSION).
988 toSerializable(), getSelf());
995 List<Modification> getAndClearReceivedModifications() {
996 synchronized (receivedModifications) {
997 List<Modification> ret = new ArrayList<>(receivedModifications);
998 receivedModifications.clear();