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

    final byte[] key1 = "key1".getBytes(StandardCharsets.UTF_8);
    final byte[] key2 = "key2".getBytes(StandardCharsets.UTF_8);
    SettableFuture<Iterable<Integer>> future1 = SettableFuture.create();
    when(mockReader.multimapFetchSingleEntryFuture(
            encodeWithCoder(key1, ByteArrayCoder.of()),
            key(NAMESPACE, tag),
            STATE_FAMILY,
            VarIntCoder.of()))
        .thenReturn(future1);
    SettableFuture<Iterable<Integer>> future2 = SettableFuture.create();
    when(mockReader.multimapFetchSingleEntryFuture(
            encodeWithCoder(key2, ByteArrayCoder.of()),
            key(NAMESPACE, tag),
            STATE_FAMILY,
            VarIntCoder.of()))
        .thenReturn(future2);

    ReadableState<Integer> result1 = mapViaMultiMapState.get(dup(key1)).readLater();
    ReadableState<Integer> result2 = mapViaMultiMapState.get(dup(key2)).readLater();
    waitAndSet(future1, Collections.singletonList(1), 30);
    waitAndSet(future2, Collections.emptyList(), 1);
    assertEquals(Integer.valueOf(1), result1.read());
    assertNull(result2.read());
  }