@Override
  public <T extends State> T state(StateNamespace namespace, StateTag<T> address) {
    return workItemState.get(namespace, address, StateContexts.nullContext());
  }
@Test
  public void testMultimapCachedSingleEntry() {
    final String tag = "multimap";
    StateTag<MultimapState<byte[], Integer>> addr =
        StateTags.multimap(tag, ByteArrayCoder.of(), VarIntCoder.of());
    MultimapState<byte[], Integer> multimapState = underTest.state(NAMESPACE, addr);

    final byte[] key = "key".getBytes(StandardCharsets.UTF_8);

    SettableFuture<Iterable<Integer>> entryFuture = SettableFuture.create();
    when(mockReader.multimapFetchSingleEntryFuture(
            encodeWithCoder(key, ByteArrayCoder.of()),
            key(NAMESPACE, tag),
            STATE_FAMILY,
            VarIntCoder.of()))
        .thenReturn(entryFuture);

    // to set up the entry key as cache complete and add some local changes
    waitAndSet(entryFuture, weightedList(1, 2, 3), 30);
    multimapState.get(key).read();
    multimapState.put(key, 2);

    when(mockReader.multimapFetchSingleEntryFuture(
            eq(encodeWithCoder(key, ByteArrayCoder.of())),
            eq(key(NAMESPACE, tag)),
            eq(STATE_FAMILY),
            eq(VarIntCoder.of())))
        .thenThrow(
            new RuntimeException(
                "The multimap is cache complete for "
                    + Arrays.toString(key)
                    + " and should not perform any windmill read."));

    Iterable<Integer> values = multimapState.get(dup(key)).read();
    assertThat(values, Matchers.containsInAnyOrder(1, 2, 2, 3));
    assertTrue(multimapState.containsKey(key).read());
  }