public static <FnT extends DoFn<?, ?>> DoFnSignature getSignature(Class<FnT> fn) {
    return signatureCache.computeIfAbsent(fn, DoFnSignatures::parseSignature);
  }
@Test
  public void testStateParameterAlwaysFetched() {
    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage("ReadableStates");
    DoFnSignature sig =
        DoFnSignatures.getSignature(
            new DoFn<KV<String, Integer>, Long>() {

              @StateId("my-id")
              private final StateSpec<MapState<Integer, Integer>> myfield =
                  StateSpecs.map(VarIntCoder.of(), VarIntCoder.of());

              @ProcessElement
              public void myProcessElement(
                  ProcessContext context,
                  @AlwaysFetched @StateId("my-id") MapState<Integer, Integer> one) {}
            }.getClass());
    StateParameter stateParameter = (StateParameter) sig.processElement().extraParameters().get(1);
    assertTrue(stateParameter.alwaysFetched());
  }