CreateIsmShardKeyAndSortKeyDoFnFactory 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 IsmRecordCoder but got %s", coder); return new CreateIsmShardKeyAndSortKeyParDoFn((IsmRecordCoder<?>) coder); }  @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 CreateIsmShardKeyAndSortKeyDoFnFactory() .create( null , CloudObject.fromSpec( ImmutableMap.of( PropertyNames.OBJECT_TYPE_NAME, "CreateIsmShardKeyAndSortKeyDoFn", PropertyNames.ENCODING, createIsmRecordEncoding())), null , null , null , null , null ); 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)) , KV.of(KV.of(42, GlobalWindow.INSTANCE) , 43 ))))); }