public static <K, V> WithKeys<K, V> of(SerializableFunction<V, K> fn) {
    checkNotNull(
        fn, "WithKeys constructed with null function. Did you mean WithKeys.of((Void) null)?");
    return new WithKeys<>(fn, null);
  }
@Test
  @Category(NeedsRunner.class)
  public void testExtractKeys() {

    PCollection<String> input =
        p.apply(Create.of(Arrays.asList(COLLECTION)).withCoder(StringUtf8Coder.of()));

    PCollection<KV<Integer, String>> output = input.apply(WithKeys.of(new LengthAsKey()));
    PAssert.that(output).containsInAnyOrder(WITH_KEYS);

    p.run();
  }