@Override
  public StreamObserver<WorkerStatusResponse> workerStatus(
      StreamObserver<WorkerStatusRequest> requestObserver) {
    if (isClosed.get()) {
      throw new IllegalStateException("BeamWorkerStatusGrpcService already closed.");
    }
    String workerId = headerAccessor.getSdkWorkerId();
    LOG.info("Beam Fn Status client connected with id {}", workerId);

    WorkerStatusClient fnApiStatusClient =
        WorkerStatusClient.forRequestObserver(workerId, requestObserver);
    connectedClient.compute(
        workerId,
        (k, existingClientFuture) -> {
          if (existingClientFuture != null) {
            try {
              if (existingClientFuture.isDone()) {
                LOG.info(
                    "SDK Worker {} was connected to status server previously, disconnecting old client",
                    workerId);
                existingClientFuture.get().close();
              } else {
                existingClientFuture.complete(fnApiStatusClient);
                return existingClientFuture;
              }
            } catch (IOException | InterruptedException | ExecutionException e) {
              LOG.warn("Error closing worker status client", e);
            }
          }
          return CompletableFuture.completedFuture(fnApiStatusClient);
        });
    return fnApiStatusClient.getResponseObserver();
  }
@Test
  public void testClientConnected() throws Exception {
    stub.workerStatus(mockObserver);
    WorkerStatusClient client = waitAndGetStatusClient(ID);
    assertNotNull(client);
  }