ReaderCache(Duration cacheDuration, Executor invalidationExecutor) {
    this.invalidationExecutor = invalidationExecutor;
    this.cache =
        CacheBuilder.newBuilder()
            .expireAfterWrite(cacheDuration.getMillis(), TimeUnit.MILLISECONDS)
            .removalListener(
                (RemovalNotification<WindmillComputationKey, CacheEntry> notification) -> {
                  if (notification.getCause() != RemovalCause.EXPLICIT) {
                    LOG.info(
                        "Asynchronously closing reader for {} as it has been idle for over {}",
                        notification.getKey(),
                        cacheDuration);
                    asyncCloseReader(notification.getKey(), notification.getValue());
                  }
                })
            .build();
  }
@Test
  public void testReaderCache() throws IOException {
    // Test basic caching expectations

    readerCache.cacheReader(
        WindmillComputationKey.create(C_ID, KEY_1, SHARDING_KEY), 1, 0, reader1);
    readerCache.cacheReader(
        WindmillComputationKey.create(C_ID, KEY_2, SHARDING_KEY), 2, 0, reader2);

    assertEquals(
        reader1,
        readerCache.acquireReader(WindmillComputationKey.create(C_ID, KEY_1, SHARDING_KEY), 1, 1));
    assertNull(
        readerCache.acquireReader(
            WindmillComputationKey.create(C_ID_1, KEY_1, SHARDING_KEY), 1, 1));
    assertNull(
        readerCache.acquireReader(
            WindmillComputationKey.create(C_ID, KEY_1, SHARDING_KEY_1), 1, 1));

    // Trying to override existing reader should throw
    try {
      readerCache.cacheReader(
          WindmillComputationKey.create(C_ID, KEY_2, SHARDING_KEY), 2, 2, reader1);
      fail("Exception should have been thrown");
    } catch (RuntimeException expected) {
      // expected
    }

    // And it should not have overwritten the old value
    assertEquals(
        reader2,
        readerCache.acquireReader(WindmillComputationKey.create(C_ID, KEY_2, SHARDING_KEY), 2, 3));

    assertNull(
        "acquireReader(WindmillComputationKey.create() should remove matching entry",
        readerCache.acquireReader(WindmillComputationKey.create(C_ID, KEY_2, SHARDING_KEY), 2, 4));

    // Make sure computationId is part of the cache key
    readerCache.cacheReader(
        WindmillComputationKey.create(C_ID_1, KEY_1, SHARDING_KEY), 1, 5, reader2);
    assertEquals(
        reader2,
        readerCache.acquireReader(
            WindmillComputationKey.create(C_ID_1, KEY_1, SHARDING_KEY), 1, 6));

    // Make sure sharding key is part of the cache key
    readerCache.cacheReader(
        WindmillComputationKey.create(C_ID, KEY_1, SHARDING_KEY_1), 1, 0, reader3);
    assertEquals(
        reader3,
        readerCache.acquireReader(
            WindmillComputationKey.create(C_ID, KEY_1, SHARDING_KEY_1), 1, 1));
  }