@Override
  public Iterable<K> get() {
    return StateFetchingIterators.readAllAndDecodeStartingFrom(
        cache, beamFnStateClient, keysRequest, keyCoder);
  }
@Test
  public void testGetWithBulkRead() throws Exception {
    FakeBeamFnStateClient fakeBeamFnStateClient =
        new FakeBeamFnStateClient(
            ImmutableMap.of(
                keysValuesStateKey(),
                KV.of(
                    KvCoder.of(ByteArrayCoder.of(), IterableCoder.of(StringUtf8Coder.of())),
                    asList(KV.of(A, asList("A1", "A2", "A3")), KV.of(B, asList("B1", "B2"))))));

    MultimapSideInput<byte[], String> multimapSideInput =
        new MultimapSideInput<>(
            Caches.noop(),
            fakeBeamFnStateClient,
            "instructionId",
            keysStateKey(),
            ByteArrayCoder.of(),
            StringUtf8Coder.of(),
            true);
    assertArrayEquals(
        new String[] {"A1", "A2", "A3"}, Iterables.toArray(multimapSideInput.get(A), String.class));
    assertArrayEquals(
        new String[] {"B1", "B2"}, Iterables.toArray(multimapSideInput.get(B), String.class));
    assertArrayEquals(
        new String[] {}, Iterables.toArray(multimapSideInput.get(UNKNOWN), String.class));
  }