ToIsmRecordForMultimapDoFnFactory implements ParDoFnFactory { @Override public ParDoFn create( PipelineOptions options, CloudObject cloudUserFn, List<SideInputInfo> sideInputInfos, TupleTag<?> mainOutputTag, Map<TupleTag<?>, Integer> outputTupleTagsToReceiverIndices, DataflowExecutionContext<?> executionContext, DataflowOperationContext operationContext) throws Exception { Coder<?> coder = CloudObjects.coderFromCloudObject( CloudObject.fromSpec(Structs.getObject(cloudUserFn, PropertyNames.ENCODING))); checkState( coder instanceof IsmRecordCoder, "Expected to received an instanceof an %s but got %s", IsmRecordCoder.class.getSimpleName(), coder); IsmRecordCoder<?> ismRecordCoder = (IsmRecordCoder<?>) coder; return new ToIsmRecordForMultimapParDoFn( KvCoder.of( ismRecordCoder.getCoderArguments().get(0), ismRecordCoder.getCoderArguments().get(1))); }  @Override ParDoFn create(
      PipelineOptions options,
      CloudObject cloudUserFn,
      List<SideInputInfo> sideInputInfos,
      TupleTag<?> mainOutputTag,
      Map<TupleTag<?>, Integer> outputTupleTagsToReceiverIndices,
      DataflowExecutionContext<?> executionContext,
      DataflowOperationContext operationContext);  }
@Test public void testConversionOfRecord() throws Exception { ParDoFn parDoFn = new ToIsmRecordForMultimapDoFnFactory() .create( null , CloudObject.fromSpec( ImmutableMap.of( PropertyNames.OBJECT_TYPE_NAME, "ToIsmRecordForMultimapDoFn", PropertyNames.ENCODING, createIsmRecordEncoding())), null , null , null , null , null ); List<Object> outputReceiver = new ArrayList<>(); parDoFn.startBundle(outputReceiver::add); parDoFn.processElement( valueInGlobalWindow( KV.of( 12 , ImmutableList.of( KV.of( KV.of(42 , GlobalWindow.INSTANCE ) , 4 ), KV.of( KV.of(42 , GlobalWindow.INSTANCE ) , 5 ), KV.of( KV.of(43 , GlobalWindow.INSTANCE ) , 6 ), KV.of( KV.of(44 , GlobalWindow.INSTANCE ) , 7 ), KV.of( KV.of(44 , GlobalWindow.INSTANCE ) , 8 ))))); assertThat( outputReceiver, contains( valueInGlobalWindow(IsmRecord.of(ImmutableList.of(42, GlobalWindow.INSTANCE, 0L), 4)), valueInGlobalWindow(IsmRecord.of(ImmutableList.of(42, GlobalWindow.INSTANCE, 1L), 5)), valueInGlobalWindow(IsmRecord.of(ImmutableList.of(43, GlobalWindow.INSTANCE, 0L), 6)), valueInGlobalWindow(IsmRecord.of(ImmutableList.of(44, GlobalWindow.INSTANCE, 0L), 7)), valueInGlobalWindow(IsmRecord.of(ImmutableList.of(44, GlobalWindow.INSTANCE, 1L), 8)))); }