docs: @Internal
  	public static Set<Annotation> readDualForwardAnnotations(Class<?> udfClass) {
  
  		// get readSet annotation from stub
  		ForwardedFieldsFirst forwardedFields1 = udfClass.getAnnotation(ForwardedFieldsFirst.class);
  		ForwardedFieldsSecond forwardedFields2 = udfClass.getAnnotation(ForwardedFieldsSecond.class);
  
  		// get readSet annotation from stub
  		NonForwardedFieldsFirst nonForwardedFields1 = udfClass.getAnnotation(NonForwardedFieldsFirst.class);
  		NonForwardedFieldsSecond nonForwardedFields2 = udfClass.getAnnotation(NonForwardedFieldsSecond.class);
  
  		ReadFieldsFirst readSet1 = udfClass.getAnnotation(ReadFieldsFirst.class);
  		ReadFieldsSecond readSet2 = udfClass.getAnnotation(ReadFieldsSecond.class);
  
  		Set<Annotation> annotations = new HashSet<Annotation>();
  
  		if (nonForwardedFields1 != null && forwardedFields1 != null) {
  			throw new InvalidProgramException("Either " + ForwardedFieldsFirst.class.getSimpleName() + " or " +
  					NonForwardedFieldsFirst.class.getSimpleName() + " can be annotated to a function, not both.");
  		} else if (forwardedFields1 != null) {
  			annotations.add(forwardedFields1);
  		} else if (nonForwardedFields1 != null) {
  			annotations.add(nonForwardedFields1);
  		}
  
  		if (forwardedFields2 != null && nonForwardedFields2 != null) {
  			throw new InvalidProgramException("Either " + ForwardedFieldsSecond.class.getSimpleName() + " or " +
  					NonForwardedFieldsSecond.class.getSimpleName() + " can be annotated to a function, not both.");
  		} else if (forwardedFields2 != null) {
  			annotations.add(forwardedFields2);
  		} else if (nonForwardedFields2 != null) {
  			annotations.add(nonForwardedFields2);
  		}
  
  		if (readSet1 != null) {
  			annotations.add(readSet1);
  		}
  		if (readSet2 != null) {
  			annotations.add(readSet2);
  		}
  
  		return !annotations.isEmpty() ? annotations : null;
  	}