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

    Cache<?, ?> cache = Caches.eternal();
    {
      // The first side input will populate the cache.
      MultimapSideInput<byte[], String> multimapSideInput =
          new MultimapSideInput<>(
              cache,
              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));
      assertArrayEquals(
          new byte[][] {A, B}, Iterables.toArray(multimapSideInput.get(), byte[].class));
    }

    {
      // The next side input will load all of its contents from the cache.
      MultimapSideInput<byte[], String> multimapSideInput =
          new MultimapSideInput<>(
              cache,
              requestBuilder -> {
                throw new IllegalStateException("Unexpected call for test.");
              },
              "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));
      assertArrayEquals(
          new byte[][] {A, B}, Iterables.toArray(multimapSideInput.get(), byte[].class));
    }
  }