@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);
  }
@Test
  public void testConversionOfRecord() throws Exception {
    ParDoFn parDoFn =
        new PairWithConstantKeyDoFnFactory()
            .create(
                null /* pipeline options */,
                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 /* 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(43));
    assertThat(outputReceiver, contains(valueInGlobalWindow(KV.of(42, 43))));
  }