public ForComputation forComputation(String computation) {
    return new ForComputation(computation);
  }
@Test
  public void testExplicitInvalidation() throws Exception {
    WindmillStateCache.ForKeyAndFamily keyCache1 =
        cache
            .forComputation("comp1")
            .forKey(computationKey("comp1", "key1", 1), 0L, 0L)
            .forFamily(STATE_FAMILY);
    WindmillStateCache.ForKeyAndFamily keyCache2 =
        cache
            .forComputation("comp1")
            .forKey(computationKey("comp1", "key2", SHARDING_KEY), 0L, 0L)
            .forFamily(STATE_FAMILY);
    WindmillStateCache.ForKeyAndFamily keyCache3 =
        cache
            .forComputation("comp2")
            .forKey(computationKey("comp2", "key1", SHARDING_KEY), 0L, 0L)
            .forFamily(STATE_FAMILY);
    WindmillStateCache.ForKeyAndFamily keyCache4 =
        cache
            .forComputation("comp1")
            .forKey(computationKey("comp1", "key1", 2), 0L, 0L)
            .forFamily(STATE_FAMILY);

    keyCache1.put(StateNamespaces.global(), new TestStateTag("tag1"), new TestState("g1"), 1);
    keyCache1.persist();
    keyCache2.put(StateNamespaces.global(), new TestStateTag("tag2"), new TestState("g2"), 2);
    keyCache2.persist();
    keyCache3.put(StateNamespaces.global(), new TestStateTag("tag3"), new TestState("g3"), 3);
    keyCache3.persist();
    keyCache4.put(StateNamespaces.global(), new TestStateTag("tag4"), new TestState("g4"), 4);
    keyCache4.persist();
    keyCache1 =
        cache
            .forComputation("comp1")
            .forKey(computationKey("comp1", "key1", 1), 0L, 1L)
            .forFamily(STATE_FAMILY);
    keyCache2 =
        cache
            .forComputation("comp1")
            .forKey(computationKey("comp1", "key2", SHARDING_KEY), 0L, 1L)
            .forFamily(STATE_FAMILY);
    keyCache3 =
        cache
            .forComputation("comp2")
            .forKey(computationKey("comp2", "key1", SHARDING_KEY), 0L, 1L)
            .forFamily(STATE_FAMILY);
    keyCache4 =
        cache
            .forComputation("comp1")
            .forKey(computationKey("comp1", "key1", 2), 0L, 1L)
            .forFamily(STATE_FAMILY);
    assertEquals(
        Optional.of(new TestState("g1")),
        keyCache1.get(StateNamespaces.global(), new TestStateTag("tag1")));
    assertEquals(
        Optional.of(new TestState("g2")),
        keyCache2.get(StateNamespaces.global(), new TestStateTag("tag2")));
    assertEquals(
        Optional.of(new TestState("g3")),
        keyCache3.get(StateNamespaces.global(), new TestStateTag("tag3")));
    assertEquals(
        Optional.of(new TestState("g4")),
        keyCache4.get(StateNamespaces.global(), new TestStateTag("tag4")));

    // Invalidation of key 1 shard 1 does not affect another shard of key 1 or other keys.
    cache.forComputation("comp1").invalidate(ByteString.copyFromUtf8("key1"), 1);
    keyCache1 =
        cache
            .forComputation("comp1")
            .forKey(computationKey("comp1", "key1", 1), 0L, 2L)
            .forFamily(STATE_FAMILY);

    assertEquals(
        Optional.empty(), keyCache1.get(StateNamespaces.global(), new TestStateTag("tag1")));
    assertEquals(
        Optional.of(new TestState("g2")),
        keyCache2.get(StateNamespaces.global(), new TestStateTag("tag2")));
    assertEquals(
        Optional.of(new TestState("g3")),
        keyCache3.get(StateNamespaces.global(), new TestStateTag("tag3")));
    assertEquals(
        Optional.of(new TestState("g4")),
        keyCache4.get(StateNamespaces.global(), new TestStateTag("tag4")));

    // Invalidation of an non-existing key affects nothing.
    cache.forComputation("comp1").invalidate(ByteString.copyFromUtf8("key1"), 3);

    assertEquals(
        Optional.of(new TestState("g2")),
        keyCache2.get(StateNamespaces.global(), new TestStateTag("tag2")));
    assertEquals(
        Optional.of(new TestState("g3")),
        keyCache3.get(StateNamespaces.global(), new TestStateTag("tag3")));
    assertEquals(
        Optional.of(new TestState("g4")),
        keyCache4.get(StateNamespaces.global(), new TestStateTag("tag4")));
  }