public static <K, V> GroupingTable<K, V, List<V>> buffering(
      GroupingKeyCreator<? super K> groupingKeyCreator,
      PairInfo pairInfo,
      SizeEstimator<? super K> keySizer,
      SizeEstimator<? super V> valueSizer) {
    return new BufferingGroupingTable<>(
        DEFAULT_MAX_GROUPING_TABLE_BYTES, groupingKeyCreator, pairInfo, keySizer, valueSizer);
  }
@Test
  public void testBufferingGroupingTable() throws Exception {
    GroupingTableBase<String, String, List<String>> table =
        (GroupingTableBase<String, String, List<String>>)
            GroupingTables.buffering(
                new IdentityGroupingKeyCreator(), new KvPairInfo(),
                new StringPowerSizeEstimator(), new StringPowerSizeEstimator());
    table.setMaxSize(1000);
    TestOutputReceiver receiver =
        new TestOutputReceiver(
            KvCoder.of(StringUtf8Coder.of(), IterableCoder.of(StringUtf8Coder.of())),
            NameContextsForTests.nameContextForTest());

    table.put("A", "a", receiver);
    table.put("B", "b1", receiver);
    table.put("B", "b2", receiver);
    table.put("C", "c", receiver);
    assertThat(receiver.outputElems, empty());

    table.put("C", "cccc", receiver);
    assertThat(receiver.outputElems, hasItem((Object) KV.of("C", Arrays.asList("c", "cccc"))));

    table.put("DDDD", "d", receiver);
    assertThat(receiver.outputElems, hasItem((Object) KV.of("DDDD", Arrays.asList("d"))));

    table.flush(receiver);
    assertThat(
        receiver.outputElems,
        IsIterableContainingInAnyOrder.<Object>containsInAnyOrder(
            KV.of("A", Arrays.asList("a")),
            KV.of("B", Arrays.asList("b1", "b2")),
            KV.of("C", Arrays.asList("c", "cccc")),
            KV.of("DDDD", Arrays.asList("d"))));
  }