@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); }
@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 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 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); }
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); }
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);  }