public static <FnT extends DoFn<?, ?>> DoFnSignature getSignature(Class<FnT> fn) {
    return signatureCache.computeIfAbsent(fn, DoFnSignatures::parseSignature);
  }
@Test
  public void testRequiresStableInputProcessElement() throws Exception {
    DoFnSignature sig =
        DoFnSignatures.getSignature(
            new DoFn<String, String>() {
              @ProcessElement
              @RequiresStableInput
              public void process(ProcessContext c) {}
            }.getClass());

    assertThat(sig.processElement().requiresStableInput(), is(true));
  }