FnApiControlClient implements Closeable, InstructionRequestHandler { @Override public CompletionStage<BeamFnApi.InstructionResponse> handle( BeamFnApi.InstructionRequest request) { LOG.debug("Sending InstructionRequest {}", request); CompletableFuture<BeamFnApi.InstructionResponse> resultFuture = new CompletableFuture<>(); outstandingRequests.put(request.getInstructionId(), resultFuture); requestReceiver.onNext(request); return resultFuture; } private  FnApiControlClient(
      String workerId,
      StreamObserver<InstructionRequest> requestReceiver,
      ConcurrentMap<String, BeamFnApi.ProcessBundleDescriptor> processBundleDescriptors); static FnApiControlClient forRequestObserver(
      String workerId,
      StreamObserver<BeamFnApi.InstructionRequest> requestObserver,
      ConcurrentMap<String, BeamFnApi.ProcessBundleDescriptor> processBundleDescriptors); @Override CompletionStage<BeamFnApi.InstructionResponse> handle(
      BeamFnApi.InstructionRequest request); StreamObserver<BeamFnApi.InstructionResponse> asResponseObserver(); BeamFnApi.ProcessBundleDescriptor getProcessBundleDescriptor(String id); @Override void registerProcessBundleDescriptor(
      BeamFnApi.ProcessBundleDescriptor processBundleDescriptor); @Override void close(); String getWorkerId(); void onClose(Consumer<FnApiControlClient> onCloseListener);  }
@Test public void testRequestSent() { String id = "instructionId"; client.handle(BeamFnApi.InstructionRequest.newBuilder().setInstructionId(id).build()); verify(mockObserver).onNext(any(BeamFnApi.InstructionRequest.class)); }