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

    PCollectionView<Integer> gradesView =
        pipeline.apply("grades", Create.of(50)).apply(View.asSingleton());

    Create.Values<Integer> studentsPercentage = Create.of(5, 45, 90, 29, 55, 65);
    PCollectionList<Integer> studentsGrades =
        pipeline
            .apply(studentsPercentage)
            .apply(
                Partition.of(
                    2,
                    (elem, numPartitions, ct) -> {
                      Integer grades = ct.sideInput(gradesView);
                      return elem < grades ? 0 : 1;
                    },
                    Requirements.requiresSideInputs(gradesView)));

    assertTrue(studentsGrades.size() == 2);
    PAssert.that(studentsGrades.get(0)).containsInAnyOrder(5, 29, 45);
    PAssert.that(studentsGrades.get(1)).containsInAnyOrder(55, 65, 90);
    pipeline.run();
  }