[FLINK-15458][conf][docs] Introduce SuffixOption annotation

They key of a SuffixOption is only a suffix, with the prefix being determined at runtime.
In the case of reporters the prefix is "metrics.reporter." with a user-provided infix.

For these options it doesn't make sense to check whether they are clashing with other options;
in practive this is entirely dependent on the prefix.

flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java

@@ -104,6 +104,16 @@ public String toString() {
  String value() default "";
  }
 
+ /**
+ * Annotation used on config option fields or options class to mark them as a suffix-option; i.e., a config option
+ * where the key is only a suffix, with the prefix being danymically provided at runtime.
+ */
+ @Target({ElementType.FIELD, ElementType.TYPE})
+ @Retention(RetentionPolicy.RUNTIME)
+ @Internal
+ public @interface SuffixOption {
+ }
+
  private Documentation(){
  }
 }

flink-docs/src/test/java/org/apache/flink/docs/configuration/ConfigOptionsDocsCompletenessITCase.java

@@ -29,6 +29,7 @@
 import org.junit.Test;
 
 import java.io.IOException;
+import java.lang.reflect.Field;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -84,6 +85,7 @@ public void testFullReferenceCompleteness() throws IOException, ClassNotFoundExc
 
  private static void assertExistingOptionsAreWellDefined(Map<String, List<ExistingOption>> allOptions) {
  allOptions.values().stream()
+ .filter(options -> !options.stream().allMatch(option -> option.isSuffixOption))
  .forEach(options -> options.stream()
  .reduce((option1, option2) -> {
  if (option1.equals(option2)) {
@@ -233,21 +235,32 @@ private static ExistingOption toExistingOption(ConfigOptionsDocGenerator.OptionW
  String defaultValue = stringifyDefault(optionWithMetaInfo);
  String typeValue = typeToHtml(optionWithMetaInfo);
  String description = htmlFormatter.format(optionWithMetaInfo.option.description());
- return new ExistingOption(key, defaultValue, typeValue, description, optionsClass);
+ boolean isSuffixOption = isSuffixOption(optionWithMetaInfo.field);
+ return new ExistingOption(key, defaultValue, typeValue, description, optionsClass, isSuffixOption);
+ }
+
+ private static boolean isSuffixOption(Field field) {
+ final Class<?> containingOptionsClass = field.getDeclaringClass();
+
+ return field.getAnnotation(Documentation.SuffixOption.class) != null ||
+ containingOptionsClass.getAnnotation(Documentation.SuffixOption.class) != null;
  }
 
  private static final class ExistingOption extends Option {
 
  private final Class<?> containingClass;
+ private final boolean isSuffixOption;
 
  private ExistingOption(
  String key,
  String defaultValue,
  String typeValue,
  String description,
- Class<?> containingClass) {
+ Class<?> containingClass,
+ boolean isSuffixOption) {
  super(key, defaultValue, typeValue, description);
  this.containingClass = containingClass;
+ this.isSuffixOption = isSuffixOption;
  }
  }
 

flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporterOptions.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.metrics.prometheus;
 
+import org.apache.flink.annotation.docs.Documentation;
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.ConfigOptions;
 import org.apache.flink.configuration.description.Description;
@@ -27,6 +28,7 @@
 /**
  * Config options for the {@link PrometheusPushGatewayReporter}.
  */
+@Documentation.SuffixOption
 public class PrometheusPushGatewayReporterOptions {
 
  public static final ConfigOption<String> HOST = ConfigOptions