public static <K, InputT, AccumT> GroupingTable<K, InputT, AccumT> combining(
      GroupingKeyCreator<? super K> groupingKeyCreator,
      PairInfo pairInfo,
      Combiner<? super K, InputT, AccumT, ?> combineFn,
      SizeEstimator<? super K> keySizer,
      SizeEstimator<? super AccumT> accumulatorSizer) {
    return new CombiningGroupingTable<>(
        DEFAULT_MAX_GROUPING_TABLE_BYTES,
        groupingKeyCreator,
        pairInfo,
        combineFn,
        keySizer,
        accumulatorSizer);
  }
@Test
  public void testCombiningGroupingTable() throws Exception {
    Combiner<Object, Integer, Long, Long> summingCombineFn =
        new Combiner<Object, Integer, Long, Long>() {

          @Override
          public Long createAccumulator(Object key) {
            return 0L;
          }

          @Override
          public Long add(Object key, Long accumulator, Integer value) {
            return accumulator + value;
          }

          @Override
          public Long merge(Object key, Iterable<Long> accumulators) {
            long sum = 0;
            for (Long part : accumulators) {
              sum += part;
            }
            return sum;
          }

          @Override
          public Long compact(Object key, Long accumulator) {
            return accumulator;
          }

          @Override
          public Long extract(Object key, Long accumulator) {
            return accumulator;
          }
        };

    GroupingTableBase<String, Integer, Long> table =
        (GroupingTableBase<String, Integer, Long>)
            GroupingTables.combining(
                new IdentityGroupingKeyCreator(),
                new KvPairInfo(),
                summingCombineFn,
                new StringPowerSizeEstimator(),
                new IdentitySizeEstimator());
    table.setMaxSize(1000);

    TestOutputReceiver receiver =
        new TestOutputReceiver(
            KvCoder.of(StringUtf8Coder.of(), BigEndianLongCoder.of()),
            NameContextsForTests.nameContextForTest());

    table.put("A", 1, receiver);
    table.put("B", 2, receiver);
    table.put("B", 3, receiver);
    table.put("C", 4, receiver);
    assertThat(receiver.outputElems, empty());

    table.put("C", 5000, receiver);
    assertThat(receiver.outputElems, hasItem((Object) KV.of("C", 5004L)));

    table.put("DDDD", 6, receiver);
    assertThat(receiver.outputElems, hasItem((Object) KV.of("DDDD", 6L)));

    table.flush(receiver);
    assertThat(
        receiver.outputElems,
        IsIterableContainingInAnyOrder.<Object>containsInAnyOrder(
            KV.of("A", 1L), KV.of("B", 2L + 3), KV.of("C", 5000L + 4), KV.of("DDDD", 6L)));
  }