LengthPrefixUnknownCoders { public static List<SideInputInfo> forSideInputInfos( List<SideInputInfo> sideInputInfos, boolean replaceWithByteArrayCoder) { ImmutableList.Builder<SideInputInfo> updatedSideInputInfos = ImmutableList.builder(); for (SideInputInfo sideInputInfo : sideInputInfos) { try { SideInputInfo updatedSideInputInfo = clone(sideInputInfo, SideInputInfo.class); for (Source source : updatedSideInputInfo.getSources()) { source.setCodec(forCodec(source.getCodec(), replaceWithByteArrayCoder)); } updatedSideInputInfos.add(updatedSideInputInfo); } catch (IOException e) { throw new RuntimeException( String.format( "Failed to replace unknown coder with " + "LengthPrefixCoder for : {%s}", sideInputInfo), e); } } return updatedSideInputInfos.build(); }  static MutableNetwork<Node, Edge> andReplaceForRunnerNetwork(
      MutableNetwork<Node, Edge> network); static MutableNetwork<Node, Edge> forSdkNetwork(MutableNetwork<Node, Edge> network); static List<SideInputInfo> forSideInputInfos(
      List<SideInputInfo> sideInputInfos, boolean replaceWithByteArrayCoder);  }
@Test public void testLengthPrefixForSideInputInfos() { List<SideInputInfo> prefixedSideInputInfos = LengthPrefixUnknownCoders.forSideInputInfos( ImmutableList.of( createSideInputInfosWithCoders(windowedValueCoder, prefixedWindowedValueCoder)), false); assertEqualsAsJson( ImmutableList.of( createSideInputInfosWithCoders(prefixedWindowedValueCoder, prefixedWindowedValueCoder)), prefixedSideInputInfos); List<SideInputInfo> prefixedAndReplacedSideInputInfos = LengthPrefixUnknownCoders.forSideInputInfos( ImmutableList.of( createSideInputInfosWithCoders(windowedValueCoder, prefixedWindowedValueCoder)), true); assertEqualsAsJson( ImmutableList.of( createSideInputInfosWithCoders( prefixedAndReplacedWindowedValueCoder, prefixedAndReplacedWindowedValueCoder)), prefixedAndReplacedSideInputInfos); }