private KvSwap() {}
@Test
  @Category(ValidatesRunner.class)
  public void testKvSwap() {
    PCollection<KV<String, Integer>> input =
        p.apply(
            Create.of(Arrays.asList(TABLE))
                .withCoder(
                    KvCoder.of(
                        StringUtf8Coder.of(), NullableCoder.of(BigEndianIntegerCoder.of()))));

    PCollection<KV<Integer, String>> output = input.apply(KvSwap.create());

    PAssert.that(output)
        .containsInAnyOrder(
            KV.of(1, "one"),
            KV.of(2, "two"),
            KV.of(3, "three"),
            KV.of(4, "four"),
            KV.of(4, "dup"),
            KV.of(5, "dup"),
            KV.of((Integer) null, "null"));
    p.run();
  }