@Test public void observeJobDoesNotSetDeadlines() throws InterruptedException { String cellOneJobId = UUID.randomUUID().toString(); String cellTwoJobId = UUID.randomUUID().toString(); cellOne.getServiceRegistry().addService(new CellWithJobStream(cellOneJobId, cellOneUpdates.serialize())); cellTwo.getServiceRegistry().addService(new CellWithJobStream(cellTwoJobId, cellTwoUpdates.serialize())); List<AssertableSubscriber<JobChangeNotification>> subscribers = new LinkedList<>(); subscribers.add(service.observeJob(cellOneJobId, UNDEFINED_CALL_METADATA).test()); subscribers.add(service.observeJob(cellTwoJobId, UNDEFINED_CALL_METADATA).test()); Thread.sleep(2 * GRPC_REQUEST_TIMEOUT_MS); for (AssertableSubscriber<JobChangeNotification> subscriber : subscribers) { subscriber.assertNoErrors(); subscriber.assertNotCompleted(); assertThat(subscriber.isUnsubscribed()).isFalse(); } }
@Override public Observable<JobChangeNotification> observeJob(String jobId, CallMetadata callMetadata) { JobId request = JobId.newBuilder().setId(jobId).build(); return jobManagementServiceHelper.findJobInAllCells(jobId, callMetadata) .flatMap(response -> singleCellCallWithNoDeadline(response.getCell(), (client, streamObserver) -> client.observeJob(request, streamObserver), callMetadata) ); }
AggregatingJobServiceGateway implements JobServiceGateway { @Override public Observable<JobChangeNotification> observeJob(String jobId, CallMetadata callMetadata) { JobId request = JobId.newBuilder().setId(jobId).build(); return jobManagementServiceHelper.findJobInAllCells(jobId, callMetadata) .flatMap(response -> singleCellCallWithNoDeadline(response.getCell(), (client, streamObserver) -> client.observeJob(request, streamObserver), callMetadata) ); } }
AggregatingJobServiceGateway implements JobServiceGateway { @Override public Observable<JobChangeNotification> observeJob(String jobId, CallMetadata callMetadata) { JobId request = JobId.newBuilder().setId(jobId).build(); return jobManagementServiceHelper.findJobInAllCells(jobId, callMetadata) .flatMap(response -> singleCellCallWithNoDeadline(response.getCell(), (client, streamObserver) -> client.observeJob(request, streamObserver), callMetadata) ); } @Inject  AggregatingJobServiceGateway(GrpcConfiguration grpcConfiguration,
                                        TitusFederationConfiguration federationConfiguration,
                                        CellConnector connector,
                                        CellRouter router,
                                        AggregatingCellClient aggregatingClient,
                                        AggregatingJobManagementServiceHelper jobManagementServiceHelper); }
AggregatingJobServiceGateway implements JobServiceGateway { @Override public Observable<JobChangeNotification> observeJob(String jobId, CallMetadata callMetadata) { JobId request = JobId.newBuilder().setId(jobId).build(); return jobManagementServiceHelper.findJobInAllCells(jobId, callMetadata) .flatMap(response -> singleCellCallWithNoDeadline(response.getCell(), (client, streamObserver) -> client.observeJob(request, streamObserver), callMetadata) ); } @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<JobChangeNotification> observeJob(String jobId, CallMetadata callMetadata) { JobId request = JobId.newBuilder().setId(jobId).build(); return jobManagementServiceHelper.findJobInAllCells(jobId, callMetadata) .flatMap(response -> singleCellCallWithNoDeadline(response.getCell(), (client, streamObserver) -> client.observeJob(request, streamObserver), callMetadata) ); } @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);  }