Keys extends PTransform<PCollection<? extends KV<K, ?>>, PCollection<K>> { public static <K> Keys<K> create() { return new Keys<>(); } private  Keys(); static Keys<K> create(); @Override PCollection<K> expand(PCollection<? extends KV<K, ?>> in);  }
@Test @Category(ValidatesRunner.class) public void testKeysEmpty() { PCollection<KV<String, Integer>> input = p.apply( Create.of(Arrays.asList(EMPTY_TABLE)) .withCoder(KvCoder.of(StringUtf8Coder.of(), BigEndianIntegerCoder.of()))); PCollection<String> output = input.apply(Keys.create()); PAssert.that(output).empty(); p.run(); }