MultimapSideInput implements MultimapView<K, V> { @Override public Iterable<K> get() { StateRequest.Builder requestBuilder = StateRequest.newBuilder(); requestBuilder .setInstructionId(instructionId) .getStateKeyBuilder() .getMultimapKeysSideInputBuilder() .setTransformId(ptransformId) .setSideInputId(sideInputId) .setWindow(encodedWindow); return new LazyCachingIteratorToIterable<>( new DataStreams.DataStreamDecoder( keyCoder, DataStreams.inbound( StateFetchingIterators.readAllStartingFrom( beamFnStateClient, requestBuilder.build())))); } MultimapSideInput(
      BeamFnStateClient beamFnStateClient,
      String instructionId,
      String ptransformId,
      String sideInputId,
      ByteString encodedWindow,
      Coder<K> keyCoder,
      Coder<V> valueCoder); @Override Iterable<K> get(); @Override Iterable<V> get(K k);  }
@Test public void testGet() throws Exception { FakeBeamFnStateClient fakeBeamFnStateClient = new FakeBeamFnStateClient( ImmutableMap.of( key("A"), encode("A1", "A2", "A3"), key("B"), encode("B1", "B2"))); MultimapSideInput<String, String> multimapSideInput = new MultimapSideInput<>( fakeBeamFnStateClient, "instructionId", "ptransformId", "sideInputId", ByteString.copyFromUtf8("encodedWindow"), StringUtf8Coder.of(), StringUtf8Coder.of()); 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)); }