AggregatingJobServiceGateway implements JobServiceGateway { @Override public Observable<Task> findTask(String taskId, CallMetadata callMetadata) { return findTaskInAllCells(taskId, callMetadata).map(CellResponse::getResult).map(this::addStackName); } @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);  }
@Test public void findTask() { Random random = new Random(); List<Task> cellOneSnapshot = new ArrayList<>(dataGenerator.newServiceJobWithTasks()); cellOne.getServiceRegistry().addService(new CellWithFixedTasksService(cellOneSnapshot)); cellTwo.getServiceRegistry().addService(new CellWithFixedTasksService(Collections.emptyList())); Task expected = withStackName(cellOneSnapshot.get(random.nextInt(cellOneSnapshot.size()))); AssertableSubscriber<Task> testSubscriber = service.findTask(expected.getId(), UNDEFINED_CALL_METADATA).test(); testSubscriber.awaitTerminalEvent(1, TimeUnit.SECONDS); testSubscriber.assertNoErrors(); testSubscriber.assertValueCount(1); testSubscriber.assertValue(expected); }
@Test public void findTaskWithFailingCell() { Random random = new Random(); List<Task> cellOneSnapshot = new ArrayList<>(dataGenerator.newServiceJobWithTasks()); cellOne.getServiceRegistry().addService(new CellWithFixedTasksService(cellOneSnapshot)); cellTwo.getServiceRegistry().addService(new CellWithFailingJobManagementService(DEADLINE_EXCEEDED)); Task expected = withStackName(cellOneSnapshot.get(random.nextInt(cellOneSnapshot.size()))); AssertableSubscriber<Task> testSubscriber = service.findTask(expected.getId(), UNDEFINED_CALL_METADATA).test(); testSubscriber.awaitTerminalEvent(1, TimeUnit.SECONDS); testSubscriber.assertNoErrors(); testSubscriber.assertValueCount(1); testSubscriber.assertValue(expected); }
@Test public void findTaskErrors() { cellOne.getServiceRegistry().addService(new CellWithFailingJobManagementService(INTERNAL)); cellTwo.getServiceRegistry().addService(new CellWithFailingJobManagementService(UNAVAILABLE)); AssertableSubscriber<Task> testSubscriber = service.findTask("any", UNDEFINED_CALL_METADATA).test(); testSubscriber.awaitTerminalEvent(1, TimeUnit.SECONDS); testSubscriber.assertError(StatusRuntimeException.class); assertThat(Status.fromThrowable(testSubscriber.getOnErrorEvents().get(0))).isEqualTo(INTERNAL); }