public static <K, InputT, AccumT> ParDoFn create(
      PipelineOptions options,
      KvCoder<K, ?> inputElementCoder,
      @Nullable CloudObject cloudUserFn,
      @Nullable List<SideInputInfo> sideInputInfos,
      List<Receiver> receivers,
      DataflowExecutionContext<?> executionContext,
      DataflowOperationContext operationContext)
      throws Exception {
    AppliedCombineFn<K, InputT, AccumT, ?> combineFn;
    SideInputReader sideInputReader;
    StepContext stepContext;
    if (cloudUserFn == null) {
      combineFn = null;
      sideInputReader = NullSideInputReader.empty();
      stepContext = null;
    } else {
      Object deserializedFn =
          SerializableUtils.deserializeFromByteArray(
              getBytes(cloudUserFn, PropertyNames.SERIALIZED_FN), "serialized combine fn");
      @SuppressWarnings("unchecked")
      AppliedCombineFn<K, InputT, AccumT, ?> combineFnUnchecked =
          ((AppliedCombineFn<K, InputT, AccumT, ?>) deserializedFn);
      combineFn = combineFnUnchecked;

      sideInputReader =
          executionContext.getSideInputReader(
              sideInputInfos, combineFn.getSideInputViews(), operationContext);
      stepContext = executionContext.getStepContext(operationContext);
    }
    return create(
        options, inputElementCoder, combineFn, sideInputReader, receivers.get(0), stepContext);
  }
@Test
  public void testCreateWithCombinerAndBatchSideInputs() throws Exception {
    PipelineOptions options = PipelineOptionsFactory.create();

    Coder keyCoder = StringUtf8Coder.of();
    Coder valueCoder = BigEndianIntegerCoder.of();
    KvCoder<String, Integer> kvCoder = KvCoder.of(keyCoder, valueCoder);

    TestOutputReceiver receiver =
        new TestOutputReceiver(
            new ElementByteSizeObservableCoder(WindowedValue.getValueOnlyCoder(kvCoder)),
            counterSet,
            NameContextsForTests.nameContextForTest());

    StepContext stepContext =
        BatchModeExecutionContext.forTesting(options, "testStage")
            .getStepContext(TestOperationContext.create(counterSet));

    when(mockSideInputReader.isEmpty()).thenReturn(false);

    ParDoFn pgbk =
        PartialGroupByKeyParDoFns.create(
            options,
            kvCoder,
            AppliedCombineFn.withInputCoder(
                Sum.ofIntegers(),
                CoderRegistry.createDefault(),
                kvCoder,
                ImmutableList.<PCollectionView<?>>of(),
                WindowingStrategy.globalDefault()),
            mockSideInputReader,
            receiver,
            stepContext);
    assertTrue(pgbk instanceof BatchSideInputPGBKParDoFn);
  }