public static <PrimaryKeyT, SecondaryKeyT, ValueT>
      SortValues<PrimaryKeyT, SecondaryKeyT, ValueT> create(
          BufferedExternalSorter.Options sorterOptions) {
    return new SortValues<>(sorterOptions);
  }
@Test
  public void testSecondaryKeySorting() {
    // Create a PCollection of <Key, <SecondaryKey, Value>> pairs.
    PCollection<KV<String, KV<String, Integer>>> input =
        p.apply(
            Create.of(
                Arrays.asList(
                    KV.of("key1", KV.of("secondaryKey2", 20)),
                    KV.of("key2", KV.of("secondaryKey2", 200)),
                    KV.of("key1", KV.of("secondaryKey3", 30)),
                    KV.of("key1", KV.of("secondaryKey1", 10)),
                    KV.of("key2", KV.of("secondaryKey1", 100)))));

    // Group by Key, bringing <SecondaryKey, Value> pairs for the same Key together.
    PCollection<KV<String, Iterable<KV<String, Integer>>>> grouped =
        input.apply(GroupByKey.create());

    // For every Key, sort the iterable of <SecondaryKey, Value> pairs by SecondaryKey.
    PCollection<KV<String, Iterable<KV<String, Integer>>>> groupedAndSorted =
        grouped.apply(SortValues.create(BufferedExternalSorter.options()));

    PAssert.that(groupedAndSorted)
        .satisfies(
            new AssertThatHasExpectedContentsForTestSecondaryKeySorting<>(
                Arrays.asList(
                    KV.of(
                        "key1",
                        Arrays.asList(
                            KV.of("secondaryKey1", 10),
                            KV.of("secondaryKey2", 20),
                            KV.of("secondaryKey3", 30))),
                    KV.of(
                        "key2",
                        Arrays.asList(KV.of("secondaryKey1", 100), KV.of("secondaryKey2", 200))))));

    p.run();
  }