class 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)) {
      // TODO(BEAM-4149): Enforce proper worker id.
      LOGGER.warn("No worker_id header provided in control request");
    }

    LOGGER.info("Beam Fn Control client connected with id {}", workerId);
    FnApiControlClient newClient = FnApiControlClient.forRequestObserver(workerId, requestObserver);
    try {
      // Add the client to the pool of vended clients before making it available - we should close
      // the client when we close even if no one has picked it up yet. This can occur after the
      // service is closed, in which case the client will be discarded when the service is
      // discarded, which should be performed by a call to #shutdownNow. The remote caller must be
      // able to handle an unexpectedly terminated connection.
      synchronized (lock) {
        checkState(
            !closed, "%s already closed", FnApiControlClientPoolService.class.getSimpleName());
        // TODO: https://issues.apache.org/jira/browse/BEAM-4151: Prevent stale client references
        // from leaking.
        vendedClients.add(newClient);
      }
      // We do not attempt to transactionally add the client to our internal list and offer it to
      // the sink.
      clientSink.put(headerAccessor.getSdkWorkerId(), newClient);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
    return newClient.asResponseObserver();
  }

  private  FnApiControlClientPoolService(
      ControlClientPool.Sink clientSink, HeaderAccessor headerAccessor);

}