@Test public void findJobsWithFailingCell() { List<Job> cellTwoSnapshot = new ArrayList<>(); for (int i = 0; i < 5; i++) { cellTwoSnapshot.addAll(dataGenerator.newBatchJobs(5, GrpcJobManagementModelConverters::toGrpcJob)); cellTwoSnapshot.addAll(dataGenerator.newServiceJobs(5, GrpcJobManagementModelConverters::toGrpcJob)); clock.advanceTime(1, TimeUnit.MINUTES); } cellOne.getServiceRegistry().addService(new CellWithFailingJobManagementService()); cellTwo.getServiceRegistry().addService(new CellWithFixedJobsService(cellTwoSnapshot, cellTwoUpdates.serialize())); JobQuery query = JobQuery.newBuilder() .setPage(toGrpcPage(Page.unlimited())) .build(); final AssertableSubscriber<JobQueryResult> testSubscriber = service.findJobs(query, UNDEFINED_CALL_METADATA).test(); testSubscriber.awaitTerminalEvent(1, TimeUnit.SECONDS); testSubscriber.assertError(Status.INTERNAL.asRuntimeException().getClass()); testSubscriber.assertNoValues(); }
@Override public Observable<JobQueryResult> findJobs(JobQuery request, CallMetadata callMetadata) { if (request.getPage().getPageSize() <= 0) { return Observable.just(JobQueryResult.newBuilder() .setPagination(emptyGrpcPagination(request.getPage())) .build()); } Set<String> fieldsFilter = Collections.emptySet(); if (request.getFieldsCount() > 0) { fieldsFilter = new HashSet<>(request.getFieldsList()); fieldsFilter.addAll(JOB_MINIMUM_FIELD_SET); request = request.toBuilder() .addAllFields(fieldsFilter) .addAllFields(JOB_FEDERATION_MINIMUM_FIELD_SET) .build(); } if (StringExt.isNotEmpty(request.getPage().getCursor()) || request.getPage().getPageNumber() == 0) { return findJobsWithCursorPagination(request, fieldsFilter, callMetadata); } return Observable.error(TitusServiceException.invalidArgument("pageNumbers are not supported, please use cursors")); }
AggregatingJobServiceGateway implements JobServiceGateway { @Override public Observable<JobQueryResult> findJobs(JobQuery request, CallMetadata callMetadata) { if (request.getPage().getPageSize() <= 0) { return Observable.just(JobQueryResult.newBuilder() .setPagination(emptyGrpcPagination(request.getPage())) .build()); } Set<String> fieldsFilter = Collections.emptySet(); if (request.getFieldsCount() > 0) { fieldsFilter = new HashSet<>(request.getFieldsList()); fieldsFilter.addAll(JOB_MINIMUM_FIELD_SET); request = request.toBuilder() .addAllFields(fieldsFilter) .addAllFields(JOB_FEDERATION_MINIMUM_FIELD_SET) .build(); } if (StringExt.isNotEmpty(request.getPage().getCursor()) || request.getPage().getPageNumber() == 0) { return findJobsWithCursorPagination(request, fieldsFilter, callMetadata); } return Observable.error(TitusServiceException.invalidArgument("pageNumbers are not supported, please use cursors")); } }
AggregatingJobServiceGateway implements JobServiceGateway { @Override public Observable<JobQueryResult> findJobs(JobQuery request, CallMetadata callMetadata) { if (request.getPage().getPageSize() <= 0) { return Observable.just(JobQueryResult.newBuilder() .setPagination(emptyGrpcPagination(request.getPage())) .build()); } Set<String> fieldsFilter = Collections.emptySet(); if (request.getFieldsCount() > 0) { fieldsFilter = new HashSet<>(request.getFieldsList()); fieldsFilter.addAll(JOB_MINIMUM_FIELD_SET); request = request.toBuilder() .addAllFields(fieldsFilter) .addAllFields(JOB_FEDERATION_MINIMUM_FIELD_SET) .build(); } if (StringExt.isNotEmpty(request.getPage().getCursor()) || request.getPage().getPageNumber() == 0) { return findJobsWithCursorPagination(request, fieldsFilter, callMetadata); } return Observable.error(TitusServiceException.invalidArgument("pageNumbers are not supported, please use cursors")); } @Inject  AggregatingJobServiceGateway(GrpcConfiguration grpcConfiguration,
                                        TitusFederationConfiguration federationConfiguration,
                                        CellConnector connector,
                                        CellRouter router,
                                        AggregatingCellClient aggregatingClient,
                                        AggregatingJobManagementServiceHelper jobManagementServiceHelper); }
