@Override
  public void run(RunJobRequest request, StreamObserver<RunJobResponse> responseObserver) {
    LOG.trace("{} {}", RunJobRequest.class.getSimpleName(), request);

    String preparationId = request.getPreparationId();
    try {
      // retrieve job preparation
      JobPreparation preparation = preparations.get(preparationId);
      if (preparation == null) {
        String errMessage = String.format("Unknown Preparation Id \"%s\".", preparationId);
        StatusException exception = Status.NOT_FOUND.withDescription(errMessage).asException();
        responseObserver.onError(exception);
        return;
      }
      try {
        PipelineValidator.validate(preparation.pipeline());
      } catch (Exception e) {
        LOG.warn("Encountered Unexpected Exception during validation", e);
        responseObserver.onError(new StatusRuntimeException(Status.INVALID_ARGUMENT.withCause(e)));
        return;
      }

      // create new invocation
      JobInvocation invocation =
          invoker.invoke(
              resolveDependencies(preparation.pipeline(), stagingSessionTokens.get(preparationId)),
              preparation.options(),
              request.getRetrievalToken());
      String invocationId = invocation.getId();

      invocation.addStateListener(
          event -> {
            if (!JobInvocation.isTerminated(event.getState())) {
              return;
            }
            String stagingSessionToken = stagingSessionTokens.get(preparationId);
            stagingSessionTokens.remove(preparationId);
            try {
              if (cleanupJobFn != null) {
                cleanupJobFn.accept(stagingSessionToken);
              }
            } catch (Exception e) {
              LOG.warn(
                  "Failed to remove job staging directory for token {}.", stagingSessionToken, e);
            } finally {
              onFinishedInvocationCleanup(invocationId);
            }
          });

      invocation.start();
      invocations.put(invocationId, invocation);
      // Cleanup this preparation because we are running it now.
      // If we fail, we need to prepare again.
      preparations.remove(preparationId);
      RunJobResponse response = RunJobResponse.newBuilder().setJobId(invocationId).build();
      responseObserver.onNext(response);
      responseObserver.onCompleted();
    } catch (StatusRuntimeException e) {
      LOG.warn("Encountered Status Exception", e);
      responseObserver.onError(e);
    } catch (Exception e) {
      String errMessage =
          String.format("Encountered Unexpected Exception for Preparation %s", preparationId);
      LOG.error(errMessage, e);
      responseObserver.onError(Status.INTERNAL.withCause(e).asException());
    }
  }
@Test
  public void testJobSubmissionUsesJobInvokerAndIsSuccess() throws Exception {
    JobApi.PrepareJobResponse prepareResponse = prepareJob();

    // run job
    JobApi.RunJobRequest runRequest =
        JobApi.RunJobRequest.newBuilder()
            .setPreparationId(prepareResponse.getPreparationId())
            .setRetrievalToken(TEST_RETRIEVAL_TOKEN)
            .build();
    RecordingObserver<JobApi.RunJobResponse> runRecorder = new RecordingObserver<>();
    service.run(runRequest, runRecorder);
    verify(invoker, times(1)).invoke(TEST_PIPELINE, TEST_OPTIONS, TEST_RETRIEVAL_TOKEN);
    assertThat(runRecorder.isSuccessful(), is(true));
    assertThat(runRecorder.values, hasSize(1));
    JobApi.RunJobResponse runResponse = runRecorder.values.get(0);
    assertThat(runResponse.getJobId(), is(TEST_JOB_ID));

    verify(invocation, times(1)).addStateListener(any());
    verify(invocation, times(1)).start();
  }