class BeamFnControlClient {

  public void processInstructionRequests(Executor executor)
      throws InterruptedException, ExecutionException {
    BeamFnApi.InstructionRequest request;
    while (!Objects.equals((request = bufferedInstructions.take()), POISON_PILL)) {
      BeamFnApi.InstructionRequest currentRequest = request;
      executor.execute(
          () -> {
            try {
              BeamFnApi.InstructionResponse response =
                  delegateOnInstructionRequestType(currentRequest);
              sendInstructionResponse(response);
            } catch (Error e) {
              sendErrorResponse(e);
              throw e;
            }
          });
    }
    onFinish.get();
  }

  public  BeamFnControlClient(
      String id,
      ApiServiceDescriptor apiServiceDescriptor,
      ManagedChannelFactory channelFactory,
      OutboundObserverFactory outboundObserverFactory,
      EnumMap<
              BeamFnApi.InstructionRequest.RequestCase,
              ThrowingFunction<BeamFnApi.InstructionRequest, BeamFnApi.InstructionResponse.Builder>>
          handlers);

}