BeamWorkerStatusGrpcService extends BeamFnWorkerStatusImplBase implements FnService { @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(); } private  BeamWorkerStatusGrpcService(
      ApiServiceDescriptor apiServiceDescriptor, HeaderAccessor headerAccessor); static BeamWorkerStatusGrpcService create(
      ApiServiceDescriptor apiServiceDescriptor, HeaderAccessor headerAccessor); @Override void close(); @Override StreamObserver<WorkerStatusResponse> workerStatus(
      StreamObserver<WorkerStatusRequest> requestObserver); String getSingleWorkerStatus(String workerId, long timeout, TimeUnit timeUnit); Map<String, String> getAllWorkerStatuses(long timeout, TimeUnit timeUnit);  }
@Test public void testClientConnected() throws Exception { StreamObserver<WorkerStatusResponse> workerStatusResponseStreamObserver = stub.workerStatus(mockObserver); WorkerStatusClient client = waitAndGetStatusClient(ID); assertNotNull(client); }