public synchronized BeamFnStateClient forApiServiceDescriptor(
      ApiServiceDescriptor apiServiceDescriptor) throws IOException {
    // We specifically are synchronized so that we only create one GrpcStateClient at a time
    // preventing a race where multiple GrpcStateClient objects might be constructed at the same
    // for the same ApiServiceDescriptor.
    BeamFnStateClient rval;
    synchronized (cache) {
      rval = cache.get(apiServiceDescriptor);
    }
    if (rval == null) {
      // We can't be synchronized on cache while constructing the GrpcStateClient since if the
      // connection fails, onError may be invoked from the gRPC thread which will invoke
      // closeAndCleanUp that clears the cache.
      rval = new GrpcStateClient(apiServiceDescriptor);
      synchronized (cache) {
        cache.put(apiServiceDescriptor, rval);
      }
    }
    return rval;
  }
@Test
  // The checker erroneously flags that the CompletableFuture is not being resolved since it is the
  // result to Executor#submit.
  @SuppressWarnings("FutureReturnValueIgnored")
  public void testServerErrorCausesPendingAndFutureCallsToFail() throws Exception {
    BeamFnStateClient client = clientCache.forApiServiceDescriptor(apiServiceDescriptor);

    Future<CompletableFuture<StateResponse>> stateResponse =
        executor.submit(() -> client.handle(StateRequest.newBuilder().setInstructionId(SUCCESS)));
    Future<Void> serverResponse =
        executor.submit(
            () -> {
              // Wait for the client to connect.
              StreamObserver<StateResponse> outboundServerObserver = outboundServerObservers.take();
              // Send an error from the server.
              outboundServerObserver.onError(
                  new StatusRuntimeException(Status.INTERNAL.withDescription(SERVER_ERROR)));
              return null;
            });

    CompletableFuture<StateResponse> inflight = stateResponse.get();
    serverResponse.get();
    try {
      inflight.get();
      fail("Expected unsuccessful response due to server error");
    } catch (ExecutionException e) {
      assertThat(e.toString(), containsString(SERVER_ERROR));
    }
  }