ClusterAdminRpcService implements ClusterAdminService { @Override public ListenableFuture<RpcResult<BackupDatastoreOutput>> backupDatastore(final BackupDatastoreInput input) { LOG.debug("backupDatastore: {}", input); if (Strings.isNullOrEmpty(input.getFilePath())) { return newFailedRpcResultFuture("A valid file path must be specified"); } final Uint32 timeout = input.getTimeout(); final Timeout opTimeout = timeout != null ? Timeout.apply(timeout.longValue(), TimeUnit.SECONDS) : SHARD_MGR_TIMEOUT; final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture = SettableFuture.create(); ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(new GetSnapshot(opTimeout)); Futures.addCallback(future, new FutureCallback<>() { @Override public void onSuccess(final List<DatastoreSnapshot> snapshots) { saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture); } @Override public void onFailure(final Throwable failure) { onDatastoreBackupFailure(input.getFilePath(), 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 testBackupDatastore() throws Exception { MemberNode node = MemberNode.builder(memberNodes).akkaConfig("Member1") .moduleShardsConfig("module-shards-member1.conf").waitForShardLeader("cars", "people") .testName("testBackupDatastore").build(); String fileName = "target/testBackupDatastore"; new File(fileName).delete(); ClusterAdminRpcService service = new ClusterAdminRpcService(node.configDataStore(), node.operDataStore(), null); RpcResult<BackupDatastoreOutput> rpcResult = service .backupDatastore(new BackupDatastoreInputBuilder() .setFilePath(fileName).build()).get(5, TimeUnit.SECONDS); verifySuccessfulRpcResult(rpcResult); try (FileInputStream fis = new FileInputStream(fileName)) { List<DatastoreSnapshot> snapshots = SerializationUtils.deserialize(fis); assertEquals("DatastoreSnapshot size", 2, snapshots.size()); ImmutableMap<String, DatastoreSnapshot> map = ImmutableMap.of(snapshots.get(0).getType(), snapshots.get(0), snapshots.get(1).getType(), snapshots.get(1)); verifyDatastoreSnapshot(node.configDataStore().getActorUtils().getDataStoreName(), map.get(node.configDataStore().getActorUtils().getDataStoreName()), "cars", "people"); } finally { new File(fileName).delete(); } node.configDataStore().getActorUtils().getShardManager().tell(node.datastoreContextBuilder() .shardInitializationTimeout(200, TimeUnit.MILLISECONDS).build(), ActorRef.noSender()); ActorRef carsShardActor = node.configDataStore().getActorUtils().findLocalShard("cars").get(); node.kit().watch(carsShardActor); carsShardActor.tell(PoisonPill.getInstance(), ActorRef.noSender()); node.kit().expectTerminated(carsShardActor); rpcResult = service.backupDatastore(new BackupDatastoreInputBuilder().setFilePath(fileName).build()) .get(5, TimeUnit.SECONDS); assertFalse("isSuccessful", rpcResult.isSuccessful()); assertEquals("getErrors", 1, rpcResult.getErrors().size()); }