public static <InputT, OutputT> TimestampExtractTransform<InputT, OutputT> of(
      PCollectionTransform<InputT, OutputT> transform) {

    return new TimestampExtractTransform<>(null, transform);
  }
@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();
  }