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