class AggregatingLoadbalancerService implements LoadBalancerService {

    @Override
    public Observable<GetJobLoadBalancersResult> getLoadBalancers(JobId jobId, CallMetadata callMetadata) {
        ClientCall<GetJobLoadBalancersResult> getLoadBalancersForJob =
                (client, responseObserver) -> wrap(client, callMetadata).getJobLoadBalancers(jobId, responseObserver);

        return aggregatingClient.callExpectingErrors(LoadBalancerServiceGrpc::newStub, getLoadBalancersForJob)
                .filter(response -> response.getResult().hasError() || response.getResult().getValue().getLoadBalancersCount() > 0)
                .reduce(ResponseMerger.emptyResponseMarker(), ResponseMerger.singleValue())
                .filter(ResponseMerger::isNotEmptyResponseMarker)
                .flatMap(response -> response.getResult().map(Observable::just).onErrorGet(Observable::error))
                .switchIfEmpty(Observable.just(GetJobLoadBalancersResult.newBuilder().setJobId(jobId.getId()).build()));
    }

    @Inject public  AggregatingLoadbalancerService(CellConnector connector,
                                          CallMetadataResolver callMetadataResolver,
                                          GrpcConfiguration grpcConfiguration,
                                          AggregatingCellClient aggregatingClient,
                                          AggregatingJobManagementServiceHelper jobManagementServiceHelper);

}