public static <FnT extends DoFn<?, ?>> DoFnSignature getSignature(Class<FnT> fn) {
    return signatureCache.computeIfAbsent(fn, DoFnSignatures::parseSignature);
  }
@Test
  public void testMultipleSchemaParameters() {
    DoFnSignature sig =
        DoFnSignatures.getSignature(
            new DoFn<String, String>() {
              @ProcessElement
              public void process(
                  @Element Row row1,
                  @Timestamp Instant ts,
                  @Element Row row2,
                  OutputReceiver<String> o,
                  @Element Integer intParameter) {}
            }.getClass());
    assertEquals(3, sig.processElement().getSchemaElementParameters().size());
    assertEquals(0, sig.processElement().getSchemaElementParameters().get(0).index());
    assertEquals(
        TypeDescriptors.rows(),
        sig.processElement().getSchemaElementParameters().get(0).elementT());
    assertEquals(1, sig.processElement().getSchemaElementParameters().get(1).index());
    assertEquals(
        TypeDescriptors.rows(),
        sig.processElement().getSchemaElementParameters().get(1).elementT());
    assertEquals(2, sig.processElement().getSchemaElementParameters().get(2).index());
    assertEquals(
        TypeDescriptors.integers(),
        sig.processElement().getSchemaElementParameters().get(2).elementT());
  }