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
  public void testRequestResponses() throws Exception {
    BeamFnStateClient client = clientCache.forApiServiceDescriptor(apiServiceDescriptor);

    CompletableFuture<StateResponse> successfulResponse =
        client.handle(StateRequest.newBuilder().setInstructionId(SUCCESS));
    CompletableFuture<StateResponse> unsuccessfulResponse =
        client.handle(StateRequest.newBuilder().setInstructionId(FAIL));

    // Wait for the client to connect.
    StreamObserver<StateResponse> outboundServerObserver = outboundServerObservers.take();
    // Ensure the client doesn't break when sent garbage.
    outboundServerObserver.onNext(StateResponse.newBuilder().setId("UNKNOWN ID").build());

    // We expect to receive and handle two requests
    handleServerRequest(outboundServerObserver, values.take());
    handleServerRequest(outboundServerObserver, values.take());

    // Ensure that the successful and unsuccessful responses were propagated.
    assertNotNull(successfulResponse.get());
    try {
      unsuccessfulResponse.get();
      fail("Expected unsuccessful response");
    } catch (ExecutionException e) {
      assertThat(e.toString(), containsString(TEST_ERROR));
    }
  }