InMemoryJobService extends JobServiceGrpc.JobServiceImplBase implements FnService { @Override public void prepare( PrepareJobRequest request, StreamObserver<PrepareJobResponse> responseObserver) { try { LOG.trace("{} {}", PrepareJobRequest.class.getSimpleName(), request); String preparationId = String.format("%s_%s", request.getJobName(), UUID.randomUUID().toString()); Struct pipelineOptions = request.getPipelineOptions(); if (pipelineOptions == null) { throw new NullPointerException("Encountered null pipeline options."); } LOG.trace("PIPELINE OPTIONS {} {}", pipelineOptions.getClass(), pipelineOptions); JobPreparation preparation = JobPreparation.builder() .setId(preparationId) .setPipeline(request.getPipeline()) .setOptions(pipelineOptions) .build(); JobPreparation previous = preparations.putIfAbsent(preparationId, preparation); if (previous != null) { String errMessage = String.format("A job with the preparation ID \"%s\" already exists.", preparationId); StatusException exception = Status.NOT_FOUND.withDescription(errMessage).asException(); responseObserver.onError(exception); return; } String stagingSessionToken = stagingServiceTokenProvider.apply(preparationId); stagingSessionTokens.putIfAbsent(preparationId, stagingSessionToken); stagingService .getService() .registerJob( stagingSessionToken, Maps.transformValues( request.getPipeline().getComponents().getEnvironmentsMap(), RunnerApi.Environment::getDependenciesList)); PrepareJobResponse response = PrepareJobResponse.newBuilder() .setPreparationId(preparationId) .setArtifactStagingEndpoint(stagingServiceDescriptor) .setStagingSessionToken(stagingSessionToken) .build(); responseObserver.onNext(response); responseObserver.onCompleted(); } catch (Exception e) { LOG.error("Could not prepare job with name {}", request.getJobName(), e); responseObserver.onError(Status.INTERNAL.withCause(e).asException()); } } private  InMemoryJobService(
      GrpcFnServer<ArtifactStagingService> stagingService,
      Function<String, String> stagingServiceTokenProvider,
      ThrowingConsumer<Exception, String> cleanupJobFn,
      JobInvoker invoker,
      int maxInvocationHistory); static InMemoryJobService create(
      GrpcFnServer<ArtifactStagingService> stagingService,
      Function<String, String> stagingServiceTokenProvider,
      ThrowingConsumer<Exception, String> cleanupJobFn,
      JobInvoker invoker); static InMemoryJobService create(
      GrpcFnServer<ArtifactStagingService> stagingService,
      Function<String, String> stagingServiceTokenProvider,
      ThrowingConsumer<Exception, String> cleanupJobFn,
      JobInvoker invoker,
      int maxInvocationHistory); @Override void prepare(
      PrepareJobRequest request, StreamObserver<PrepareJobResponse> responseObserver); @Override void run(RunJobRequest request, StreamObserver<RunJobResponse> responseObserver); @Override void getJobs(GetJobsRequest request, StreamObserver<GetJobsResponse> responseObserver); @Override void getState(GetJobStateRequest request, StreamObserver<JobStateEvent> responseObserver); @Override void getPipeline(
      GetJobPipelineRequest request, StreamObserver<GetJobPipelineResponse> responseObserver); @Override void cancel(CancelJobRequest request, StreamObserver<CancelJobResponse> responseObserver); @Override void getStateStream(
      GetJobStateRequest request, StreamObserver<JobStateEvent> responseObserver); @Override void getMessageStream(
      JobMessagesRequest request, StreamObserver<JobMessagesResponse> responseObserver); @Override void getJobMetrics(
      JobApi.GetJobMetricsRequest request,
      StreamObserver<JobApi.GetJobMetricsResponse> responseObserver); @Override void describePipelineOptions(
      DescribePipelineOptionsRequest request,
      StreamObserver<DescribePipelineOptionsResponse> responseObserver); @Override void close(); static final int DEFAULT_MAX_INVOCATION_HISTORY; }
@Test public void testPrepareIsSuccessful() { JobApi.PrepareJobRequest request = JobApi.PrepareJobRequest.newBuilder() .setJobName(TEST_JOB_NAME) .setPipeline(RunnerApi.Pipeline.getDefaultInstance()) .setPipelineOptions(Struct.getDefaultInstance()) .build(); RecordingObserver<JobApi.PrepareJobResponse> recorder = new RecordingObserver<>(); service.prepare(request, recorder); assertThat(recorder.isSuccessful(), is(true)); assertThat(recorder.values, hasSize(1)); JobApi.PrepareJobResponse response = recorder.values.get(0); assertThat(response.getArtifactStagingEndpoint(), notNullValue()); assertThat(response.getPreparationId(), notNullValue()); }