@Test public void testAddReplicasForAllShards() throws Exception { String name = "testAddReplicasForAllShards"; String moduleShardsConfig = "module-shards-member1.conf"; MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name) .moduleShardsConfig(moduleShardsConfig).waitForShardLeader("cars", "people").build(); ModuleShardConfiguration petsModuleConfig = new ModuleShardConfiguration(URI.create("pets-ns"), "pets-module", "pets", null, Collections.singletonList(MEMBER_1)); leaderNode1.configDataStore().getActorUtils().getShardManager().tell( new CreateShard(petsModuleConfig, Shard.builder(), null), leaderNode1.kit().getRef()); leaderNode1.kit().expectMsgClass(Success.class); leaderNode1.kit().waitUntilLeader(leaderNode1.configDataStore().getActorUtils(), "pets"); MemberNode newReplicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name) .moduleShardsConfig(moduleShardsConfig).build(); leaderNode1.waitForMembersUp("member-2"); newReplicaNode2.waitForMembersUp("member-1"); newReplicaNode2.configDataStore().getActorUtils().getShardManager().tell( new CreateShard(petsModuleConfig, Shard.builder(), null), newReplicaNode2.kit().getRef()); newReplicaNode2.kit().expectMsgClass(Success.class); newReplicaNode2.operDataStore().getActorUtils().getShardManager().tell( new CreateShard(new ModuleShardConfiguration(URI.create("no-leader-ns"), "no-leader-module", "no-leader", null, Collections.singletonList(MEMBER_1)), Shard.builder(), null), newReplicaNode2.kit().getRef()); newReplicaNode2.kit().expectMsgClass(Success.class); ClusterAdminRpcService service = new ClusterAdminRpcService(newReplicaNode2.configDataStore(), newReplicaNode2.operDataStore(), null); RpcResult<AddReplicasForAllShardsOutput> rpcResult = service.addReplicasForAllShards( new AddReplicasForAllShardsInputBuilder().build()).get(10, TimeUnit.SECONDS); AddReplicasForAllShardsOutput result = verifySuccessfulRpcResult(rpcResult); verifyShardResults(result.getShardResult(), successShardResult("cars", DataStoreType.Config), successShardResult("people", DataStoreType.Config), successShardResult("pets", DataStoreType.Config), successShardResult("cars", DataStoreType.Operational), successShardResult("people", DataStoreType.Operational), failedShardResult("no-leader", DataStoreType.Operational)); verifyRaftPeersPresent(newReplicaNode2.configDataStore(), "cars", "member-1"); verifyRaftPeersPresent(newReplicaNode2.configDataStore(), "people", "member-1"); verifyRaftPeersPresent(newReplicaNode2.configDataStore(), "pets", "member-1"); verifyRaftPeersPresent(newReplicaNode2.operDataStore(), "cars", "member-1"); verifyRaftPeersPresent(newReplicaNode2.operDataStore(), "people", "member-1"); }
@Override public ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards( final AddReplicasForAllShardsInput input) { LOG.info("Adding replicas for all shards"); final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>(); sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, AddShardReplica::new); sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, AddShardReplica::new); return waitForShardResults(shardResultData, shardResults -> new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(), "Failed to add replica"); }
ClusterAdminRpcService implements ClusterAdminService { @Override public ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards( final AddReplicasForAllShardsInput input) { LOG.info("Adding replicas for all shards"); final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>(); sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, AddShardReplica::new); sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, AddShardReplica::new); return waitForShardResults(shardResultData, shardResults -> new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(), "Failed to add replica"); } }
ClusterAdminRpcService implements ClusterAdminService { @Override public ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards( final AddReplicasForAllShardsInput input) { LOG.info("Adding replicas for all shards"); final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>(); sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, AddShardReplica::new); sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, AddShardReplica::new); return waitForShardResults(shardResultData, shardResults -> new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(), "Failed to add replica"); } ClusterAdminRpcService(final DistributedDataStoreInterface configDataStore,
            final DistributedDataStoreInterface operDataStore,
            final BindingNormalizedNodeSerializer serializer); }
ClusterAdminRpcService implements ClusterAdminService { @Override public ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards( final AddReplicasForAllShardsInput input) { LOG.info("Adding replicas for all shards"); final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>(); sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, AddShardReplica::new); sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, AddShardReplica::new); return waitForShardResults(shardResultData, shardResults -> new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(), "Failed to add replica"); } 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); }
ClusterAdminRpcService implements ClusterAdminService { @Override public ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards( final AddReplicasForAllShardsInput input) { LOG.info("Adding replicas for all shards"); final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>(); sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, AddShardReplica::new); sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, AddShardReplica::new); return waitForShardResults(shardResultData, shardResults -> new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(), "Failed to add replica"); } 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);  }