class 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()))));
  }

  public  MultimapSideInput(
      BeamFnStateClient beamFnStateClient,
      String instructionId,
      String ptransformId,
      String sideInputId,
      ByteString encodedWindow,
      Coder<K> keyCoder,
      Coder<V> valueCoder);

}