public CompletableFuture<Object> terminationFuture() {
    return onFinish;
  }
@Test
  public void testDelegation() throws Exception {
    AtomicBoolean clientClosedStream = new AtomicBoolean();
    BlockingQueue<BeamFnApi.InstructionResponse> values = new LinkedBlockingQueue<>();
    BlockingQueue<StreamObserver<BeamFnApi.InstructionRequest>> outboundServerObservers =
        new LinkedBlockingQueue<>();
    CallStreamObserver<BeamFnApi.InstructionResponse> inboundServerObserver =
        TestStreams.withOnNext(values::add)
            .withOnCompleted(() -> clientClosedStream.set(true))
            .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.PROCESS_BUNDLE,
          value -> {
            assertEquals(value.getInstructionId(), BeamFnLoggingMDC.getInstructionId());
            return BeamFnApi.InstructionResponse.newBuilder()
                .setProcessBundle(BeamFnApi.ProcessBundleResponse.getDefaultInstance());
          });
      handlers.put(
          BeamFnApi.InstructionRequest.RequestCase.REGISTER,
          value -> {
            assertEquals(value.getInstructionId(), BeamFnLoggingMDC.getInstructionId());
            throw FAILURE;
          });

      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();

      outboundServerObserver.onNext(SUCCESSFUL_REQUEST);
      assertEquals(SUCCESSFUL_RESPONSE, values.take());

      // Ensure that conversion of an unknown request type is properly converted to a
      // failure response.
      outboundServerObserver.onNext(UNKNOWN_HANDLER_REQUEST);
      assertEquals(UNKNOWN_HANDLER_RESPONSE, values.take());

      // Ensure that all exceptions are caught and translated to failures
      outboundServerObserver.onNext(FAILURE_REQUEST);
      assertEquals(FAILURE_RESPONSE, values.take());

      // Ensure that the server completing the stream translates to the completable future
      // being completed allowing for a successful shutdown of the client.
      outboundServerObserver.onCompleted();
      client.terminationFuture().get();
    } finally {
      server.shutdownNow();
    }
  }