@Override
  public <T extends State> T state(StateNamespace namespace, StateTag<T> address) {
    return workItemState.get(namespace, address, StateContexts.nullContext());
  }
@Test
  public void testMultimapPutAndGet() {
    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>> future = SettableFuture.create();
    when(mockReader.multimapFetchSingleEntryFuture(
            encodeWithCoder(key, ByteArrayCoder.of()),
            key(NAMESPACE, tag),
            STATE_FAMILY,
            VarIntCoder.of()))
        .thenReturn(future);

    multimapState.put(key, 1);
    ReadableState<Iterable<Integer>> result = multimapState.get(dup(key)).readLater();
    waitAndSet(future, Arrays.asList(1, 2, 3), 30);
    assertThat(result.read(), Matchers.containsInAnyOrder(1, 1, 2, 3));

    multimapState.remove(key);
    multimapState.put(key, 4);
    multimapState.remove(key);
    multimapState.put(key, 5);
    assertThat(result.read(), Matchers.containsInAnyOrder(5));
    multimapState.clear();
    assertThat(multimapState.get(key).read(), Matchers.emptyIterable());
  }