Fix intermittent failure in testRemoveShardReplica
[controller.git] / opendaylight / md-sal / sal-cluster-admin / src / test / java / org / opendaylight / controller / cluster / datastore / admin / ClusterAdminRpcServiceTest.java
1 /*
2  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
3  *
4  * This program and the accompanying materials are made available under the
5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6  * and is available at http://www.eclipse.org/legal/epl-v10.html
7  */
8 package org.opendaylight.controller.cluster.datastore.admin;
9
10 import static org.hamcrest.CoreMatchers.anyOf;
11 import static org.hamcrest.CoreMatchers.containsString;
12 import static org.junit.Assert.assertEquals;
13 import static org.junit.Assert.assertNotNull;
14 import static org.junit.Assert.assertNull;
15 import static org.junit.Assert.assertThat;
16 import static org.junit.Assert.assertTrue;
17 import static org.junit.Assert.fail;
18 import static org.opendaylight.controller.cluster.datastore.MemberNode.verifyNoShardPresent;
19 import static org.opendaylight.controller.cluster.datastore.MemberNode.verifyRaftPeersPresent;
20 import static org.opendaylight.controller.cluster.datastore.MemberNode.verifyRaftState;
21 import akka.actor.ActorRef;
22 import akka.actor.PoisonPill;
23 import akka.actor.Status.Success;
24 import akka.cluster.Cluster;
25 import com.google.common.base.Optional;
26 import com.google.common.collect.ImmutableList;
27 import com.google.common.collect.ImmutableMap;
28 import com.google.common.collect.Iterables;
29 import com.google.common.collect.Lists;
30 import com.google.common.collect.Sets;
31 import java.io.File;
32 import java.io.FileInputStream;
33 import java.net.URI;
34 import java.util.AbstractMap.SimpleEntry;
35 import java.util.ArrayList;
36 import java.util.Arrays;
37 import java.util.HashMap;
38 import java.util.HashSet;
39 import java.util.List;
40 import java.util.Map;
41 import java.util.Map.Entry;
42 import java.util.Set;
43 import java.util.concurrent.TimeUnit;
44 import org.apache.commons.lang3.SerializationUtils;
45 import org.junit.After;
46 import org.junit.Before;
47 import org.junit.Test;
48 import org.opendaylight.controller.cluster.access.concepts.MemberName;
49 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
50 import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
51 import org.opendaylight.controller.cluster.datastore.MemberNode;
52 import org.opendaylight.controller.cluster.datastore.MemberNode.RaftStateVerifier;
53 import org.opendaylight.controller.cluster.datastore.Shard;
54 import org.opendaylight.controller.cluster.datastore.config.ModuleShardConfiguration;
55 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
56 import org.opendaylight.controller.cluster.datastore.messages.CreateShard;
57 import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshot;
58 import org.opendaylight.controller.cluster.raft.RaftState;
59 import org.opendaylight.controller.cluster.raft.ReplicatedLogImplEntry;
60 import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload;
61 import org.opendaylight.controller.cluster.raft.persisted.ServerInfo;
62 import org.opendaylight.controller.cluster.raft.base.messages.UpdateElectionTerm;
63 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
64 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
65 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
66 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
67 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
68 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutput;
69 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaInputBuilder;
70 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreInputBuilder;
71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsInputBuilder;
72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutput;
73 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardInputBuilder;
74 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.DataStoreType;
75 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutput;
76 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasInputBuilder;
77 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutput;
78 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaInputBuilder;
79 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.member.voting.states.input.MemberVotingStateBuilder;
80 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResult;
81 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultBuilder;
82 import org.opendaylight.yangtools.yang.common.RpcError;
83 import org.opendaylight.yangtools.yang.common.RpcResult;
84 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
85
86 /**
87  * Unit tests for ClusterAdminRpcService.
88  *
89  * @author Thomas Pantelis
90  */
91 public class ClusterAdminRpcServiceTest {
92     private static final MemberName MEMBER_1 = MemberName.forName("member-1");
93     private static final MemberName MEMBER_2 = MemberName.forName("member-2");
94     private static final MemberName MEMBER_3 = MemberName.forName("member-3");
95     private final List<MemberNode> memberNodes = new ArrayList<>();
96
97     @Before
98     public void setUp() {
99         InMemoryJournal.clear();
100         InMemorySnapshotStore.clear();
101     }
102
103     @After
104     public void tearDown() {
105         for (MemberNode m : Lists.reverse(memberNodes)) {
106             m.cleanup();
107         }
108         memberNodes.clear();
109     }
110
111     @Test
112     public void testBackupDatastore() throws Exception {
113         MemberNode node = MemberNode.builder(memberNodes).akkaConfig("Member1").
114                 moduleShardsConfig("module-shards-member1.conf").
115                 waitForShardLeader("cars", "people").testName("testBackupDatastore").build();
116
117         String fileName = "target/testBackupDatastore";
118         new File(fileName).delete();
119
120         ClusterAdminRpcService service = new ClusterAdminRpcService(node.configDataStore(), node.operDataStore());
121
122         RpcResult<Void> rpcResult = service .backupDatastore(new BackupDatastoreInputBuilder().
123                 setFilePath(fileName).build()).get(5, TimeUnit.SECONDS);
124         verifySuccessfulRpcResult(rpcResult);
125
126         try(FileInputStream fis = new FileInputStream(fileName)) {
127             List<DatastoreSnapshot> snapshots = SerializationUtils.deserialize(fis);
128             assertEquals("DatastoreSnapshot size", 2, snapshots.size());
129
130             ImmutableMap<String, DatastoreSnapshot> map = ImmutableMap.of(snapshots.get(0).getType(), snapshots.get(0),
131                     snapshots.get(1).getType(), snapshots.get(1));
132             verifyDatastoreSnapshot(node.configDataStore().getActorContext().getDataStoreName(),
133                     map.get(node.configDataStore().getActorContext().getDataStoreName()), "cars", "people");
134         } finally {
135             new File(fileName).delete();
136         }
137
138         // Test failure by killing a shard.
139
140         node.configDataStore().getActorContext().getShardManager().tell(node.datastoreContextBuilder().
141                 shardInitializationTimeout(200, TimeUnit.MILLISECONDS).build(), ActorRef.noSender());
142
143         ActorRef carsShardActor = node.configDataStore().getActorContext().findLocalShard("cars").get();
144         node.kit().watch(carsShardActor);
145         carsShardActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
146         node.kit().expectTerminated(carsShardActor);
147
148         rpcResult = service.backupDatastore(new BackupDatastoreInputBuilder().setFilePath(fileName).build()).
149                 get(5, TimeUnit.SECONDS);
150         assertEquals("isSuccessful", false, rpcResult.isSuccessful());
151         assertEquals("getErrors", 1, rpcResult.getErrors().size());
152     }
153
154     private static void verifyDatastoreSnapshot(String type, DatastoreSnapshot datastoreSnapshot, String... expShardNames) {
155         assertNotNull("Missing DatastoreSnapshot for type " + type, datastoreSnapshot);
156         Set<String> shardNames = new HashSet<>();
157         for(DatastoreSnapshot.ShardSnapshot s: datastoreSnapshot.getShardSnapshots()) {
158             shardNames.add(s.getName());
159         }
160
161         assertEquals("DatastoreSnapshot shard names", Sets.newHashSet(expShardNames), shardNames);
162     }
163
164     @Test
165     public void testAddShardReplica() throws Exception {
166         String name = "testAddShardReplica";
167         String moduleShardsConfig = "module-shards-cars-member-1.conf";
168         MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ).
169                 moduleShardsConfig(moduleShardsConfig).waitForShardLeader("cars").build();
170
171         MemberNode newReplicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name).
172                 moduleShardsConfig(moduleShardsConfig).build();
173
174         leaderNode1.waitForMembersUp("member-2");
175
176         doAddShardReplica(newReplicaNode2, "cars", "member-1");
177
178         MemberNode newReplicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name).
179                 moduleShardsConfig(moduleShardsConfig).build();
180
181         leaderNode1.waitForMembersUp("member-3");
182         newReplicaNode2.waitForMembersUp("member-3");
183
184         doAddShardReplica(newReplicaNode3, "cars", "member-1", "member-2");
185
186         verifyRaftPeersPresent(newReplicaNode2.configDataStore(), "cars", "member-1", "member-3");
187         verifyRaftPeersPresent(newReplicaNode2.operDataStore(), "cars", "member-1", "member-3");
188
189         // Write data to member-2's config datastore and read/verify via member-3
190         NormalizedNode<?, ?> configCarsNode = writeCarsNodeAndVerify(newReplicaNode2.configDataStore(),
191                 newReplicaNode3.configDataStore());
192
193         // Write data to member-3's oper datastore and read/verify via member-2
194         writeCarsNodeAndVerify(newReplicaNode3.operDataStore(), newReplicaNode2.operDataStore());
195
196         // Verify all data has been replicated. We expect 3 log entries and thus last applied index of 2 -
197         // 2 ServerConfigurationPayload entries and the transaction payload entry.
198
199         RaftStateVerifier verifier = raftState -> {
200             assertEquals("Commit index", 2, raftState.getCommitIndex());
201             assertEquals("Last applied index", 2, raftState.getLastApplied());
202         };
203
204         verifyRaftState(leaderNode1.configDataStore(), "cars", verifier);
205         verifyRaftState(leaderNode1.operDataStore(), "cars", verifier);
206
207         verifyRaftState(newReplicaNode2.configDataStore(), "cars", verifier);
208         verifyRaftState(newReplicaNode2.operDataStore(), "cars", verifier);
209
210         verifyRaftState(newReplicaNode3.configDataStore(), "cars", verifier);
211         verifyRaftState(newReplicaNode3.operDataStore(), "cars", verifier);
212
213         // Restart member-3 and verify the cars config shard is re-instated.
214
215         Cluster.get(leaderNode1.kit().getSystem()).down(Cluster.get(newReplicaNode3.kit().getSystem()).selfAddress());
216         newReplicaNode3.cleanup();
217
218         newReplicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name).
219                 moduleShardsConfig(moduleShardsConfig).createOperDatastore(false).build();
220
221         verifyRaftState(newReplicaNode3.configDataStore(), "cars", verifier);
222         readCarsNodeAndVerify(newReplicaNode3.configDataStore(), configCarsNode);
223     }
224
225     @Test
226     public void testAddShardReplicaFailures() throws Exception {
227         String name = "testAddShardReplicaFailures";
228         MemberNode memberNode = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name).
229                 moduleShardsConfig("module-shards-cars-member-1.conf").build();
230
231         ClusterAdminRpcService service = new ClusterAdminRpcService(memberNode.configDataStore(),
232                 memberNode.operDataStore());
233
234         RpcResult<Void> rpcResult = service.addShardReplica(new AddShardReplicaInputBuilder().
235                 setDataStoreType(DataStoreType.Config).build()).get(10, TimeUnit.SECONDS);
236         verifyFailedRpcResult(rpcResult);
237
238         rpcResult = service.addShardReplica(new AddShardReplicaInputBuilder().setShardName("cars").
239                 build()).get(10, TimeUnit.SECONDS);
240         verifyFailedRpcResult(rpcResult);
241
242         rpcResult = service.addShardReplica(new AddShardReplicaInputBuilder().setShardName("people").
243                 setDataStoreType(DataStoreType.Config).build()).get(10, TimeUnit.SECONDS);
244         verifyFailedRpcResult(rpcResult);
245     }
246
247     private static NormalizedNode<?, ?> writeCarsNodeAndVerify(DistributedDataStore writeToStore,
248             DistributedDataStore readFromStore) throws Exception {
249         DOMStoreWriteTransaction writeTx = writeToStore.newWriteOnlyTransaction();
250         NormalizedNode<?, ?> carsNode = CarsModel.create();
251         writeTx.write(CarsModel.BASE_PATH, carsNode);
252
253         DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
254         Boolean canCommit = cohort .canCommit().get(7, TimeUnit.SECONDS);
255         assertEquals("canCommit", true, canCommit);
256         cohort.preCommit().get(5, TimeUnit.SECONDS);
257         cohort.commit().get(5, TimeUnit.SECONDS);
258
259         readCarsNodeAndVerify(readFromStore, carsNode);
260         return carsNode;
261     }
262
263     private static void readCarsNodeAndVerify(DistributedDataStore readFromStore,
264             NormalizedNode<?, ?> expCarsNode) throws Exception {
265         Optional<NormalizedNode<?, ?>> optional = readFromStore.newReadOnlyTransaction().
266                 read(CarsModel.BASE_PATH).get(15, TimeUnit.SECONDS);
267         assertEquals("isPresent", true, optional.isPresent());
268         assertEquals("Data node", expCarsNode, optional.get());
269     }
270
271     private static void doAddShardReplica(MemberNode memberNode, String shardName, String... peerMemberNames)
272             throws Exception {
273         memberNode.waitForMembersUp(peerMemberNames);
274
275         ClusterAdminRpcService service = new ClusterAdminRpcService(memberNode.configDataStore(),
276                 memberNode.operDataStore());
277
278         RpcResult<Void> rpcResult = service.addShardReplica(new AddShardReplicaInputBuilder().setShardName(shardName).
279                 setDataStoreType(DataStoreType.Config).build()).get(10, TimeUnit.SECONDS);
280         verifySuccessfulRpcResult(rpcResult);
281
282         verifyRaftPeersPresent(memberNode.configDataStore(), shardName, peerMemberNames);
283
284         Optional<ActorRef> optional = memberNode.operDataStore().getActorContext().findLocalShard(shardName);
285         assertEquals("Oper shard present", false, optional.isPresent());
286
287         rpcResult = service.addShardReplica(new AddShardReplicaInputBuilder().setShardName(shardName).
288                 setDataStoreType(DataStoreType.Operational).build()).get(10, TimeUnit.SECONDS);
289         verifySuccessfulRpcResult(rpcResult);
290
291         verifyRaftPeersPresent(memberNode.operDataStore(), shardName, peerMemberNames);
292     }
293
294     private static <T> T verifySuccessfulRpcResult(RpcResult<T> rpcResult) {
295         if(!rpcResult.isSuccessful()) {
296             if(rpcResult.getErrors().size() > 0) {
297                 RpcError error = Iterables.getFirst(rpcResult.getErrors(), null);
298                 throw new AssertionError("Rpc failed with error: " + error, error.getCause());
299             }
300
301             fail("Rpc failed with no error");
302         }
303
304         return rpcResult.getResult();
305     }
306
307     private static void verifyFailedRpcResult(RpcResult<Void> rpcResult) {
308         assertEquals("RpcResult", false, rpcResult.isSuccessful());
309         assertEquals("RpcResult errors size", 1, rpcResult.getErrors().size());
310         RpcError error = Iterables.getFirst(rpcResult.getErrors(), null);
311         assertNotNull("RpcResult error message null", error.getMessage());
312     }
313
314     @Test
315     public void testRemoveShardReplica() throws Exception {
316         String name = "testRemoveShardReplica";
317         String moduleShardsConfig = "module-shards-member1-and-2-and-3.conf";
318         MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ).
319                 moduleShardsConfig(moduleShardsConfig).
320                 datastoreContextBuilder(DatastoreContext.newBuilder().
321                         shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)).build();
322
323         MemberNode replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name).
324                 moduleShardsConfig(moduleShardsConfig).build();
325
326         MemberNode replicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name).
327                 moduleShardsConfig(moduleShardsConfig).build();
328
329         leaderNode1.configDataStore().waitTillReady();
330         replicaNode3.configDataStore().waitTillReady();
331         verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars", "member-2", "member-3");
332         verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1", "member-3");
333         verifyRaftPeersPresent(replicaNode3.configDataStore(), "cars", "member-1", "member-2");
334
335         // Invoke RPC service on member-3 to remove it's local shard
336
337         ClusterAdminRpcService service3 = new ClusterAdminRpcService(replicaNode3.configDataStore(),
338                 replicaNode3.operDataStore());
339
340         RpcResult<Void> rpcResult = service3.removeShardReplica(new RemoveShardReplicaInputBuilder().
341                 setShardName("cars").setMemberName("member-3").setDataStoreType(DataStoreType.Config).build()).
342                         get(10, TimeUnit.SECONDS);
343         verifySuccessfulRpcResult(rpcResult);
344
345         verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars", "member-2");
346         verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1");
347         verifyNoShardPresent(replicaNode3.configDataStore(), "cars");
348
349         // Restart member-2 and verify member-3 isn't present.
350
351         Cluster.get(leaderNode1.kit().getSystem()).down(Cluster.get(replicaNode2.kit().getSystem()).selfAddress());
352         replicaNode2.cleanup();
353
354         replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name).
355                 moduleShardsConfig(moduleShardsConfig).build();
356
357         replicaNode2.configDataStore().waitTillReady();
358         verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1");
359
360         // Invoke RPC service on member-1 to remove member-2
361
362         ClusterAdminRpcService service1 = new ClusterAdminRpcService(leaderNode1.configDataStore(),
363                 leaderNode1.operDataStore());
364
365         rpcResult = service1.removeShardReplica(new RemoveShardReplicaInputBuilder().
366                 setShardName("cars").setMemberName("member-2").setDataStoreType(DataStoreType.Config).build()).
367                         get(10, TimeUnit.SECONDS);
368         verifySuccessfulRpcResult(rpcResult);
369
370         verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars");
371         verifyNoShardPresent(replicaNode2.configDataStore(), "cars");
372     }
373
374     @Test
375     public void testRemoveShardLeaderReplica() throws Exception {
376         String name = "testRemoveShardLeaderReplica";
377         String moduleShardsConfig = "module-shards-member1-and-2-and-3.conf";
378         MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ).
379                 moduleShardsConfig(moduleShardsConfig).
380                 datastoreContextBuilder(DatastoreContext.newBuilder().
381                         shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)).build();
382
383         MemberNode replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name).
384                 moduleShardsConfig(moduleShardsConfig).build();
385
386         MemberNode replicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name).
387                 moduleShardsConfig(moduleShardsConfig).build();
388
389         leaderNode1.configDataStore().waitTillReady();
390         verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars", "member-2", "member-3");
391         verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1", "member-3");
392         verifyRaftPeersPresent(replicaNode3.configDataStore(), "cars", "member-1", "member-2");
393
394         replicaNode2.waitForMembersUp("member-1", "member-3");
395         replicaNode3.waitForMembersUp("member-1", "member-2");
396
397         // Invoke RPC service on leader member-1 to remove it's local shard
398
399         ClusterAdminRpcService service1 = new ClusterAdminRpcService(leaderNode1.configDataStore(),
400                 leaderNode1.operDataStore());
401
402         RpcResult<Void> rpcResult = service1.removeShardReplica(new RemoveShardReplicaInputBuilder().
403                 setShardName("cars").setMemberName("member-1").setDataStoreType(DataStoreType.Config).build()).
404                         get(10, TimeUnit.SECONDS);
405         verifySuccessfulRpcResult(rpcResult);
406
407         verifyRaftState(replicaNode2.configDataStore(), "cars", raftState ->
408                 assertThat("Leader Id", raftState.getLeader(), anyOf(containsString("member-2"), containsString("member-3"))));
409
410         verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-3");
411         verifyRaftPeersPresent(replicaNode3.configDataStore(), "cars", "member-2");
412         verifyNoShardPresent(leaderNode1.configDataStore(), "cars");
413     }
414
415     @Test
416     public void testAddReplicasForAllShards() throws Exception {
417         String name = "testAddReplicasForAllShards";
418         String moduleShardsConfig = "module-shards-member1.conf";
419         MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ).
420                 moduleShardsConfig(moduleShardsConfig).waitForShardLeader("cars", "people").build();
421
422         ModuleShardConfiguration petsModuleConfig = new ModuleShardConfiguration(URI.create("pets-ns"), "pets-module",
423                 "pets", null, Arrays.asList(MEMBER_1));
424         leaderNode1.configDataStore().getActorContext().getShardManager().tell(
425                 new CreateShard(petsModuleConfig, Shard.builder(), null), leaderNode1.kit().getRef());
426         leaderNode1.kit().expectMsgClass(Success.class);
427         leaderNode1.kit().waitUntilLeader(leaderNode1.configDataStore().getActorContext(), "pets");
428
429         MemberNode newReplicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name).
430                 moduleShardsConfig(moduleShardsConfig).build();
431
432         leaderNode1.waitForMembersUp("member-2");
433         newReplicaNode2.waitForMembersUp("member-1");
434
435         newReplicaNode2.configDataStore().getActorContext().getShardManager().tell(
436                 new CreateShard(petsModuleConfig, Shard.builder(), null), newReplicaNode2.kit().getRef());
437         newReplicaNode2.kit().expectMsgClass(Success.class);
438
439         newReplicaNode2.operDataStore().getActorContext().getShardManager().tell(
440                 new CreateShard(new ModuleShardConfiguration(URI.create("no-leader-ns"), "no-leader-module",
441                         "no-leader", null, Arrays.asList(MEMBER_1)), Shard.builder(), null),
442                                 newReplicaNode2.kit().getRef());
443         newReplicaNode2.kit().expectMsgClass(Success.class);
444
445         ClusterAdminRpcService service = new ClusterAdminRpcService(newReplicaNode2.configDataStore(),
446                 newReplicaNode2.operDataStore());
447
448         RpcResult<AddReplicasForAllShardsOutput> rpcResult = service.addReplicasForAllShards().get(10, TimeUnit.SECONDS);
449         AddReplicasForAllShardsOutput result = verifySuccessfulRpcResult(rpcResult);
450         verifyShardResults(result.getShardResult(), successShardResult("cars", DataStoreType.Config),
451                 successShardResult("people", DataStoreType.Config),
452                 successShardResult("pets", DataStoreType.Config),
453                 successShardResult("cars", DataStoreType.Operational),
454                 successShardResult("people", DataStoreType.Operational),
455                 failedShardResult("no-leader", DataStoreType.Operational));
456
457         verifyRaftPeersPresent(newReplicaNode2.configDataStore(), "cars", "member-1");
458         verifyRaftPeersPresent(newReplicaNode2.configDataStore(), "people", "member-1");
459         verifyRaftPeersPresent(newReplicaNode2.configDataStore(), "pets", "member-1");
460         verifyRaftPeersPresent(newReplicaNode2.operDataStore(), "cars", "member-1");
461         verifyRaftPeersPresent(newReplicaNode2.operDataStore(), "people", "member-1");
462     }
463
464     @Test
465     public void testRemoveAllShardReplicas() throws Exception {
466         String name = "testRemoveAllShardReplicas";
467         String moduleShardsConfig = "module-shards-member1-and-2-and-3.conf";
468         MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ).
469                 moduleShardsConfig(moduleShardsConfig).datastoreContextBuilder(DatastoreContext.newBuilder().
470                         shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)).build();
471
472         MemberNode replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name).
473                 moduleShardsConfig(moduleShardsConfig).build();
474
475         MemberNode replicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name).
476                 moduleShardsConfig(moduleShardsConfig).build();
477
478         leaderNode1.configDataStore().waitTillReady();
479         verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars", "member-2", "member-3");
480         verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1", "member-3");
481         verifyRaftPeersPresent(replicaNode3.configDataStore(), "cars", "member-1", "member-2");
482
483         ModuleShardConfiguration petsModuleConfig = new ModuleShardConfiguration(URI.create("pets-ns"), "pets-module",
484                 "pets", null, Arrays.asList(MEMBER_1, MEMBER_2, MEMBER_3));
485         leaderNode1.configDataStore().getActorContext().getShardManager().tell(
486                 new CreateShard(petsModuleConfig, Shard.builder(), null), leaderNode1.kit().getRef());
487         leaderNode1.kit().expectMsgClass(Success.class);
488
489         replicaNode2.configDataStore().getActorContext().getShardManager().tell(
490                 new CreateShard(petsModuleConfig, Shard.builder(), null), replicaNode2.kit().getRef());
491         replicaNode2.kit().expectMsgClass(Success.class);
492
493         replicaNode3.configDataStore().getActorContext().getShardManager().tell(
494                 new CreateShard(petsModuleConfig, Shard.builder(), null), replicaNode3.kit().getRef());
495         replicaNode3.kit().expectMsgClass(Success.class);
496
497         verifyRaftPeersPresent(leaderNode1.configDataStore(), "pets", "member-2", "member-3");
498         verifyRaftPeersPresent(replicaNode2.configDataStore(), "pets", "member-1", "member-3");
499         verifyRaftPeersPresent(replicaNode3.configDataStore(), "pets", "member-1", "member-2");
500
501         ClusterAdminRpcService service3 = new ClusterAdminRpcService(replicaNode3.configDataStore(),
502                 replicaNode3.operDataStore());
503
504         RpcResult<RemoveAllShardReplicasOutput> rpcResult = service3.removeAllShardReplicas(
505                 new RemoveAllShardReplicasInputBuilder().setMemberName("member-3").build()).get(10, TimeUnit.SECONDS);
506         RemoveAllShardReplicasOutput result = verifySuccessfulRpcResult(rpcResult);
507         verifyShardResults(result.getShardResult(), successShardResult("cars", DataStoreType.Config),
508                 successShardResult("people", DataStoreType.Config),
509                 successShardResult("pets", DataStoreType.Config),
510                 successShardResult("cars", DataStoreType.Operational),
511                 successShardResult("people", DataStoreType.Operational));
512
513         verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars", "member-2");
514         verifyRaftPeersPresent(leaderNode1.configDataStore(), "people", "member-2");
515         verifyRaftPeersPresent(leaderNode1.configDataStore(), "pets", "member-2");
516         verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1");
517         verifyRaftPeersPresent(replicaNode2.configDataStore(), "people", "member-1");
518         verifyRaftPeersPresent(replicaNode2.configDataStore(), "pets", "member-1");
519         verifyNoShardPresent(replicaNode3.configDataStore(), "cars");
520         verifyNoShardPresent(replicaNode3.configDataStore(), "people");
521         verifyNoShardPresent(replicaNode3.configDataStore(), "pets");
522     }
523
524     @Test
525     public void testChangeMemberVotingStatesForShard() throws Exception {
526         String name = "testChangeMemberVotingStatusForShard";
527         String moduleShardsConfig = "module-shards-member1-and-2-and-3.conf";
528         MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ).
529                 moduleShardsConfig(moduleShardsConfig).datastoreContextBuilder(DatastoreContext.newBuilder().
530                         shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)).build();
531
532         MemberNode replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name).
533                 moduleShardsConfig(moduleShardsConfig).build();
534
535         MemberNode replicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name).
536                 moduleShardsConfig(moduleShardsConfig).build();
537
538         leaderNode1.configDataStore().waitTillReady();
539         replicaNode3.configDataStore().waitTillReady();
540         verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars", "member-2", "member-3");
541         verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1", "member-3");
542         verifyRaftPeersPresent(replicaNode3.configDataStore(), "cars", "member-1", "member-2");
543
544         // Invoke RPC service on member-3 to change voting status
545
546         ClusterAdminRpcService service3 = new ClusterAdminRpcService(replicaNode3.configDataStore(),
547                 replicaNode3.operDataStore());
548
549         RpcResult<Void> rpcResult = service3.changeMemberVotingStatesForShard(
550                 new ChangeMemberVotingStatesForShardInputBuilder().setShardName("cars").
551                     setDataStoreType(DataStoreType.Config).setMemberVotingState(ImmutableList.of(
552                         new MemberVotingStateBuilder().setMemberName("member-2").setVoting(false).build(),
553                         new MemberVotingStateBuilder().setMemberName("member-3").setVoting(false).build())).build()).
554                 get(10, TimeUnit.SECONDS);
555         verifySuccessfulRpcResult(rpcResult);
556
557         verifyVotingStates(leaderNode1.configDataStore(), "cars", new SimpleEntry<>("member-1", true), new SimpleEntry<>("member-2", false),
558                 new SimpleEntry<>("member-3", false));
559         verifyVotingStates(replicaNode2.configDataStore(), "cars", new SimpleEntry<>("member-1", true), new SimpleEntry<>("member-2", false),
560                 new SimpleEntry<>("member-3", false));
561         verifyVotingStates(replicaNode3.configDataStore(), "cars", new SimpleEntry<>("member-1", true), new SimpleEntry<>("member-2", false),
562                 new SimpleEntry<>("member-3", false));
563     }
564
565     @Test
566     public void testChangeMemberVotingStatesForSingleNodeShard() throws Exception {
567         String name = "testChangeMemberVotingStatesForSingleNodeShard";
568         String moduleShardsConfig = "module-shards-member1.conf";
569         MemberNode leaderNode = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ).
570                 moduleShardsConfig(moduleShardsConfig).datastoreContextBuilder(DatastoreContext.newBuilder().
571                         shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)).build();
572
573         leaderNode.configDataStore().waitTillReady();
574
575         // Invoke RPC service on member-3 to change voting status
576
577         ClusterAdminRpcService service = new ClusterAdminRpcService(leaderNode.configDataStore(),
578                 leaderNode.operDataStore());
579
580         RpcResult<Void> rpcResult = service.changeMemberVotingStatesForShard(
581                 new ChangeMemberVotingStatesForShardInputBuilder().setShardName("cars").
582                     setDataStoreType(DataStoreType.Config).setMemberVotingState(ImmutableList.of(
583                         new MemberVotingStateBuilder().setMemberName("member-1").setVoting(false).build())).build()).
584                 get(10, TimeUnit.SECONDS);
585         verifyFailedRpcResult(rpcResult);
586
587         verifyVotingStates(leaderNode.configDataStore(), "cars", new SimpleEntry<>("member-1", true));
588     }
589
590     @Test
591     public void testChangeMemberVotingStatesForAllShards() throws Exception {
592         String name = "testChangeMemberVotingStatesForAllShards";
593         String moduleShardsConfig = "module-shards-member1-and-2-and-3.conf";
594         MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ).
595                 moduleShardsConfig(moduleShardsConfig).datastoreContextBuilder(DatastoreContext.newBuilder().
596                         shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)).build();
597
598         MemberNode replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name).
599                 moduleShardsConfig(moduleShardsConfig).build();
600
601         MemberNode replicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name).
602                 moduleShardsConfig(moduleShardsConfig).build();
603
604         leaderNode1.configDataStore().waitTillReady();
605         leaderNode1.operDataStore().waitTillReady();
606         replicaNode3.configDataStore().waitTillReady();
607         replicaNode3.operDataStore().waitTillReady();
608         verifyRaftPeersPresent(leaderNode1.configDataStore(), "cars", "member-2", "member-3");
609         verifyRaftPeersPresent(replicaNode2.configDataStore(), "cars", "member-1", "member-3");
610         verifyRaftPeersPresent(replicaNode3.configDataStore(), "cars", "member-1", "member-2");
611
612         // Invoke RPC service on member-3 to change voting status
613
614         ClusterAdminRpcService service3 = new ClusterAdminRpcService(replicaNode3.configDataStore(),
615                 replicaNode3.operDataStore());
616
617         RpcResult<ChangeMemberVotingStatesForAllShardsOutput> rpcResult = service3.changeMemberVotingStatesForAllShards(
618                 new ChangeMemberVotingStatesForAllShardsInputBuilder().setMemberVotingState(ImmutableList.of(
619                         new MemberVotingStateBuilder().setMemberName("member-2").setVoting(false).build(),
620                         new MemberVotingStateBuilder().setMemberName("member-3").setVoting(false).build())).build()).
621                 get(10, TimeUnit.SECONDS);
622         ChangeMemberVotingStatesForAllShardsOutput result = verifySuccessfulRpcResult(rpcResult);
623         verifyShardResults(result.getShardResult(), successShardResult("cars", DataStoreType.Config),
624                 successShardResult("people", DataStoreType.Config),
625                 successShardResult("cars", DataStoreType.Operational),
626                 successShardResult("people", DataStoreType.Operational));
627
628         verifyVotingStates(new DistributedDataStore[]{leaderNode1.configDataStore(), leaderNode1.operDataStore(),
629                 replicaNode2.configDataStore(), replicaNode2.operDataStore(),
630                 replicaNode3.configDataStore(), replicaNode3.operDataStore()},
631                 new String[]{"cars", "people"}, new SimpleEntry<>("member-1", true), new SimpleEntry<>("member-2", false),
632                 new SimpleEntry<>("member-3", false));
633     }
634
635     @Test
636     public void testFlipMemberVotingStates() throws Exception {
637         String name = "testFlipMemberVotingStates";
638
639         ServerConfigurationPayload persistedServerConfig = new ServerConfigurationPayload(Arrays.asList(
640                 new ServerInfo("member-1", true), new ServerInfo("member-2", true),
641                 new ServerInfo("member-3", false)));
642
643         setupPersistedServerConfigPayload(persistedServerConfig, "member-1", name, "cars", "people");
644         setupPersistedServerConfigPayload(persistedServerConfig, "member-2", name, "cars", "people");
645         setupPersistedServerConfigPayload(persistedServerConfig, "member-3", name, "cars", "people");
646
647         String moduleShardsConfig = "module-shards-member1-and-2-and-3.conf";
648         MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ).
649                 moduleShardsConfig(moduleShardsConfig).datastoreContextBuilder(DatastoreContext.newBuilder().
650                         shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)).build();
651
652         MemberNode replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name).
653                 moduleShardsConfig(moduleShardsConfig).build();
654
655         MemberNode replicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name).
656                 moduleShardsConfig(moduleShardsConfig).build();
657
658         leaderNode1.configDataStore().waitTillReady();
659         leaderNode1.operDataStore().waitTillReady();
660         replicaNode3.configDataStore().waitTillReady();
661         replicaNode3.operDataStore().waitTillReady();
662         verifyVotingStates(leaderNode1.configDataStore(), "cars", new SimpleEntry<>("member-1", true),
663                 new SimpleEntry<>("member-2", true), new SimpleEntry<>("member-3", false));
664
665         ClusterAdminRpcService service3 = new ClusterAdminRpcService(replicaNode3.configDataStore(),
666                 replicaNode3.operDataStore());
667
668         RpcResult<FlipMemberVotingStatesForAllShardsOutput> rpcResult = service3.flipMemberVotingStatesForAllShards().
669                 get(10, TimeUnit.SECONDS);
670         FlipMemberVotingStatesForAllShardsOutput result = verifySuccessfulRpcResult(rpcResult);
671         verifyShardResults(result.getShardResult(), successShardResult("cars", DataStoreType.Config),
672                 successShardResult("people", DataStoreType.Config),
673                 successShardResult("cars", DataStoreType.Operational),
674                 successShardResult("people", DataStoreType.Operational));
675
676         verifyVotingStates(new DistributedDataStore[]{leaderNode1.configDataStore(), leaderNode1.operDataStore(),
677                 replicaNode2.configDataStore(), replicaNode2.operDataStore(),
678                 replicaNode3.configDataStore(), replicaNode3.operDataStore()},
679                 new String[]{"cars", "people"},
680                 new SimpleEntry<>("member-1", false), new SimpleEntry<>("member-2", false),
681                 new SimpleEntry<>("member-3", true));
682
683         // Leadership should have transferred to member 3 since it is the only remaining voting member.
684         verifyRaftState(leaderNode1.configDataStore(), "cars", raftState -> {
685             assertNotNull("Expected non-null leader Id", raftState.getLeader());
686             assertTrue("Expected leader member-1. Actual: " + raftState.getLeader(),
687                     raftState.getLeader().contains("member-3"));
688         });
689
690         verifyRaftState(leaderNode1.operDataStore(), "cars", raftState -> {
691             assertNotNull("Expected non-null leader Id", raftState.getLeader());
692             assertTrue("Expected leader member-1. Actual: " + raftState.getLeader(),
693                     raftState.getLeader().contains("member-3"));
694         });
695
696         // Flip the voting states back to the original states.
697
698         rpcResult = service3.flipMemberVotingStatesForAllShards(). get(10, TimeUnit.SECONDS);
699         result = verifySuccessfulRpcResult(rpcResult);
700         verifyShardResults(result.getShardResult(), successShardResult("cars", DataStoreType.Config),
701                 successShardResult("people", DataStoreType.Config),
702                 successShardResult("cars", DataStoreType.Operational),
703                 successShardResult("people", DataStoreType.Operational));
704
705         verifyVotingStates(new DistributedDataStore[]{leaderNode1.configDataStore(), leaderNode1.operDataStore(),
706                 replicaNode2.configDataStore(), replicaNode2.operDataStore(),
707                 replicaNode3.configDataStore(), replicaNode3.operDataStore()},
708                 new String[]{"cars", "people"},
709                 new SimpleEntry<>("member-1", true), new SimpleEntry<>("member-2", true),
710                 new SimpleEntry<>("member-3", false));
711
712         // Leadership should have transferred to member 1 or 2.
713         verifyRaftState(leaderNode1.configDataStore(), "cars", raftState -> {
714             assertNotNull("Expected non-null leader Id", raftState.getLeader());
715             assertTrue("Expected leader member-1 or member-2. Actual: " + raftState.getLeader(),
716                     raftState.getLeader().contains("member-1") || raftState.getLeader().contains("member-2"));
717         });
718     }
719
720     @Test
721     public void testFlipMemberVotingStatesWithNoInitialLeader() throws Exception {
722         String name = "testFlipMemberVotingStatesWithNoInitialLeader";
723
724         // Members 1, 2, and 3 are initially started up as non-voting. Members 4, 5, and 6 are initially
725         // non-voting and simulated as down by not starting them up.
726         ServerConfigurationPayload persistedServerConfig = new ServerConfigurationPayload(Arrays.asList(
727                 new ServerInfo("member-1", false), new ServerInfo("member-2", false),
728                 new ServerInfo("member-3", false), new ServerInfo("member-4", true),
729                 new ServerInfo("member-5", true), new ServerInfo("member-6", true)));
730
731         setupPersistedServerConfigPayload(persistedServerConfig, "member-1", name, "cars", "people");
732         setupPersistedServerConfigPayload(persistedServerConfig, "member-2", name, "cars", "people");
733         setupPersistedServerConfigPayload(persistedServerConfig, "member-3", name, "cars", "people");
734
735         String moduleShardsConfig = "module-shards-member1-and-2-and-3.conf";
736         MemberNode replicaNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ).
737                 moduleShardsConfig(moduleShardsConfig).datastoreContextBuilder(DatastoreContext.newBuilder().
738                         shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)).build();
739
740         MemberNode replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name).
741                 moduleShardsConfig(moduleShardsConfig).build();
742
743         MemberNode replicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name).
744                 moduleShardsConfig(moduleShardsConfig).build();
745
746         // Initially there won't be a leader b/c all the up nodes are non-voting.
747
748         replicaNode1.waitForMembersUp("member-2", "member-3");
749
750         verifyVotingStates(replicaNode1.configDataStore(), "cars", new SimpleEntry<>("member-1", false),
751                 new SimpleEntry<>("member-2", false), new SimpleEntry<>("member-3", false),
752                 new SimpleEntry<>("member-4", true), new SimpleEntry<>("member-5", true),
753                 new SimpleEntry<>("member-6", true));
754
755         verifyRaftState(replicaNode1.configDataStore(), "cars", raftState ->
756             assertEquals("Expected raft state", RaftState.Follower.toString(), raftState.getRaftState()));
757
758         ClusterAdminRpcService service1 = new ClusterAdminRpcService(replicaNode1.configDataStore(),
759                 replicaNode1.operDataStore());
760
761         RpcResult<FlipMemberVotingStatesForAllShardsOutput> rpcResult = service1.flipMemberVotingStatesForAllShards().
762                 get(10, TimeUnit.SECONDS);
763         FlipMemberVotingStatesForAllShardsOutput result = verifySuccessfulRpcResult(rpcResult);
764         verifyShardResults(result.getShardResult(), successShardResult("cars", DataStoreType.Config),
765                 successShardResult("people", DataStoreType.Config),
766                 successShardResult("cars", DataStoreType.Operational),
767                 successShardResult("people", DataStoreType.Operational));
768
769         verifyVotingStates(new DistributedDataStore[]{replicaNode1.configDataStore(), replicaNode1.operDataStore(),
770                 replicaNode2.configDataStore(), replicaNode2.operDataStore(),
771                 replicaNode3.configDataStore(), replicaNode3.operDataStore()},
772                 new String[]{"cars", "people"},
773                 new SimpleEntry<>("member-1", true), new SimpleEntry<>("member-2", true),
774                 new SimpleEntry<>("member-3", true), new SimpleEntry<>("member-4", false),
775                 new SimpleEntry<>("member-5", false), new SimpleEntry<>("member-6", false));
776
777         // Since member 1 was changed to voting and there was no leader, it should've started and election
778         // and become leader
779         verifyRaftState(replicaNode1.configDataStore(), "cars", raftState -> {
780             assertNotNull("Expected non-null leader Id", raftState.getLeader());
781             assertTrue("Expected leader member-1. Actual: " + raftState.getLeader(),
782                     raftState.getLeader().contains("member-1"));
783         });
784
785         verifyRaftState(replicaNode1.operDataStore(), "cars", raftState -> {
786             assertNotNull("Expected non-null leader Id", raftState.getLeader());
787             assertTrue("Expected leader member-1. Actual: " + raftState.getLeader(),
788                     raftState.getLeader().contains("member-1"));
789         });
790     }
791
792     @Test
793     public void testFlipMemberVotingStatesWithVotingMembersDown() throws Exception {
794         String name = "testFlipMemberVotingStatesWithVotingMembersDown";
795
796         // Members 4, 5, and 6 are initially non-voting and simulated as down by not starting them up.
797         ServerConfigurationPayload persistedServerConfig = new ServerConfigurationPayload(Arrays.asList(
798                 new ServerInfo("member-1", true), new ServerInfo("member-2", true),
799                 new ServerInfo("member-3", true), new ServerInfo("member-4", false),
800                 new ServerInfo("member-5", false), new ServerInfo("member-6", false)));
801
802         setupPersistedServerConfigPayload(persistedServerConfig, "member-1", name, "cars", "people");
803         setupPersistedServerConfigPayload(persistedServerConfig, "member-2", name, "cars", "people");
804         setupPersistedServerConfigPayload(persistedServerConfig, "member-3", name, "cars", "people");
805
806         String moduleShardsConfig = "module-shards-member1-and-2-and-3.conf";
807         MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ).
808                 moduleShardsConfig(moduleShardsConfig).datastoreContextBuilder(DatastoreContext.newBuilder().
809                         shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)).build();
810
811         MemberNode replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name).
812                 moduleShardsConfig(moduleShardsConfig).build();
813
814         MemberNode replicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name).
815                 moduleShardsConfig(moduleShardsConfig).build();
816
817         leaderNode1.configDataStore().waitTillReady();
818         leaderNode1.operDataStore().waitTillReady();
819         verifyVotingStates(leaderNode1.configDataStore(), "cars", new SimpleEntry<>("member-1", true),
820                 new SimpleEntry<>("member-2", true), new SimpleEntry<>("member-3", true),
821                 new SimpleEntry<>("member-4", false), new SimpleEntry<>("member-5", false),
822                 new SimpleEntry<>("member-6", false));
823
824         ClusterAdminRpcService service1 = new ClusterAdminRpcService(leaderNode1.configDataStore(),
825                 leaderNode1.operDataStore());
826
827         RpcResult<FlipMemberVotingStatesForAllShardsOutput> rpcResult = service1.flipMemberVotingStatesForAllShards().
828                 get(10, TimeUnit.SECONDS);
829         FlipMemberVotingStatesForAllShardsOutput result = verifySuccessfulRpcResult(rpcResult);
830         verifyShardResults(result.getShardResult(), successShardResult("cars", DataStoreType.Config),
831                 successShardResult("people", DataStoreType.Config),
832                 successShardResult("cars", DataStoreType.Operational),
833                 successShardResult("people", DataStoreType.Operational));
834
835         // Members 2 and 3 are now non-voting but should get replicated with the new new server config.
836         verifyVotingStates(new DistributedDataStore[]{leaderNode1.configDataStore(), leaderNode1.operDataStore(),
837                 replicaNode2.configDataStore(), replicaNode2.operDataStore(),
838                 replicaNode3.configDataStore(), replicaNode3.operDataStore()},
839                 new String[]{"cars", "people"},
840                 new SimpleEntry<>("member-1", false), new SimpleEntry<>("member-2", false),
841                 new SimpleEntry<>("member-3", false), new SimpleEntry<>("member-4", true),
842                 new SimpleEntry<>("member-5", true), new SimpleEntry<>("member-6", true));
843
844         // The leader (member 1) was changed to non-voting but it shouldn't be able to step down as leader yet
845         // b/c it can't get a majority consensus with all voting members down. So verify it remains the leader.
846         verifyRaftState(leaderNode1.configDataStore(), "cars", raftState -> {
847             assertNotNull("Expected non-null leader Id", raftState.getLeader());
848             assertTrue("Expected leader member-1", raftState.getLeader().contains("member-1"));
849         });
850     }
851
852     private void setupPersistedServerConfigPayload(ServerConfigurationPayload serverConfig,
853             String member, String datastoreTypeSuffix, String... shards) {
854         String[] datastoreTypes = {"config_", "oper_"};
855         for(String type: datastoreTypes) {
856             for(String shard: shards) {
857                 List<ServerInfo> newServerInfo = new ArrayList<>(serverConfig.getServerConfig().size());
858                 for(ServerInfo info: serverConfig.getServerConfig()) {
859                     newServerInfo.add(new ServerInfo(ShardIdentifier.create(shard, MemberName.forName(info.getId()),
860                             type + datastoreTypeSuffix).toString(), info.isVoting()));
861                 }
862
863                 String shardID = ShardIdentifier.create(shard, MemberName.forName(member),
864                         type + datastoreTypeSuffix).toString();
865                 InMemoryJournal.addEntry(shardID, 1, new UpdateElectionTerm(1, null));
866                 InMemoryJournal.addEntry(shardID, 2, new ReplicatedLogImplEntry(0, 1,
867                         new ServerConfigurationPayload(newServerInfo)));
868             }
869         }
870     }
871
872     @SafeVarargs
873     private static void verifyVotingStates(DistributedDataStore[] datastores, String[] shards,
874             SimpleEntry<String, Boolean>... expStates) throws Exception {
875         for(DistributedDataStore datastore: datastores) {
876             for(String shard: shards) {
877                 verifyVotingStates(datastore, shard, expStates);
878             }
879         }
880     }
881
882     @SafeVarargs
883     private static void verifyVotingStates(DistributedDataStore datastore, String shardName,
884             SimpleEntry<String, Boolean>... expStates) throws Exception {
885         String localMemberName = datastore.getActorContext().getCurrentMemberName().getName();
886         Map<String, Boolean> expStateMap = new HashMap<>();
887         for(Entry<String, Boolean> e: expStates) {
888             expStateMap.put(ShardIdentifier.create(shardName, MemberName.forName(e.getKey()),
889                     datastore.getActorContext().getDataStoreName()).toString(), e.getValue());
890         }
891
892         verifyRaftState(datastore, shardName, raftState -> {
893             String localPeerId = ShardIdentifier.create(shardName, MemberName.forName(localMemberName),
894                     datastore.getActorContext().getDataStoreName()).toString();
895             assertEquals("Voting state for " + localPeerId, expStateMap.get(localPeerId), raftState.isVoting());
896             for(Entry<String, Boolean> e: raftState.getPeerVotingStates().entrySet()) {
897                 assertEquals("Voting state for " + e.getKey(), expStateMap.get(e.getKey()), e.getValue());
898             }
899         });
900     }
901
902     private static void verifyShardResults(List<ShardResult> shardResults, ShardResult... expShardResults) {
903         Map<String, ShardResult> expResultsMap = new HashMap<>();
904         for(ShardResult r: expShardResults) {
905             expResultsMap.put(r.getShardName() + "-" + r.getDataStoreType(), r);
906         }
907
908         for(ShardResult result: shardResults) {
909             ShardResult exp = expResultsMap.remove(result.getShardName() + "-" + result.getDataStoreType());
910             assertNotNull(String.format("Unexpected result for shard %s, type %s", result.getShardName(),
911                     result.getDataStoreType()), exp);
912             assertEquals("isSucceeded", exp.isSucceeded(), result.isSucceeded());
913             if(exp.isSucceeded()) {
914                 assertNull("Expected null error message", result.getErrorMessage());
915             } else {
916                 assertNotNull("Expected error message", result.getErrorMessage());
917             }
918         }
919
920         if(!expResultsMap.isEmpty()) {
921             fail("Missing shard results for " + expResultsMap.keySet());
922         }
923     }
924
925     private static ShardResult successShardResult(String shardName, DataStoreType type) {
926         return new ShardResultBuilder().setDataStoreType(type).setShardName(shardName).setSucceeded(true).build();
927     }
928
929     private static ShardResult failedShardResult(String shardName, DataStoreType type) {
930         return new ShardResultBuilder().setDataStoreType(type).setShardName(shardName).setSucceeded(false).build();
931     }
932 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.