@Override
  public <T extends State> T state(StateNamespace namespace, StateTag<T> address) {
    return workItemState.get(namespace, address, StateContexts.nullContext());
  }
@Test
  public void testMultimapModifyAfterReadDoesNotAffectResult() {
    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[] key1 = "key1".getBytes(StandardCharsets.UTF_8);
    final byte[] key2 = "key2".getBytes(StandardCharsets.UTF_8);
    final byte[] key3 = "key3".getBytes(StandardCharsets.UTF_8);
    final byte[] key4 = "key4".getBytes(StandardCharsets.UTF_8);

    SettableFuture<Iterable<Map.Entry<ByteString, Iterable<Integer>>>> entriesFuture =
        SettableFuture.create();
    when(mockReader.multimapFetchAllFuture(
            false, key(NAMESPACE, tag), STATE_FAMILY, VarIntCoder.of()))
        .thenReturn(entriesFuture);
    SettableFuture<Iterable<Map.Entry<ByteString, Iterable<Integer>>>> keysFuture =
        SettableFuture.create();
    when(mockReader.multimapFetchAllFuture(
            true, key(NAMESPACE, tag), STATE_FAMILY, VarIntCoder.of()))
        .thenReturn(keysFuture);
    SettableFuture<Iterable<Integer>> getKey1Future = SettableFuture.create();
    SettableFuture<Iterable<Integer>> getKey2Future = SettableFuture.create();
    SettableFuture<Iterable<Integer>> getKey4Future = SettableFuture.create();
    when(mockReader.multimapFetchSingleEntryFuture(
            encodeWithCoder(key1, ByteArrayCoder.of()),
            key(NAMESPACE, tag),
            STATE_FAMILY,
            VarIntCoder.of()))
        .thenReturn(getKey1Future);
    when(mockReader.multimapFetchSingleEntryFuture(
            encodeWithCoder(key2, ByteArrayCoder.of()),
            key(NAMESPACE, tag),
            STATE_FAMILY,
            VarIntCoder.of()))
        .thenReturn(getKey2Future);
    when(mockReader.multimapFetchSingleEntryFuture(
            encodeWithCoder(key4, ByteArrayCoder.of()),
            key(NAMESPACE, tag),
            STATE_FAMILY,
            VarIntCoder.of()))
        .thenReturn(getKey4Future);

    ReadableState<Iterable<Map.Entry<byte[], Integer>>> entriesResult =
        multimapState.entries().readLater();
    ReadableState<Iterable<byte[]>> keysResult = multimapState.keys().readLater();
    waitAndSet(
        entriesFuture,
        Arrays.asList(multimapEntry(key1, 1, 2, 3), multimapEntry(key2, 2, 3, 4)),
        200);
    waitAndSet(keysFuture, Arrays.asList(multimapEntry(key1), multimapEntry(key2)), 200);

    // make key4 to be known nonexistent.
    multimapState.remove(key4);

    ReadableState<Iterable<Integer>> key1Future = multimapState.get(key1).readLater();
    waitAndSet(getKey1Future, Arrays.asList(1, 2, 3), 200);
    ReadableState<Iterable<Integer>> key2Future = multimapState.get(key2).readLater();
    waitAndSet(getKey2Future, Arrays.asList(2, 3, 4), 200);
    ReadableState<Iterable<Integer>> key4Future = multimapState.get(key4).readLater();
    waitAndSet(getKey4Future, Collections.emptyList(), 200);

    multimapState.put(key1, 7);
    multimapState.put(dup(key2), 8);
    multimapState.put(dup(key3), 8);

    Iterable<Map.Entry<byte[], Integer>> entries = entriesResult.read();
    Iterable<byte[]> keys = keysResult.read();
    Iterable<Integer> key1Values = key1Future.read();
    Iterable<Integer> key2Values = key2Future.read();
    Iterable<Integer> key4Values = key4Future.read();

    // values added/removed after read should not be reflected in result
    multimapState.remove(key1);
    multimapState.put(key2, 9);
    multimapState.put(key4, 10);

    assertEquals(9, Iterables.size(entries));
    assertThat(
        entries,
        Matchers.containsInAnyOrder(
            multimapEntryMatcher(key1, 1),
            multimapEntryMatcher(key1, 2),
            multimapEntryMatcher(key1, 3),
            multimapEntryMatcher(key1, 7),
            multimapEntryMatcher(key2, 4),
            multimapEntryMatcher(key2, 2),
            multimapEntryMatcher(key2, 3),
            multimapEntryMatcher(key2, 8),
            multimapEntryMatcher(key3, 8)));

    assertEquals(3, Iterables.size(keys));
    assertThat(keys, Matchers.containsInAnyOrder(key1, key2, key3));

    assertEquals(4, Iterables.size(key1Values));
    assertThat(key1Values, Matchers.containsInAnyOrder(1, 2, 3, 7));

    assertEquals(4, Iterables.size(key2Values));
    assertThat(key2Values, Matchers.containsInAnyOrder(2, 3, 4, 8));

    assertTrue(Iterables.isEmpty(key4Values));
  }