public CompletableFuture<Object> terminationFuture() {
    return onFinish;
  }
@Test
  public void testJavaErrorResponse() throws Exception {
    BlockingQueue<StreamObserver<BeamFnApi.InstructionRequest>> outboundServerObservers =
        new LinkedBlockingQueue<>();
    BlockingQueue<Throwable> error = new LinkedBlockingQueue<>();
    CallStreamObserver<BeamFnApi.InstructionResponse> inboundServerObserver =
        TestStreams.<BeamFnApi.InstructionResponse>withOnNext(
                response -> fail(String.format("Unexpected Response %s", response)))
            .withOnError(error::add)
            .build();

    Endpoints.ApiServiceDescriptor apiServiceDescriptor =
        Endpoints.ApiServiceDescriptor.newBuilder()
            .setUrl(this.getClass().getName() + "-" + UUID.randomUUID().toString())
            .build();
    Server server =
        InProcessServerBuilder.forName(apiServiceDescriptor.getUrl())
            .addService(
                new BeamFnControlGrpc.BeamFnControlImplBase() {
                  @Override
                  public StreamObserver<BeamFnApi.InstructionResponse> control(
                      StreamObserver<BeamFnApi.InstructionRequest> outboundObserver) {
                    Uninterruptibles.putUninterruptibly(outboundServerObservers, outboundObserver);
                    return inboundServerObserver;
                  }
                })
            .build();
    server.start();
    try {
      EnumMap<
              BeamFnApi.InstructionRequest.RequestCase,
              ThrowingFunction<BeamFnApi.InstructionRequest, BeamFnApi.InstructionResponse.Builder>>
          handlers = new EnumMap<>(BeamFnApi.InstructionRequest.RequestCase.class);
      handlers.put(
          BeamFnApi.InstructionRequest.RequestCase.REGISTER,
          value -> {
            assertEquals(value.getInstructionId(), BeamFnLoggingMDC.getInstructionId());
            throw new Error("Test Error");
          });

      ExecutorService executor = Executors.newCachedThreadPool();
      BeamFnControlClient client =
          new BeamFnControlClient(
              apiServiceDescriptor,
              ManagedChannelFactory.createInProcess(),
              OutboundObserverFactory.trivial(),
              executor,
              handlers);

      // Get the connected client and attempt to send and receive an instruction
      StreamObserver<BeamFnApi.InstructionRequest> outboundServerObserver =
          outboundServerObservers.take();

      // Ensure that all exceptions are caught and translated to failures
      outboundServerObserver.onNext(
          InstructionRequest.newBuilder()
              .setInstructionId("0")
              .setRegister(RegisterRequest.getDefaultInstance())
              .build());
      // There should be an error reported to the StreamObserver.
      assertThat(error.take(), not(nullValue()));

      // Ensure that the client shuts down when an Error is thrown from the harness
      try {
        client.terminationFuture().get();
        throw new IllegalStateException("The future should have terminated with an error");
      } catch (ExecutionException errorWrapper) {
        assertThat(errorWrapper.getCause().getMessage(), containsString("Test Error"));
      }
    } finally {
      server.shutdownNow();
    }
  }