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

    ReadableState<Iterable<Integer>> result1 = multimapState.get(key1).readLater();
    ReadableState<Iterable<Integer>> result2 = multimapState.get(dup(key2)).readLater();
    multimapState.clear();
    waitAndSet(future, Arrays.asList(1, 2), 30);
    assertThat(result1.read(), Matchers.emptyIterable());
    assertThat(result2.read(), Matchers.emptyIterable());
    assertThat(multimapState.keys().read(), Matchers.emptyIterable());
    assertThat(multimapState.entries().read(), Matchers.emptyIterable());
    assertTrue(multimapState.isEmpty().read());
  }