ClusterAdminRpcService implements ClusterAdminService { @Override public ListenableFuture<RpcResult<AddShardReplicaOutput>> addShardReplica(final AddShardReplicaInput input) { final String shardName = input.getShardName(); if (Strings.isNullOrEmpty(shardName)) { return newFailedRpcResultFuture("A valid shard name must be specified"); } DataStoreType dataStoreType = input.getDataStoreType(); if (dataStoreType == null) { return newFailedRpcResultFuture("A valid DataStoreType must be specified"); } LOG.info("Adding replica for shard {}", shardName); final SettableFuture<RpcResult<AddShardReplicaOutput>> returnFuture = SettableFuture.create(); ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName)); Futures.addCallback(future, new FutureCallback<Success>() { @Override public void onSuccess(final Success success) { LOG.info("Successfully added replica for shard {}", shardName); returnFuture.set(newSuccessfulResult(new AddShardReplicaOutputBuilder().build())); } @Override public void onFailure(final Throwable failure) { onMessageFailure(String.format("Failed to add replica for shard %s", shardName), returnFuture, failure); } }, MoreExecutors.directExecutor()); return returnFuture; } ClusterAdminRpcService(final DistributedDataStoreInterface configDataStore,
            final DistributedDataStoreInterface operDataStore,
            final BindingNormalizedNodeSerializer serializer); @Override ListenableFuture<RpcResult<AddShardReplicaOutput>> addShardReplica(final AddShardReplicaInput input); @Override ListenableFuture<RpcResult<RemoveShardReplicaOutput>> removeShardReplica(
            final RemoveShardReplicaInput input); @Override ListenableFuture<RpcResult<LocateShardOutput>> locateShard(final LocateShardInput input); @Override ListenableFuture<RpcResult<MakeLeaderLocalOutput>> makeLeaderLocal(final MakeLeaderLocalInput input); @Override ListenableFuture<RpcResult<AddPrefixShardReplicaOutput>> addPrefixShardReplica(
            final AddPrefixShardReplicaInput input); @Override ListenableFuture<RpcResult<RemovePrefixShardReplicaOutput>> removePrefixShardReplica(
            final RemovePrefixShardReplicaInput input); @Override ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards(
            final AddReplicasForAllShardsInput input); @Override ListenableFuture<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(
            final RemoveAllShardReplicasInput input); @Override ListenableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> changeMemberVotingStatesForShard(
            final ChangeMemberVotingStatesForShardInput input); @Override ListenableFuture<RpcResult<ChangeMemberVotingStatesForAllShardsOutput>> changeMemberVotingStatesForAllShards(
            final ChangeMemberVotingStatesForAllShardsInput input); @Override ListenableFuture<RpcResult<FlipMemberVotingStatesForAllShardsOutput>> flipMemberVotingStatesForAllShards(
            final FlipMemberVotingStatesForAllShardsInput input); @Override ListenableFuture<RpcResult<GetShardRoleOutput>> getShardRole(final GetShardRoleInput input); @Override ListenableFuture<RpcResult<GetPrefixShardRoleOutput>> getPrefixShardRole(
            final GetPrefixShardRoleInput input); @Override ListenableFuture<RpcResult<BackupDatastoreOutput>> backupDatastore(final BackupDatastoreInput input); @Override ListenableFuture<RpcResult<GetKnownClientsForAllShardsOutput>> getKnownClientsForAllShards(
            final GetKnownClientsForAllShardsInput input);  }
@Test public void testAddShardReplica() throws Exception { String name = "testAddShardReplica"; String moduleShardsConfig = "module-shards-cars-member-1.conf"; MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name) .moduleShardsConfig(moduleShardsConfig).waitForShardLeader("cars").build(); MemberNode newReplicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name) .moduleShardsConfig(moduleShardsConfig).build(); leaderNode1.waitForMembersUp("member-2"); doAddShardReplica(newReplicaNode2, "cars", "member-1"); MemberNode newReplicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name) .moduleShardsConfig(moduleShardsConfig).build(); leaderNode1.waitForMembersUp("member-3"); newReplicaNode2.waitForMembersUp("member-3"); doAddShardReplica(newReplicaNode3, "cars", "member-1", "member-2"); verifyRaftPeersPresent(newReplicaNode2.configDataStore(), "cars", "member-1", "member-3"); verifyRaftPeersPresent(newReplicaNode2.operDataStore(), "cars", "member-1", "member-3"); final NormalizedNode<?, ?> configCarsNode = writeCarsNodeAndVerify(newReplicaNode2.configDataStore(), newReplicaNode3.configDataStore()); writeCarsNodeAndVerify(newReplicaNode3.operDataStore(), newReplicaNode2.operDataStore()); RaftStateVerifier verifier = raftState -> { assertEquals("Commit index", 4, raftState.getCommitIndex()); assertEquals("Last applied index", 4, raftState.getLastApplied()); }; verifyRaftState(leaderNode1.configDataStore(), "cars", verifier); verifyRaftState(leaderNode1.operDataStore(), "cars", verifier); verifyRaftState(newReplicaNode2.configDataStore(), "cars", verifier); verifyRaftState(newReplicaNode2.operDataStore(), "cars", verifier); verifyRaftState(newReplicaNode3.configDataStore(), "cars", verifier); verifyRaftState(newReplicaNode3.operDataStore(), "cars", verifier); Cluster.get(leaderNode1.kit().getSystem()).down(Cluster.get(newReplicaNode3.kit().getSystem()).selfAddress()); newReplicaNode3.cleanup(); newReplicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name) .moduleShardsConfig(moduleShardsConfig).createOperDatastore(false).build(); verifyRaftState(newReplicaNode3.configDataStore(), "cars", verifier); readCarsNodeAndVerify(newReplicaNode3.configDataStore(), configCarsNode); }
@Test public void testAddShardReplicaFailures() throws Exception { String name = "testAddShardReplicaFailures"; MemberNode memberNode = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name) .moduleShardsConfig("module-shards-cars-member-1.conf").build(); ClusterAdminRpcService service = new ClusterAdminRpcService(memberNode.configDataStore(), memberNode.operDataStore(), null); RpcResult<AddShardReplicaOutput> rpcResult = service.addShardReplica(new AddShardReplicaInputBuilder() .setDataStoreType(DataStoreType.Config).build()).get(10, TimeUnit.SECONDS); verifyFailedRpcResult(rpcResult); rpcResult = service.addShardReplica(new AddShardReplicaInputBuilder().setShardName("cars") .build()).get(10, TimeUnit.SECONDS); verifyFailedRpcResult(rpcResult); rpcResult = service.addShardReplica(new AddShardReplicaInputBuilder().setShardName("people") .setDataStoreType(DataStoreType.Config).build()).get(10, TimeUnit.SECONDS); verifyFailedRpcResult(rpcResult); }