public static <K, InputT> GroupIntoBatches<K, InputT> ofSize(long batchSize) {
    Preconditions.checkState(batchSize < Long.MAX_VALUE);
    return new GroupIntoBatches<K, InputT>(BatchingParams.createDefault()).withSize(batchSize);
  }
@Test
  @Category({
    ValidatesRunner.class,
    NeedsRunner.class,
    UsesTimersInParDo.class,
    UsesStatefulParDo.class,
    UsesOnWindowExpiration.class
  })
  public void testWithUnevenBatches() {
    PCollection<KV<String, Iterable<String>>> collection =
        pipeline
            .apply("Input data", Create.of(createTestData(ODD_NUM_ELEMENTS)))
            .apply(GroupIntoBatches.ofSize(BATCH_SIZE))
            // 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(checkBatchSizes(input));
                return null;
              }
            });
    PAssert.thatSingleton("Incorrect collection size", collection.apply("Count", Count.globally()))
        .isEqualTo(
            // round up division for positive numbers
            // https://math.stackexchange.com/questions/2591316/proof-for-integer-division-algorithm-that-rounds-up.
            (ODD_NUM_ELEMENTS + BATCH_SIZE - 1) / BATCH_SIZE);
    pipeline.run();
  }