@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)));
  }
@Test
  public void testConversionOfRecord() throws Exception {
    ParDoFn parDoFn =
        new ToIsmRecordForMultimapDoFnFactory()
            .create(
                null /* pipeline options */,
                CloudObject.fromSpec(
                    ImmutableMap.of(
                        PropertyNames.OBJECT_TYPE_NAME,
                        "ToIsmRecordForMultimapDoFn",
                        PropertyNames.ENCODING,
                        createIsmRecordEncoding())),
                null /* side input infos */,
                null /* main output tag */,
                null /* output tag to receiver index */,
                null /* exection context */,
                null /* operation context */);

    List<Object> outputReceiver = new ArrayList<>();
    parDoFn.startBundle(outputReceiver::add);
    parDoFn.processElement(
        valueInGlobalWindow(
            KV.of(
                12 /* shard key */,
                ImmutableList.of(
                    KV.of(
                        KV.of(42 /* user key */, GlobalWindow.INSTANCE /* window */) /* sort key */,
                        4 /* value */),
                    KV.of(
                        KV.of(42 /* user key */, GlobalWindow.INSTANCE /* window */) /* sort key */,
                        5 /* value */),
                    KV.of(
                        KV.of(43 /* user key */, GlobalWindow.INSTANCE /* window */) /* sort key */,
                        6 /* value */),
                    KV.of(
                        KV.of(44 /* user key */, GlobalWindow.INSTANCE /* window */) /* sort key */,
                        7 /* value */),
                    KV.of(
                        KV.of(44 /* user key */, GlobalWindow.INSTANCE /* window */) /* sort key */,
                        8 /* value */)))));
    assertThat(
        outputReceiver,
        contains(
            valueInGlobalWindow(IsmRecord.of(ImmutableList.of(42, GlobalWindow.INSTANCE, 0L), 4)),
            /* same structural value  as above */
            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)),
            /* same structural value as above and final value */
            valueInGlobalWindow(IsmRecord.of(ImmutableList.of(44, GlobalWindow.INSTANCE, 1L), 8))));
  }