AggregatingJobServiceGateway implements JobServiceGateway { @Override public Observable<JobQueryResult> findJobs(JobQuery request, CallMetadata callMetadata) { if (request.getPage().getPageSize() <= 0) { return Observable.just(JobQueryResult.newBuilder() .setPagination(emptyGrpcPagination(request.getPage())) .build()); } Set<String> fieldsFilter = Collections.emptySet(); if (request.getFieldsCount() > 0) { fieldsFilter = new HashSet<>(request.getFieldsList()); fieldsFilter.addAll(JOB_MINIMUM_FIELD_SET); request = request.toBuilder() .addAllFields(fieldsFilter) .addAllFields(JOB_FEDERATION_MINIMUM_FIELD_SET) .build(); } if (StringExt.isNotEmpty(request.getPage().getCursor()) || request.getPage().getPageNumber() == 0) { return findJobsWithCursorPagination(request, fieldsFilter, callMetadata); } return Observable.error(TitusServiceException.invalidArgument("pageNumbers are not supported, please use cursors")); } @Inject  AggregatingJobServiceGateway(GrpcConfiguration grpcConfiguration,
                                        TitusFederationConfiguration federationConfiguration,
                                        CellConnector connector,
                                        CellRouter router,
                                        AggregatingCellClient aggregatingClient,
                                        AggregatingJobManagementServiceHelper jobManagementServiceHelper); @Override Observable<String> createJob(JobDescriptor jobDescriptor, CallMetadata callMetadata); @Override Completable updateJobCapacity(JobCapacityUpdate request, CallMetadata callMetadata); @Override Completable updateJobCapacityWithOptionalAttributes(JobCapacityUpdateWithOptionalAttributes request, CallMetadata callMetadata); @Override Completable updateJobProcesses(JobProcessesUpdate request, CallMetadata callMetadata); @Override Completable updateJobStatus(JobStatusUpdate request, CallMetadata callMetadata); @Override Mono<Void> updateJobDisruptionBudget(JobDisruptionBudgetUpdate request, CallMetadata callMetadata); @Override Mono<Void> updateJobAttributes(JobAttributesUpdate request, CallMetadata callMetadata); @Override Mono<Void> deleteJobAttributes(JobAttributesDeleteRequest request, CallMetadata callMetadata); @Override Observable<Job> findJob(String jobId, CallMetadata callMetadata); @Override Observable<JobQueryResult> findJobs(JobQuery request, CallMetadata callMetadata); @Override Observable<JobChangeNotification> observeJob(String jobId, CallMetadata callMetadata); @Override Observable<JobChangeNotification> observeJobs(ObserveJobsQuery query, CallMetadata callMetadata); @Override Completable killJob(String jobId, CallMetadata callMetadata); @Override Observable<Task> findTask(String taskId, CallMetadata callMetadata); @Override Observable<TaskQueryResult> findTasks(TaskQuery request, CallMetadata callMetadata); @Override Completable killTask(TaskKillRequest request, CallMetadata callMetadata); @Override Completable updateTaskAttributes(TaskAttributesUpdate attributesUpdate, CallMetadata callMetadata); @Override Completable deleteTaskAttributes(TaskAttributesDeleteRequest deleteRequest, CallMetadata callMetadata); @Override Completable moveTask(TaskMoveRequest taskMoveRequest, CallMetadata callMetadata); }
AggregatingJobServiceGateway implements JobServiceGateway { @Override public Observable<JobQueryResult> findJobs(JobQuery request, CallMetadata callMetadata) { if (request.getPage().getPageSize() <= 0) { return Observable.just(JobQueryResult.newBuilder() .setPagination(emptyGrpcPagination(request.getPage())) .build()); } Set<String> fieldsFilter = Collections.emptySet(); if (request.getFieldsCount() > 0) { fieldsFilter = new HashSet<>(request.getFieldsList()); fieldsFilter.addAll(JOB_MINIMUM_FIELD_SET); request = request.toBuilder() .addAllFields(fieldsFilter) .addAllFields(JOB_FEDERATION_MINIMUM_FIELD_SET) .build(); } if (StringExt.isNotEmpty(request.getPage().getCursor()) || request.getPage().getPageNumber() == 0) { return findJobsWithCursorPagination(request, fieldsFilter, callMetadata); } return Observable.error(TitusServiceException.invalidArgument("pageNumbers are not supported, please use cursors")); } @Inject  AggregatingJobServiceGateway(GrpcConfiguration grpcConfiguration,
                                        TitusFederationConfiguration federationConfiguration,
                                        CellConnector connector,
                                        CellRouter router,
                                        AggregatingCellClient aggregatingClient,
                                        AggregatingJobManagementServiceHelper jobManagementServiceHelper); @Override Observable<String> createJob(JobDescriptor jobDescriptor, CallMetadata callMetadata); @Override Completable updateJobCapacity(JobCapacityUpdate request, CallMetadata callMetadata); @Override Completable updateJobCapacityWithOptionalAttributes(JobCapacityUpdateWithOptionalAttributes request, CallMetadata callMetadata); @Override Completable updateJobProcesses(JobProcessesUpdate request, CallMetadata callMetadata); @Override Completable updateJobStatus(JobStatusUpdate request, CallMetadata callMetadata); @Override Mono<Void> updateJobDisruptionBudget(JobDisruptionBudgetUpdate request, CallMetadata callMetadata); @Override Mono<Void> updateJobAttributes(JobAttributesUpdate request, CallMetadata callMetadata); @Override Mono<Void> deleteJobAttributes(JobAttributesDeleteRequest request, CallMetadata callMetadata); @Override Observable<Job> findJob(String jobId, CallMetadata callMetadata); @Override Observable<JobQueryResult> findJobs(JobQuery request, CallMetadata callMetadata); @Override Observable<JobChangeNotification> observeJob(String jobId, CallMetadata callMetadata); @Override Observable<JobChangeNotification> observeJobs(ObserveJobsQuery query, CallMetadata callMetadata); @Override Completable killJob(String jobId, CallMetadata callMetadata); @Override Observable<Task> findTask(String taskId, CallMetadata callMetadata); @Override Observable<TaskQueryResult> findTasks(TaskQuery request, CallMetadata callMetadata); @Override Completable killTask(TaskKillRequest request, CallMetadata callMetadata); @Override Completable updateTaskAttributes(TaskAttributesUpdate attributesUpdate, CallMetadata callMetadata); @Override Completable deleteTaskAttributes(TaskAttributesDeleteRequest deleteRequest, CallMetadata callMetadata); @Override Completable moveTask(TaskMoveRequest taskMoveRequest, CallMetadata callMetadata);  }