@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 IsmRecordCoder but got %s",
        coder);
    return new CreateIsmShardKeyAndSortKeyParDoFn((IsmRecordCoder<?>) coder);
  }
@Test
  public void testConversionOfRecord() throws Exception {
    ParDoFn parDoFn =
        new CreateIsmShardKeyAndSortKeyDoFnFactory()
            .create(
                null /* pipeline options */,
                CloudObject.fromSpec(
                    ImmutableMap.of(
                        PropertyNames.OBJECT_TYPE_NAME,
                        "CreateIsmShardKeyAndSortKeyDoFn",
                        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(42, 43)));
    IsmRecordCoder<?> coder =
        (IsmRecordCoder)
            CloudObjects.coderFromCloudObject(CloudObject.fromSpec(createIsmRecordEncoding()));
    assertThat(
        outputReceiver,
        contains(
            valueInGlobalWindow(
                KV.of(
                    coder.hash(ImmutableList.of(42)) /* hash key */,
                    KV.of(KV.of(42, GlobalWindow.INSTANCE) /* sort key */, 43 /* value */)))));
  }