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 testEvenOddPartition() {

    PCollectionList<Integer> outputs =
        pipeline
            .apply(Create.of(591, 11789, 1257, 24578, 24799, 307))
            .apply(Partition.of(2, new ModFn()));
    assertTrue(outputs.size() == 2);
    PAssert.that(outputs.get(0)).containsInAnyOrder(24578);
    PAssert.that(outputs.get(1)).containsInAnyOrder(591, 11789, 1257, 24799, 307);
    pipeline.run();
  }