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 testCachingOfClient() throws Exception {
    Endpoints.ApiServiceDescriptor otherApiServiceDescriptor =
        Endpoints.ApiServiceDescriptor.newBuilder()
            .setUrl(apiServiceDescriptor.getUrl() + "-other")
            .build();
    Server testServer2 =
        InProcessServerBuilder.forName(otherApiServiceDescriptor.getUrl())
            .addService(
                new BeamFnStateImplBase() {
                  @Override
                  public StreamObserver<StateRequest> state(
                      StreamObserver<StateResponse> outboundObserver) {
                    throw new RuntimeException();
                  }
                })
            .build();
    testServer2.start();

    try {
      assertSame(
          clientCache.forApiServiceDescriptor(apiServiceDescriptor),
          clientCache.forApiServiceDescriptor(apiServiceDescriptor));
      assertNotSame(
          clientCache.forApiServiceDescriptor(apiServiceDescriptor),
          clientCache.forApiServiceDescriptor(otherApiServiceDescriptor));
    } finally {
      testServer2.shutdownNow();
    }
  }