public static <K, InputT> GroupIntoBatches<K, InputT> ofByteSize(long batchSizeBytes) {
    return new GroupIntoBatches<K, InputT>(BatchingParams.createDefault())
        .withByteSize(batchSizeBytes);
  }
@Test
  @Category({
    ValidatesRunner.class,
    NeedsRunner.class,
    UsesTestStream.class,
    UsesTimersInParDo.class,
    UsesStatefulParDo.class,
    UsesOnWindowExpiration.class
  })
  public void testInGlobalWindowBatchSizeByteSizeFn() {
    SerializableFunction<String, Long> getElementByteSizeFn =
        s -> {
          try {
            return 2 * StringUtf8Coder.of().getEncodedElementByteSize(s);
          } catch (Exception e) {
            throw new RuntimeException(e);
          }
        };

    // to ensure ordered processing
    TestStream.Builder<KV<String, String>> streamBuilder =
        TestStream.create(KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()))
            .advanceWatermarkTo(Instant.EPOCH);

    long offset = 0L;
    for (KV<String, String> kv : data) {
      streamBuilder =
          streamBuilder.addElements(
              TimestampedValue.of(kv, Instant.EPOCH.plus(Duration.standardSeconds(offset))));
      offset++;
    }

    // fire them all at once
    TestStream<KV<String, String>> stream = streamBuilder.advanceWatermarkToInfinity();

    PCollection<KV<String, Iterable<String>>> collection =
        pipeline
            .apply("Input data", stream)
            .apply(GroupIntoBatches.ofByteSize(BATCH_SIZE_BYTES, getElementByteSizeFn))
            // set output coder
            .setCoder(KvCoder.of(StringUtf8Coder.of(), IterableCoder.of(StringUtf8Coder.of())));
    PAssert.that("Incorrect batch size in one or more elements", collection)
        .satisfies(
            new SerializableFunction<Iterable<KV<String, Iterable<String>>>, Void>() {
              @Override
              public Void apply(Iterable<KV<String, Iterable<String>>> input) {
                assertTrue(checkBatchByteSizes(input, getElementByteSizeFn));
                assertEquals("Invalid batch count", 9L, Iterables.size(input));
                return null;
              }
            });
    pipeline.run();
  }