public static <T> Partition<T> of(
      int numPartitions,
      PartitionWithSideInputsFn<? super T> partitionFn,
      Requirements requirements) {
    Contextful ctfFn =
        Contextful.fn(
            (T element, Contextful.Fn.Context c) ->
                partitionFn.partitionFor(element, numPartitions, c),
            requirements);
    return new Partition<>(new PartitionDoFn<T>(numPartitions, ctfFn, partitionFn));
  }
@Test
  @Category(NeedsRunner.class)
  public void testModPartitionWithLambda() {

    PCollectionList<Integer> outputs =
        pipeline
            .apply(Create.of(1, 2, 4, 5))
            .apply(Partition.of(3, (element, numPartitions) -> element % numPartitions));
    assertEquals(3, outputs.size());
    PAssert.that(outputs.get(0)).empty();
    PAssert.that(outputs.get(1)).containsInAnyOrder(1, 4);
    PAssert.that(outputs.get(2)).containsInAnyOrder(2, 5);
    pipeline.run();
  }