FnApiControlClientPoolService extends BeamFnControlGrpc.BeamFnControlImplBase implements FnService { @Override public StreamObserver<BeamFnApi.InstructionResponse> control( StreamObserver<BeamFnApi.InstructionRequest> requestObserver) { final String workerId = headerAccessor.getSdkWorkerId(); if (Strings.isNullOrEmpty(workerId)) { LOG.warn("No worker_id header provided in control request"); } LOG.info("Beam Fn Control client connected with id {}", workerId); FnApiControlClient newClient = FnApiControlClient.forRequestObserver(workerId, requestObserver, processBundleDescriptors); try { synchronized (lock) { checkState( !closed, "%s already closed", FnApiControlClientPoolService.class.getSimpleName()); vendedClients.put(workerId, newClient); } clientSink.put(headerAccessor.getSdkWorkerId(), newClient); } catch (Exception e) { throw new RuntimeException(e); } return newClient.asResponseObserver(); } private  FnApiControlClientPoolService(
      ControlClientPool.Sink clientSink, HeaderAccessor headerAccessor); static FnApiControlClientPoolService offeringClientsToPool(
      ControlClientPool.Sink clientPool, HeaderAccessor headerAccessor); @Override StreamObserver<BeamFnApi.InstructionResponse> control(
      StreamObserver<BeamFnApi.InstructionRequest> requestObserver); @Override void getProcessBundleDescriptor(
      BeamFnApi.GetProcessBundleDescriptorRequest request,
      StreamObserver<BeamFnApi.ProcessBundleDescriptor> responseObserver); @Override void close();  }
@Test public void testIncomingConnection() throws Exception { StreamObserver<BeamFnApi.InstructionRequest> requestObserver = mock(StreamObserver.class); StreamObserver<BeamFnApi.InstructionResponse> responseObserver = controlService.control(requestObserver); InstructionRequestHandler client = pool.getSource().take("", Duration.ofSeconds(2)); String id = "fakeInstruction"; CompletionStage<BeamFnApi.InstructionResponse> responseFuture = client.handle(BeamFnApi.InstructionRequest.newBuilder().setInstructionId(id).build()); verify(requestObserver).onNext(any(BeamFnApi.InstructionRequest.class)); assertThat(MoreFutures.isDone(responseFuture), is(false)); responseObserver.onNext( BeamFnApi.InstructionResponse.newBuilder().setInstructionId(id).build()); MoreFutures.get(responseFuture); }