PairWithConstantKeyDoFnFactory 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))); Object key = CoderUtils.decodeFromByteArray( coder, Structs.getBytes(cloudUserFn, WorkerPropertyNames.ENCODED_KEY)); return new PairWithConstantKeyParDoFn(key); }  @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 PairWithConstantKeyDoFnFactory() .create( null , CloudObject.fromSpec( ImmutableMap.of( PropertyNames.OBJECT_TYPE_NAME, "PairWithConstantKeyDoFn", WorkerPropertyNames.ENCODED_KEY, StringUtils.byteArrayToJsonString( CoderUtils.encodeToByteArray(BigEndianIntegerCoder.of(), 42)), PropertyNames.ENCODING, ImmutableMap.of( PropertyNames.OBJECT_TYPE_NAME, "kind:fixed_big_endian_int32"))), null , null , null , null , null ); List<Object> outputReceiver = new ArrayList<>(); parDoFn.startBundle(outputReceiver::add); parDoFn.processElement(valueInGlobalWindow(43)); assertThat(outputReceiver, contains(valueInGlobalWindow(KV.of(42, 43)))); }