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 testPartialGroupByKeyWithCombiner() throws Exception {
    Coder keyCoder = StringUtf8Coder.of();
    Coder valueCoder = BigEndianIntegerCoder.of();

    TestOutputReceiver receiver =
        new TestOutputReceiver(
            new ElementByteSizeObservableCoder(
                WindowedValue.getValueOnlyCoder(KvCoder.of(keyCoder, valueCoder))),
            counterSet,
            NameContextsForTests.nameContextForTest());

    Combiner<WindowedValue<String>, Integer, Integer, Integer> combineFn = new TestCombiner();

    ParDoFn pgbkParDoFn =
        new SimplePartialGroupByKeyParDoFn(
            GroupingTables.combining(
                new WindowingCoderGroupingKeyCreator(keyCoder),
                PairInfo.create(),
                combineFn,
                new CoderSizeEstimator(WindowedValue.getValueOnlyCoder(keyCoder)),
                new CoderSizeEstimator(valueCoder)),
            receiver);

    pgbkParDoFn.startBundle(receiver);

    pgbkParDoFn.processElement(WindowedValue.valueInGlobalWindow(KV.of("hi", 4)));
    pgbkParDoFn.processElement(WindowedValue.valueInGlobalWindow(KV.of("there", 5)));
    pgbkParDoFn.processElement(WindowedValue.valueInGlobalWindow(KV.of("hi", 6)));
    pgbkParDoFn.processElement(WindowedValue.valueInGlobalWindow(KV.of("joe", 7)));
    pgbkParDoFn.processElement(WindowedValue.valueInGlobalWindow(KV.of("there", 8)));
    pgbkParDoFn.processElement(WindowedValue.valueInGlobalWindow(KV.of("hi", 9)));

    pgbkParDoFn.finishBundle();

    assertThat(
        receiver.outputElems,
        IsIterableContainingInAnyOrder.<Object>containsInAnyOrder(
            WindowedValue.valueInGlobalWindow(KV.of("hi", 19)),
            WindowedValue.valueInGlobalWindow(KV.of("there", 13)),
            WindowedValue.valueInGlobalWindow(KV.of("joe", 7))));

    // Exact counter values depend on size of encoded data.  If encoding
    // changes, then these expected counters should change to match.
    CounterUpdateExtractor<?> updateExtractor = Mockito.mock(CounterUpdateExtractor.class);
    counterSet.extractUpdates(false, updateExtractor);
    verify(updateExtractor).longSum(getObjectCounterName("test_receiver_out"), false, 3L);
    verify(updateExtractor)
        .longMean(
            getMeanByteCounterName("test_receiver_out"),
            false,
            LongCounterMean.ZERO.addValue(25L, 3));
    verifyNoMoreInteractions(updateExtractor);
  }