@Test public void testGetShardRole() throws Exception { String name = "testGetShardRole"; String moduleShardsConfig = "module-shards-default-member-1.conf"; final MemberNode member1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name) .moduleShardsConfig(moduleShardsConfig).build(); member1.kit().waitUntilLeader(member1.configDataStore().getActorUtils(), "default"); final RpcResult<GetShardRoleOutput> successResult = getShardRole(member1, Mockito.mock(BindingNormalizedNodeSerializer.class), "default"); verifySuccessfulRpcResult(successResult); assertEquals("Leader", successResult.getResult().getRole()); final RpcResult<GetShardRoleOutput> failedResult = getShardRole(member1, Mockito.mock(BindingNormalizedNodeSerializer.class), "cars"); verifyFailedRpcResult(failedResult); final ActorRef shardManager1 = member1.configDataStore().getActorUtils().getShardManager(); shardManager1.tell(new PrefixShardCreated(new PrefixShardConfiguration( new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH), "prefix", Collections.singleton(MEMBER_1))), ActorRef.noSender()); member1.kit().waitUntilLeader(member1.configDataStore().getActorUtils(), ClusterUtils.getCleanShardName(CarsModel.BASE_PATH)); final InstanceIdentifier<Cars> identifier = InstanceIdentifier.create(Cars.class); final BindingNormalizedNodeSerializer serializer = Mockito.mock(BindingNormalizedNodeSerializer.class); Mockito.doReturn(CarsModel.BASE_PATH).when(serializer).toYangInstanceIdentifier(identifier); final RpcResult<GetPrefixShardRoleOutput> prefixSuccessResult = getPrefixShardRole(member1, identifier, serializer); verifySuccessfulRpcResult(prefixSuccessResult); assertEquals("Leader", prefixSuccessResult.getResult().getRole()); final InstanceIdentifier<People> peopleId = InstanceIdentifier.create(People.class); Mockito.doReturn(PeopleModel.BASE_PATH).when(serializer).toYangInstanceIdentifier(peopleId); final RpcResult<GetPrefixShardRoleOutput> prefixFail = getPrefixShardRole(member1, peopleId, serializer); verifyFailedRpcResult(prefixFail); }
@Override public ListenableFuture<RpcResult<GetShardRoleOutput>> getShardRole(final GetShardRoleInput 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("Getting role for shard {}, datastore type {}", shardName, dataStoreType); final SettableFuture<RpcResult<GetShardRoleOutput>> returnFuture = SettableFuture.create(); ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType, new GetShardRole(shardName)); Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() { @Override public void onSuccess(final GetShardRoleReply reply) { if (reply == null) { returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder( "No Shard role present. Please retry..").build()); return; } LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName); final GetShardRoleOutputBuilder builder = new GetShardRoleOutputBuilder(); if (reply.getRole() != null) { builder.setRole(reply.getRole()); } returnFuture.set(newSuccessfulResult(builder.build())); } @Override public void onFailure(final Throwable failure) { returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder( "Failed to get shard role.", failure).build()); } }, MoreExecutors.directExecutor()); return returnFuture; }
ClusterAdminRpcService implements ClusterAdminService { @Override public ListenableFuture<RpcResult<GetShardRoleOutput>> getShardRole(final GetShardRoleInput 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("Getting role for shard {}, datastore type {}", shardName, dataStoreType); final SettableFuture<RpcResult<GetShardRoleOutput>> returnFuture = SettableFuture.create(); ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType, new GetShardRole(shardName)); Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() { @Override public void onSuccess(final GetShardRoleReply reply) { if (reply == null) { returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder( "No Shard role present. Please retry..").build()); return; } LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName); final GetShardRoleOutputBuilder builder = new GetShardRoleOutputBuilder(); if (reply.getRole() != null) { builder.setRole(reply.getRole()); } returnFuture.set(newSuccessfulResult(builder.build())); } @Override public void onFailure(final Throwable failure) { returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder( "Failed to get shard role.", failure).build()); } }, MoreExecutors.directExecutor()); return returnFuture; } }
ClusterAdminRpcService implements ClusterAdminService { @Override public ListenableFuture<RpcResult<GetShardRoleOutput>> getShardRole(final GetShardRoleInput 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("Getting role for shard {}, datastore type {}", shardName, dataStoreType); final SettableFuture<RpcResult<GetShardRoleOutput>> returnFuture = SettableFuture.create(); ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType, new GetShardRole(shardName)); Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() { @Override public void onSuccess(final GetShardRoleReply reply) { if (reply == null) { returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder( "No Shard role present. Please retry..").build()); return; } LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName); final GetShardRoleOutputBuilder builder = new GetShardRoleOutputBuilder(); if (reply.getRole() != null) { builder.setRole(reply.getRole()); } returnFuture.set(newSuccessfulResult(builder.build())); } @Override public void onFailure(final Throwable failure) { returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder( "Failed to get shard role.", failure).build()); } }, MoreExecutors.directExecutor()); return returnFuture; } ClusterAdminRpcService(final DistributedDataStoreInterface configDataStore,
            final DistributedDataStoreInterface operDataStore,
            final BindingNormalizedNodeSerializer serializer); }
ClusterAdminRpcService implements ClusterAdminService { @Override public ListenableFuture<RpcResult<GetShardRoleOutput>> getShardRole(final GetShardRoleInput 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("Getting role for shard {}, datastore type {}", shardName, dataStoreType); final SettableFuture<RpcResult<GetShardRoleOutput>> returnFuture = SettableFuture.create(); ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType, new GetShardRole(shardName)); Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() { @Override public void onSuccess(final GetShardRoleReply reply) { if (reply == null) { returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder( "No Shard role present. Please retry..").build()); return; } LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName); final GetShardRoleOutputBuilder builder = new GetShardRoleOutputBuilder(); if (reply.getRole() != null) { builder.setRole(reply.getRole()); } returnFuture.set(newSuccessfulResult(builder.build())); } @Override public void onFailure(final Throwable failure) { returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder( "Failed to get shard role.", failure).build()); } }, 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<GetShardRoleOutput>> getShardRole(final GetShardRoleInput 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("Getting role for shard {}, datastore type {}", shardName, dataStoreType); final SettableFuture<RpcResult<GetShardRoleOutput>> returnFuture = SettableFuture.create(); ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType, new GetShardRole(shardName)); Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() { @Override public void onSuccess(final GetShardRoleReply reply) { if (reply == null) { returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder( "No Shard role present. Please retry..").build()); return; } LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName); final GetShardRoleOutputBuilder builder = new GetShardRoleOutputBuilder(); if (reply.getRole() != null) { builder.setRole(reply.getRole()); } returnFuture.set(newSuccessfulResult(builder.build())); } @Override public void onFailure(final Throwable failure) { returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder( "Failed to get shard role.", failure).build()); } }, 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);  }