TimestampExtractTransform extends PTransform<PCollection<InputT>, PCollection<OutputT>> { public static <InputT, OutputT> TimestampExtractTransform<InputT, OutputT> of( PCollectionTransform<InputT, OutputT> transform) { return new TimestampExtractTransform<>(null, transform); } private  TimestampExtractTransform(
      @Nullable String name, PCollectionTransform<InputT, OutputT> timestampedTransform); static TimestampExtractTransform<InputT, OutputT> of(
      PCollectionTransform<InputT, OutputT> transform); static TimestampExtractTransform<InputT, OutputT> of(
      String name, PCollectionTransform<InputT, OutputT> transform); @Override PCollection<OutputT> expand(PCollection<InputT> input);  }
@SuppressWarnings("unchecked") @Test(timeout = 10000) public void testTransform() { Pipeline p = Pipeline.create(); PCollection<Integer> input = p.apply(Create.of(1, 2, 3)); PCollection<KV<Integer, Long>> result = input.apply( TimestampExtractTransform.of( in -> CountByKey.of(in).keyBy(KV::getValue, TypeDescriptors.integers()).output())); PAssert.that(result).containsInAnyOrder(KV.of(1, 1L), KV.of(2, 1L), KV.of(3, 1L)); p.run().waitUntilFinish(); }