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.messages.AppendEntries;
68 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
69 import org.opendaylight.controller.cluster.raft.messages.RequestVote;
70 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
71 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
72 import org.opendaylight.controller.md.sal.common.api.clustering.Entity;
73 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipChange;
74 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListener;
75 import org.opendaylight.yangtools.yang.common.QName;
76 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
77 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
78 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
81 * Unit tests for EntityOwnershipShard.
83 * @author Thomas Pantelis
85 public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest {
86 private static final String ENTITY_TYPE = "test type";
87 private static final YangInstanceIdentifier ENTITY_ID1 =
88 YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity1"));
89 private static final YangInstanceIdentifier ENTITY_ID2 =
90 YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity2"));
91 private static final YangInstanceIdentifier ENTITY_ID3 =
92 YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity3"));
93 private static final YangInstanceIdentifier ENTITY_ID4 =
94 YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity4"));
95 private static final YangInstanceIdentifier ENTITY_ID5 =
96 YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "entity5"));
97 private static final SchemaContext SCHEMA_CONTEXT = SchemaContextHelper.entityOwners();
98 private static final AtomicInteger NEXT_SHARD_NUM = new AtomicInteger();
99 private static final String LOCAL_MEMBER_NAME = "member-1";
101 private final Builder dataStoreContextBuilder = DatastoreContext.newBuilder();
102 private final TestActorFactory actorFactory = new TestActorFactory(getSystem());
105 public void tearDown() {
106 actorFactory.close();
110 public void testOnRegisterCandidateLocal() throws Exception {
111 ShardTestKit kit = new ShardTestKit(getSystem());
113 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
115 ShardTestKit.waitUntilLeader(shard);
117 YangInstanceIdentifier entityId = ENTITY_ID1;
118 Entity entity = new Entity(ENTITY_TYPE, entityId);
120 shard.tell(new RegisterCandidateLocal(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);
128 public void testOnRegisterCandidateLocalWithNoInitialLeader() throws Exception {
129 ShardTestKit kit = new ShardTestKit(getSystem());
131 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2);
133 String peerId = newShardId("follower").toString();
134 TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId, false).
135 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
137 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
138 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
139 withDispatcher(Dispatchers.DefaultDispatcherId()));
141 YangInstanceIdentifier entityId = ENTITY_ID1;
142 Entity entity = new Entity(ENTITY_TYPE, entityId);
144 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
145 kit.expectMsgClass(SuccessReply.class);
147 // Now grant the vote so the shard becomes the leader. This should retry the commit.
148 peer.underlyingActor().grantVote = true;
150 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
151 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
155 public void testOnRegisterCandidateLocalWithNoInitialConsensus() throws Exception {
156 ShardTestKit kit = new ShardTestKit(getSystem());
158 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
159 shardTransactionCommitTimeoutInSeconds(1);
161 String peerId = newShardId("follower").toString();
162 TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId).
163 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
165 MockFollower follower = peer.underlyingActor();
167 // Drop AppendEntries so consensus isn't reached.
168 follower.dropAppendEntries = true;
170 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
171 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
172 withDispatcher(Dispatchers.DefaultDispatcherId()));
174 ShardTestKit.waitUntilLeader(shard);
176 YangInstanceIdentifier entityId = ENTITY_ID1;
177 Entity entity = new Entity(ENTITY_TYPE, entityId);
179 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
180 kit.expectMsgClass(SuccessReply.class);
182 // Wait enough time for the commit to timeout.
183 Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
185 // Resume AppendEntries - the follower should ack the commit which should then result in the candidate
186 // write being applied to the state.
187 follower.dropAppendEntries = false;
189 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
190 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
194 public void testOnRegisterCandidateLocalWithIsolatedLeader() throws Exception {
195 ShardTestKit kit = new ShardTestKit(getSystem());
197 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
198 shardIsolatedLeaderCheckIntervalInMillis(50);
200 String peerId = newShardId("follower").toString();
201 TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId).
202 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
204 MockFollower follower = peer.underlyingActor();
206 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(
207 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build()).
208 withDispatcher(Dispatchers.DefaultDispatcherId()));
210 ShardTestKit.waitUntilLeader(shard);
212 // Drop AppendEntries and wait enough time for the shard to switch to IsolatedLeader.
213 follower.dropAppendEntries = true;
214 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
216 YangInstanceIdentifier entityId = ENTITY_ID1;
217 Entity entity = new Entity(ENTITY_TYPE, entityId);
219 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
220 kit.expectMsgClass(SuccessReply.class);
222 // Resume AppendEntries - the candidate write should now be committed.
223 follower.dropAppendEntries = false;
224 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
225 verifyOwner(shard, ENTITY_TYPE, entityId, LOCAL_MEMBER_NAME);
229 public void testOnRegisterCandidateLocalWithRemoteLeader() throws Exception {
230 ShardTestKit kit = new ShardTestKit(getSystem());
232 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2).
233 shardBatchedModificationCount(5);
235 String peerId = newShardId("leader").toString();
236 TestActorRef<MockLeader> peer = actorFactory.createTestActor(Props.create(MockLeader.class).
237 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
239 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(Props.create(
240 TestEntityOwnershipShard.class, newShardId(LOCAL_MEMBER_NAME),
241 ImmutableMap.<String, String>builder().put(peerId, peer.path().toString()).build(),
242 dataStoreContextBuilder.build()).withDispatcher(Dispatchers.DefaultDispatcherId()));
244 shard.tell(new AppendEntries(1L, peerId, -1L, -1L, Collections.<ReplicatedLogEntry>emptyList(), -1L, -1L,
245 DataStoreVersions.CURRENT_VERSION), peer);
247 shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef());
248 kit.expectMsgClass(SuccessReply.class);
250 MockLeader leader = peer.underlyingActor();
251 assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
252 leader.modificationsReceived, 5, TimeUnit.SECONDS));
253 verifyBatchedEntityCandidate(leader.getAndClearReceivedModifications(), ENTITY_TYPE, ENTITY_ID1,
256 // Test with initial commit timeout and subsequent retry.
258 leader.modificationsReceived = new CountDownLatch(1);
259 leader.sendReply = false;
261 shard.tell(dataStoreContextBuilder.shardTransactionCommitTimeoutInSeconds(1).build(), ActorRef.noSender());
263 shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID2)), kit.getRef());
264 kit.expectMsgClass(SuccessReply.class);
266 assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
267 leader.modificationsReceived, 5, TimeUnit.SECONDS));
268 verifyBatchedEntityCandidate(leader.getAndClearReceivedModifications(), ENTITY_TYPE, ENTITY_ID2,
271 // Send a bunch of registration messages quickly and verify.
275 leader.modificationsReceived = new CountDownLatch(max);
276 List<YangInstanceIdentifier> entityIds = new ArrayList<>();
277 for(int i = 1; i <= max; i++) {
278 YangInstanceIdentifier id = YangInstanceIdentifier.of(QName.create("test", "2015-08-14", "test" + i));
280 shard.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, id)), kit.getRef());
283 assertEquals("Leader received BatchedModifications", true, Uninterruptibles.awaitUninterruptibly(
284 leader.modificationsReceived, 10, TimeUnit.SECONDS));
286 // Sleep a little to ensure no additional BatchedModifications are received.
288 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
290 List<Modification> receivedMods = leader.getAndClearReceivedModifications();
291 for(int i = 0; i < max; i++) {
292 verifyBatchedEntityCandidate(receivedMods.get(i), ENTITY_TYPE, entityIds.get(i), LOCAL_MEMBER_NAME);
295 assertEquals("# modifications received", max, receivedMods.size());
299 public void testOnUnregisterCandidateLocal() throws Exception {
300 ShardTestKit kit = new ShardTestKit(getSystem());
301 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
302 ShardTestKit.waitUntilLeader(shard);
304 Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
308 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
309 kit.expectMsgClass(SuccessReply.class);
311 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
312 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
316 shard.tell(new UnregisterCandidateLocal(entity), kit.getRef());
317 kit.expectMsgClass(SuccessReply.class);
319 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, "");
323 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
324 kit.expectMsgClass(SuccessReply.class);
326 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
327 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
331 public void testOwnershipChanges() throws Exception {
332 ShardTestKit kit = new ShardTestKit(getSystem());
333 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
334 ShardTestKit.waitUntilLeader(shard);
336 Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
337 ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
339 // Add a remote candidate
341 String remoteMemberName1 = "remoteMember1";
342 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
346 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
347 kit.expectMsgClass(SuccessReply.class);
349 // Verify the remote candidate becomes owner
351 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
352 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
353 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
355 // Add another remote candidate and verify ownership doesn't change
357 String remoteMemberName2 = "remoteMember2";
358 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
360 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
361 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
362 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
364 // Remove the second remote candidate and verify ownership doesn't change
366 deleteNode(candidatePath(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
368 verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
369 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
370 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
372 // Remove the first remote candidate and verify the local candidate becomes owner
374 deleteNode(candidatePath(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
376 verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
377 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
379 // Add the second remote candidate back and verify ownership doesn't change
381 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName2), shardDataTree);
383 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
384 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
385 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
387 // Unregister the local candidate and verify the second remote candidate becomes owner
389 shard.tell(new UnregisterCandidateLocal(entity), kit.getRef());
390 kit.expectMsgClass(SuccessReply.class);
392 verifyEntityCandidateRemoved(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
393 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName2);
397 public void testOwnerChangesOnPeerAvailabilityChanges() throws Exception {
398 ShardTestKit kit = new ShardTestKit(getSystem());
400 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(500).shardElectionTimeoutFactor(10000);
402 String peerMemberName1 = "peerMember1";
403 String peerMemberName2 = "peerMember2";
405 ShardIdentifier leaderId = newShardId(LOCAL_MEMBER_NAME);
406 ShardIdentifier peerId1 = newShardId(peerMemberName1);
407 ShardIdentifier peerId2 = newShardId(peerMemberName2);
409 TestActorRef<EntityOwnershipShard> peer1 = actorFactory.createTestActor(newShardProps(peerId1,
410 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId2.toString(), "").build(),
411 peerMemberName1, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
413 TestActorRef<EntityOwnershipShard> peer2 = actorFactory.createTestActor(newShardProps(peerId2,
414 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId1.toString(), "").build(),
415 peerMemberName2, EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString());
417 TestActorRef<EntityOwnershipShard> leader = actorFactory.createTestActor(newShardProps(leaderId,
418 ImmutableMap.<String, String>builder().put(peerId1.toString(), peer1.path().toString()).
419 put(peerId2.toString(), peer2.path().toString()).build(), LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build()).
420 withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString());
421 leader.tell(ElectionTimeout.INSTANCE, leader);
423 ShardTestKit.waitUntilLeader(leader);
425 // Send PeerDown and PeerUp with no entities
427 leader.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
428 leader.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
430 // Add candidates for entity1 with the local leader as the owner
432 leader.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID1)), kit.getRef());
433 kit.expectMsgClass(SuccessReply.class);
434 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
436 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, peerMemberName2), kit);
437 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
439 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, peerMemberName1), kit);
440 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName1);
441 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
443 // Add candidates for entity2 with peerMember2 as the owner
445 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName2), kit);
446 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
448 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName1), kit);
449 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
450 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
452 // Add candidates for entity3 with peerMember2 as the owner.
454 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName2), kit);
455 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
457 leader.tell(new RegisterCandidateLocal(new Entity(ENTITY_TYPE, ENTITY_ID3)), kit.getRef());
458 kit.expectMsgClass(SuccessReply.class);
459 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
461 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName1), kit);
462 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName1);
463 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
465 // Add only candidate peerMember2 for entity4.
467 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID4, peerMemberName2), kit);
468 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
469 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
471 // Add only candidate peerMember1 for entity5.
473 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID5, peerMemberName1), kit);
474 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID5, peerMemberName1);
475 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID5, peerMemberName1);
477 // Kill peerMember2 and send PeerDown - the entities (2, 3, 4) owned by peerMember2 should get a new
481 peer2.tell(PoisonPill.getInstance(), ActorRef.noSender());
482 kit.expectMsgClass(JavaTestKit.duration("5 seconds"), Terminated.class);
485 leader.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
486 // Send PeerDown again - should be noop
487 leader.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
488 peer1.tell(new PeerDown(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
490 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, ""); // no other candidates so should clear
491 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
492 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
493 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
495 verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
496 verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
497 verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
498 verifyNoEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID4, peerMemberName2);
500 // Reinstate peerMember2 - no owners should change
502 peer2 = actorFactory.createTestActor(newShardProps(peerId2,
503 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId1.toString(), "").build(),
504 peerMemberName2, EntityOwnerSelectionStrategyConfig.newBuilder().build()). withDispatcher(Dispatchers.DefaultDispatcherId()), peerId2.toString());
505 leader.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
506 // Send PeerUp again - should be noop
507 leader.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
508 peer1.tell(new PeerUp(peerId2.getMemberName(), peerId2.toString()), ActorRef.noSender());
510 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
511 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
512 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
513 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, "");
515 // Add back candidate peerMember2 for entities 1, 2, & 3.
517 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, peerMemberName2), kit);
518 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID2, peerMemberName2), kit);
519 commitModification(leader, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID3, peerMemberName2), kit);
520 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
521 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
522 verifyCommittedEntityCandidate(leader, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
523 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
524 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName1);
525 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
527 // Kill peerMember1 and send PeerDown - entity 2 should get a new owner selected
529 peer1.tell(PoisonPill.getInstance(), ActorRef.noSender());
530 leader.tell(new PeerDown(peerId1.getMemberName(), peerId1.toString()), ActorRef.noSender());
532 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
534 // Verify the reinstated peerMember2 is fully synced.
536 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, "");
537 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
538 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
539 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
541 // Reinstate peerMember1 and verify no owner changes
543 peer1 = actorFactory.createTestActor(newShardProps(peerId1,
544 ImmutableMap.<String, String>builder().put(leaderId.toString(), ""). put(peerId2.toString(), "").build(),
545 peerMemberName1, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), peerId1.toString());
546 leader.tell(new PeerUp(peerId1.getMemberName(), peerId1.toString()), ActorRef.noSender());
548 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID4, "");
549 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
550 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
551 verifyOwner(leader, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
553 // Verify the reinstated peerMember1 is fully synced.
555 verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID4, "");
556 verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID3, LOCAL_MEMBER_NAME);
557 verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
558 verifyOwner(peer1, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
560 // Kill the local leader and elect peer2 the leader. This should cause a new owner to be selected for
561 // the entities (1 and 3) previously owned by the local leader member.
563 peer2.tell(new PeerAddressResolved(peerId1.toString(), peer1.path().toString()), ActorRef.noSender());
564 peer2.tell(new PeerUp(leaderId.getMemberName(), leaderId.toString()), ActorRef.noSender());
565 peer2.tell(new PeerUp(peerId1.getMemberName(), peerId1.toString()), ActorRef.noSender());
567 leader.tell(PoisonPill.getInstance(), ActorRef.noSender());
568 peer2.tell(new PeerDown(leaderId.getMemberName(), leaderId.toString()), ActorRef.noSender());
569 peer2.tell(ElectionTimeout.INSTANCE, peer2);
571 ShardTestKit.waitUntilLeader(peer2);
573 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID4, "");
574 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID3, peerMemberName2);
575 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID2, peerMemberName2);
576 verifyOwner(peer2, ENTITY_TYPE, ENTITY_ID1, peerMemberName2);
580 public void testLocalCandidateRemovedWithCandidateRegistered() throws Exception {
581 ShardTestKit kit = new ShardTestKit(getSystem());
583 dataStoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(10000);
584 ShardIdentifier leaderId = newShardId("leader");
585 ShardIdentifier localId = newShardId(LOCAL_MEMBER_NAME);
587 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(Props.create(
588 TestEntityOwnershipShard.class, localId,
589 ImmutableMap.<String, String>builder().put(leaderId.toString(), "".toString()).build(),
590 dataStoreContextBuilder.build()).withDispatcher(Dispatchers.DefaultDispatcherId()));
592 TestActorRef<EntityOwnershipShard> leader = actorFactory.createTestActor(newShardProps(leaderId,
593 ImmutableMap.<String, String>builder().put(localId.toString(), shard.path().toString()).build(),
594 LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build()).withDispatcher(Dispatchers.DefaultDispatcherId()), leaderId.toString());
595 leader.tell(ElectionTimeout.INSTANCE, leader);
597 ShardTestKit.waitUntilLeader(leader);
599 shard.tell(new PeerAddressResolved(leaderId.toString(), leader.path().toString()), ActorRef.noSender());
601 Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
602 EntityOwnershipListener listener = mock(EntityOwnershipListener.class);
604 shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
605 kit.expectMsgClass(SuccessReply.class);
607 // Register local candidate
609 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
610 kit.expectMsgClass(SuccessReply.class);
611 verifyCommittedEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME);
612 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true));
615 // Simulate a replicated commit from the leader to remove the local candidate that would occur after a
616 // network partition is healed.
618 leader.tell(new PeerDown(localId.getMemberName(), localId.toString()), ActorRef.noSender());
620 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, true, false, false));
622 // Since the the shard has a local candidate registered, it should re-add its candidate to the entity.
624 verifyCommittedEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME);
625 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity, false, true, true));
627 // Unregister the local candidate and verify it's removed and no re-added.
629 shard.tell(new UnregisterCandidateLocal(entity), kit.getRef());
630 kit.expectMsgClass(SuccessReply.class);
632 verifyNoEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME);
633 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
634 verifyNoEntityCandidate(shard, entity.getType(), entity.getId(), LOCAL_MEMBER_NAME);
638 public void testListenerRegistration() throws Exception {
639 ShardTestKit kit = new ShardTestKit(getSystem());
640 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps());
641 ShardTestKit.waitUntilLeader(shard);
642 ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
644 String otherEntityType = "otherEntityType";
645 Entity entity1 = new Entity(ENTITY_TYPE, ENTITY_ID1);
646 Entity entity2 = new Entity(ENTITY_TYPE, ENTITY_ID2);
647 Entity entity3 = new Entity(ENTITY_TYPE, ENTITY_ID3);
648 Entity entity4 = new Entity(otherEntityType, ENTITY_ID3);
649 EntityOwnershipListener listener = mock(EntityOwnershipListener.class);
653 shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
654 kit.expectMsgClass(SuccessReply.class);
656 // Register a couple candidates for the desired entity type and verify listener is notified.
658 shard.tell(new RegisterCandidateLocal(entity1), kit.getRef());
659 kit.expectMsgClass(SuccessReply.class);
661 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, false, true, true));
663 shard.tell(new RegisterCandidateLocal(entity2), kit.getRef());
664 kit.expectMsgClass(SuccessReply.class);
666 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity2, false, true, true));
669 // Register another candidate for another entity type and verify listener is not notified.
671 shard.tell(new RegisterCandidateLocal(entity4), kit.getRef());
672 kit.expectMsgClass(SuccessReply.class);
674 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
675 verify(listener, never()).ownershipChanged(ownershipChange(entity4));
677 // Register remote candidate for entity1
679 String remoteMemberName = "remoteMember";
680 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, entity1.getId(), remoteMemberName),
682 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, entity1.getId(), remoteMemberName);
684 // Unregister the local candidate for entity1 and verify listener is notified
686 shard.tell(new UnregisterCandidateLocal(entity1), kit.getRef());
687 kit.expectMsgClass(SuccessReply.class);
689 verify(listener, timeout(5000)).ownershipChanged(ownershipChange(entity1, true, false, true));
692 // Unregister the listener, add a candidate for entity3 and verify listener isn't notified
694 shard.tell(new UnregisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
695 kit.expectMsgClass(SuccessReply.class);
697 shard.tell(new RegisterCandidateLocal(entity3), kit.getRef());
698 kit.expectMsgClass(SuccessReply.class);
700 verifyOwner(shard, ENTITY_TYPE, entity3.getId(), LOCAL_MEMBER_NAME);
701 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
702 verify(listener, never()).ownershipChanged(any(EntityOwnershipChange.class));
704 // Re-register the listener and verify it gets notified of currently owned entities
708 shard.tell(new RegisterListenerLocal(listener, ENTITY_TYPE), kit.getRef());
709 kit.expectMsgClass(SuccessReply.class);
711 verify(listener, timeout(5000).times(2)).ownershipChanged(or(ownershipChange(entity2, false, true, true),
712 ownershipChange(entity3, false, true, true)));
713 Uninterruptibles.sleepUninterruptibly(300, TimeUnit.MILLISECONDS);
714 verify(listener, never()).ownershipChanged(ownershipChange(entity4));
715 verify(listener, times(1)).ownershipChanged(ownershipChange(entity1));
718 private static void commitModification(TestActorRef<EntityOwnershipShard> shard, NormalizedNode<?, ?> node,
719 JavaTestKit sender) {
720 BatchedModifications modifications = newBatchedModifications();
721 modifications.addModification(new MergeModification(ENTITY_OWNERS_PATH, node));
723 shard.tell(modifications, sender.getRef());
724 sender.expectMsgClass(CommitTransactionReply.class);
727 private static BatchedModifications newBatchedModifications() {
728 BatchedModifications modifications = new BatchedModifications("tnx", DataStoreVersions.CURRENT_VERSION, "");
729 modifications.setDoCommitOnReady(true);
730 modifications.setReady(true);
731 modifications.setTotalMessagesSent(1);
732 return modifications;
735 private void verifyEntityCandidateRemoved(final TestActorRef<EntityOwnershipShard> shard, String entityType,
736 YangInstanceIdentifier entityId, String candidateName) {
737 verifyNodeRemoved(candidatePath(entityType, entityId, candidateName),
738 new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
740 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
742 return AbstractShardTest.readStore(shard, path);
743 } catch(Exception e) {
744 throw new AssertionError("Failed to read " + path, e);
750 private void verifyCommittedEntityCandidate(final TestActorRef<EntityOwnershipShard> shard, String entityType,
751 YangInstanceIdentifier entityId, String candidateName) {
752 verifyEntityCandidate(entityType, entityId, candidateName, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
754 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
756 return AbstractShardTest.readStore(shard, path);
757 } catch(Exception e) {
758 throw new AssertionError("Failed to read " + path, e);
764 private void verifyNoEntityCandidate(final TestActorRef<EntityOwnershipShard> shard, String entityType,
765 YangInstanceIdentifier entityId, String candidateName) {
766 verifyEntityCandidate(entityType, entityId, candidateName, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
768 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
770 return AbstractShardTest.readStore(shard, path);
771 } catch(Exception e) {
772 throw new AssertionError("Failed to read " + path, e);
778 private void verifyBatchedEntityCandidate(List<Modification> mods, String entityType,
779 YangInstanceIdentifier entityId, String candidateName) throws Exception {
780 assertEquals("BatchedModifications size", 1, mods.size());
781 verifyBatchedEntityCandidate(mods.get(0), entityType, entityId, candidateName);
784 private void verifyBatchedEntityCandidate(Modification mod, String entityType,
785 YangInstanceIdentifier entityId, String candidateName) throws Exception {
786 assertEquals("Modification type", MergeModification.class, mod.getClass());
787 verifyEntityCandidate(((MergeModification)mod).getData(), entityType,
788 entityId, candidateName, true);
791 private static void verifyOwner(final TestActorRef<EntityOwnershipShard> shard, String entityType,
792 YangInstanceIdentifier entityId, String localMemberName) {
793 verifyOwner(localMemberName, entityType, entityId, new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
795 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
797 return AbstractShardTest.readStore(shard, path);
798 } catch(Exception e) {
805 private Props newShardProps() {
806 return newShardProps(Collections.<String,String>emptyMap());
809 private Props newShardProps(EntityOwnerSelectionStrategyConfig strategyConfig, Map<String, String> peers) {
810 return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, strategyConfig);
813 private Props newShardProps(Map<String, String> peers) {
814 return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build());
817 private Props newShardProps(ShardIdentifier shardId, Map<String,String> peers, String memberName,
818 EntityOwnerSelectionStrategyConfig config) {
819 return EntityOwnershipShard.newBuilder().id(shardId).peerAddresses(peers).
820 datastoreContext(dataStoreContextBuilder.build()).schemaContext(SCHEMA_CONTEXT).
821 localMemberName(MemberName.forName(memberName)).ownerSelectionStrategyConfig(config).props()
822 .withDispatcher(Dispatchers.DefaultDispatcherId());
825 private static ShardIdentifier newShardId(String memberName) {
826 return ShardIdentifier.create("entity-ownership", MemberName.forName(memberName),
827 "operational" + NEXT_SHARD_NUM.getAndIncrement());
830 private static class TestEntityOwnershipShard extends EntityOwnershipShard {
832 TestEntityOwnershipShard(ShardIdentifier name, Map<String, String> peerAddresses,
833 DatastoreContext datastoreContext) {
834 super(newBuilder().id(name).peerAddresses(peerAddresses).datastoreContext(datastoreContext).
835 schemaContext(SCHEMA_CONTEXT).localMemberName(MemberName.forName(LOCAL_MEMBER_NAME)));
839 public void handleCommand(Object message) {
840 if(!(message instanceof ElectionTimeout)) {
841 super.handleCommand(message);
846 private static class MockFollower extends UntypedActor {
847 volatile boolean grantVote;
848 volatile boolean dropAppendEntries;
849 private final String myId;
851 public MockFollower(String myId) {
855 public MockFollower(String myId, boolean grantVote) {
857 this.grantVote = grantVote;
861 public void onReceive(Object message) {
862 if(message instanceof RequestVote) {
864 getSender().tell(new RequestVoteReply(((RequestVote)message).getTerm(), true), getSelf());
866 } else if(message instanceof AppendEntries) {
867 if(!dropAppendEntries) {
868 AppendEntries req = (AppendEntries) message;
869 long lastIndex = req.getLeaderCommit();
870 if (req.getEntries().size() > 0) {
871 for(ReplicatedLogEntry entry : req.getEntries()) {
872 lastIndex = entry.getIndex();
876 getSender().tell(new AppendEntriesReply(myId, req.getTerm(), true, lastIndex, req.getTerm(),
877 DataStoreVersions.CURRENT_VERSION), getSelf());
885 public void testDelayedEntityOwnerSelectionWhenMaxPeerRequestsReceived() throws Exception {
886 ShardTestKit kit = new ShardTestKit(getSystem());
887 EntityOwnerSelectionStrategyConfig.Builder builder
888 = EntityOwnerSelectionStrategyConfig.newBuilder().addStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500);
890 String peerId = newShardId("follower").toString();
891 TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId, false).
892 withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
894 peer.underlyingActor().grantVote = true;
896 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(builder.build(),
897 ImmutableMap.of(peerId.toString(), peer.path().toString())));
898 ShardTestKit.waitUntilLeader(shard);
900 Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
901 ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
903 // Add a remote candidate
905 String remoteMemberName1 = "follower";
906 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
910 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
911 kit.expectMsgClass(SuccessReply.class);
913 // Verify the local candidate becomes owner
915 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
916 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
917 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
921 public void testDelayedEntityOwnerSelection() throws Exception {
922 ShardTestKit kit = new ShardTestKit(getSystem());
923 EntityOwnerSelectionStrategyConfig.Builder builder
924 = EntityOwnerSelectionStrategyConfig.newBuilder().addStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500);
926 String follower1Id = newShardId("follower1").toString();
927 TestActorRef<MockFollower> follower1 = actorFactory.createTestActor(Props.create(MockFollower.class, follower1Id, false).
928 withDispatcher(Dispatchers.DefaultDispatcherId()), follower1Id);
930 follower1.underlyingActor().grantVote = true;
932 String follower2Id = newShardId("follower").toString();
933 TestActorRef<MockFollower> follower2 = actorFactory.createTestActor(Props.create(MockFollower.class, follower2Id, false).
934 withDispatcher(Dispatchers.DefaultDispatcherId()), follower2Id);
936 follower2.underlyingActor().grantVote = true;
939 TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(builder.build(),
940 ImmutableMap.of(follower1Id.toString(), follower2.path().toString(), follower2Id.toString(), follower2.path().toString())));
941 ShardTestKit.waitUntilLeader(shard);
943 Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
944 ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
946 // Add a remote candidate
948 String remoteMemberName1 = "follower";
949 writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
953 shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
954 kit.expectMsgClass(SuccessReply.class);
956 // Verify the local candidate becomes owner
958 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
959 verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
960 verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
963 private static class MockLeader extends UntypedActor {
964 volatile CountDownLatch modificationsReceived = new CountDownLatch(1);
965 List<Modification> receivedModifications = new ArrayList<>();
966 volatile boolean sendReply = true;
970 public void onReceive(Object message) {
971 if(message instanceof BatchedModifications) {
973 Uninterruptibles.sleepUninterruptibly(delay, TimeUnit.MILLISECONDS);
977 BatchedModifications mods = (BatchedModifications) message;
978 synchronized (receivedModifications) {
979 for(int i = 0; i < mods.getModifications().size(); i++) {
980 receivedModifications.add(mods.getModifications().get(i));
981 modificationsReceived.countDown();
985 getSender().tell(CommitTransactionReply.instance(DataStoreVersions.CURRENT_VERSION).
986 toSerializable(), getSelf());
993 List<Modification> getAndClearReceivedModifications() {
994 synchronized (receivedModifications) {
995 List<Modification> ret = new ArrayList<>(receivedModifications);
996 receivedModifications.clear();