AggregatingJobServiceGateway implements JobServiceGateway { @Override public Observable<String> createJob(JobDescriptor jobDescriptor, CallMetadata callMetadata) { Cell cell = router.routeKey(jobDescriptor).orElse(null); if (cell == null) { return Observable.error(new IllegalStateException("Internal system error. Routing rule not found")); } logger.debug("Routing JobDescriptor {} to Cell {}", jobDescriptor, cell); Optional<JobManagementServiceStub> optionalClient = CellConnectorUtil.toStub(cell, connector, JobManagementServiceGrpc::newStub); if (!optionalClient.isPresent()) { return Observable.error(TitusServiceException.cellNotFound(cell.getName())); } JobManagementServiceStub client = optionalClient.get(); JobDescriptor withStackName = addStackName(jobDescriptor); return createRequestObservable(emitter -> { StreamObserver<JobId> streamObserver = GrpcUtil.createClientResponseObserver( emitter, jobId -> emitter.onNext(jobId.getId()), emitter::onError, emitter::onCompleted ); wrap(client, callMetadata).createJob(withStackName, streamObserver); }, grpcConfiguration.getRequestTimeoutMs()); } @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 createJobInjectsFederatedStackName() { Cell firstCell = cells.get(0); CellWithCachedJobsService cachedJobsService = new CellWithCachedJobsService(firstCell.getName()); cellToServiceMap.get(firstCell).getServiceRegistry().addService(cachedJobsService); JobDescriptor jobDescriptor = JobDescriptor.newBuilder() .setApplicationName("app1") .setCapacityGroup("app1CapGroup") .build(); String jobId = service.createJob(jobDescriptor, JobManagerConstants.UNDEFINED_CALL_METADATA).toBlocking().first(); Optional<JobDescriptor> createdJob = cachedJobsService.getCachedJob(jobId); assertThat(createdJob).isPresent(); assertThat(createdJob.get().getAttributesMap()).containsEntry(JOB_ATTRIBUTES_STACK, stackName); }