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

    // Compute the set of integers either 1 or 2 mod 3, the hard way.
    PCollectionList<Integer> outputs =
        pipeline
            .apply(Create.of(2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12))
            .apply(Partition.of(3, new ModFn()));

    List<PCollection<Integer>> outputsList = new ArrayList<>(outputs.getAll());
    outputsList.remove(0);
    outputs = PCollectionList.of(outputsList);
    assertTrue(outputs.size() == 2);

    PCollection<Integer> output = outputs.apply(Flatten.pCollections());
    PAssert.that(output).containsInAnyOrder(2, 4, 5, 7, 8, 10, 11);
    pipeline.run();
  }