[FLINK-8135][REST][docs] Add description to MessageParameter

flink-core/src/main/java/org/apache/flink/util/StringUtils.java

@@ -28,7 +28,9 @@
 
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.Objects;
 import java.util.Random;
+import java.util.stream.Collectors;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
@@ -348,6 +350,21 @@ public static String concatenateWithAnd(@Nullable String s1, @Nullable String s2
  }
  }
 
+ /**
+ * Generates a string containing a comma-separated list of values in double-quotes.
+ * Uses lower-cased values returned from {@link Object#toString()} method for each element in the given array.
+ * Null values are skipped.
+ *
+ * @param values array of elements for the list
+ *
+ * @return The string with quoted list of elements
+ */
+ public static String toQuotedListString(Object[] values) {
+ return Arrays.stream(values).filter(Objects::nonNull)
+ .map(v -> v.toString().toLowerCase())
+ .collect(Collectors.joining(", ", "\"", "\""));
+ }
+
  // ------------------------------------------------------------------------
 
  /** Prevent instantiation of this utility class. */

flink-docs/src/main/java/org/apache/flink/docs/rest/RestAPIDocGenerator.java

@@ -226,7 +226,7 @@ private static String createPathParameterHtmlList(Collection<MessagePathParamete
  pathParameterList.append(
  String.format("<li><code>%s</code> - %s</li>\n",
  messagePathParameter.getKey(),
- "description")
+ messagePathParameter.getDescription())
  ));
  return pathParameterList.toString();
  }
@@ -240,7 +240,7 @@ private static String createQueryParameterHtmlList(Collection<MessageQueryParame
  String.format("<li><code>%s</code> (%s): %s</li>\n",
  parameter.getKey(),
  parameter.isMandatory() ? "mandatory" : "optional",
- "description")
+ parameter.getDescription())
  ));
  return queryParameterList.toString();
  }

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/AllowNonRestoredStateQueryParameter.java

@@ -42,4 +42,10 @@ public Boolean convertStringToValue(final String value) {
  public String convertValueToString(final Boolean value) {
  return value.toString();
  }
+
+ @Override
+ public String getDescription() {
+ return "Boolean value that specifies whether the job submission should be rejected " +
+ "if the savepoint contains state that cannot be mapped back to the job.";
+ }
 }

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/EntryClassQueryParameter.java

@@ -28,4 +28,10 @@ public class EntryClassQueryParameter extends StringQueryParameter {
  public EntryClassQueryParameter() {
  super("entry-class", MessageParameterRequisiteness.OPTIONAL);
  }
+
+ @Override
+ public String getDescription() {
+ return "String value that specifies the fully qualified name of the entry point class. " +
+ "Overrides the class defined in the jar file manifest.";
+ }
 }

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarIdPathParameter.java

@@ -48,4 +48,11 @@ protected String convertFromString(final String value) throws ConversionExceptio
  protected String convertToString(final String value) {
  return value;
  }
+
+ @Override
+ public String getDescription() {
+ return "String value that identifies a jar. When uploading the jar a path is returned, where the filename " +
+ "is the ID. This value is equivalent to the `" + JarListInfo.JarFileInfo.JAR_FILE_FIELD_ID + "` field " +
+ "in the list of uploaded jars (" + JarListHeaders.URL + ").";
+ }
 }

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/ParallelismQueryParameter.java

@@ -41,4 +41,9 @@ public Integer convertStringToValue(final String value) {
  public String convertValueToString(final Integer value) {
  return value.toString();
  }
+
+ @Override
+ public String getDescription() {
+ return "Positive integer value that specifies the desired parallelism for the job.";
+ }
 }

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/ProgramArgsQueryParameter.java

@@ -30,4 +30,8 @@ public ProgramArgsQueryParameter() {
  super("program-args", MessageParameterRequisiteness.OPTIONAL);
  }
 
+ @Override
+ public String getDescription() {
+ return "String value that specifies the arguments for the program or plan.";
+ }
 }

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SavepointPathQueryParameter.java

@@ -28,4 +28,9 @@ public class SavepointPathQueryParameter extends StringQueryParameter {
  public SavepointPathQueryParameter() {
  super(KEY, MessageParameterRequisiteness.OPTIONAL);
  }
+
+ @Override
+ public String getDescription() {
+ return "String value that specifies the path of the savepoint to restore the job from.";
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/AccumulatorsIncludeSerializedValueQueryParameter.java

@@ -38,4 +38,9 @@ public String convertValueToString(Boolean value) {
  public Boolean convertStringToValue(String value) {
  return Boolean.valueOf(value);
  }
+
+ @Override
+ public String getDescription() {
+ return "Boolean value that specifies whether serialized user task accumulators should be included in the response.";
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobIDPathParameter.java

@@ -40,4 +40,9 @@ protected JobID convertFromString(String value) {
  protected String convertToString(JobID value) {
  return value.toString();
  }
+
+ @Override
+ public String getDescription() {
+ return "32-character hexadecimal string value that identifies a job.";
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexIdPathParameter.java

@@ -40,4 +40,9 @@ protected JobVertexID convertFromString(String value) throws ConversionException
  protected String convertToString(JobVertexID value) {
  return value.toString();
  }
+
+ @Override
+ public String getDescription() {
+ return "32-character hexadecimal string value that identifies a job vertex.";
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/MessageParameter.java

@@ -137,4 +137,10 @@ protected enum MessageParameterRequisiteness {
  MANDATORY,
  OPTIONAL
  }
+
+ /**
+ * Returns a description for REST API HTML documentation.
+ * @return escaped HTML string
+ */
+ public abstract String getDescription();
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/RescalingParallelismQueryParameter.java

@@ -38,4 +38,9 @@ public Integer convertStringToValue(String value) {
  public String convertValueToString(Integer value) {
  return value.toString();
  }
+
+ @Override
+ public String getDescription() {
+ return "Positive integer value that specifies the desired parallelism.";
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/SubtaskIndexPathParameter.java

@@ -44,4 +44,9 @@ protected String convertToString(final Integer value) {
  return value.toString();
  }
 
+ @Override
+ public String getDescription() {
+ return "Positive integer value that identifies a subtask.";
+ }
+
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/TerminationModeQueryParameter.java

@@ -19,6 +19,7 @@
 package org.apache.flink.runtime.rest.messages;
 
 import org.apache.flink.runtime.rest.handler.legacy.JobCancellationHandler;
+import org.apache.flink.util.StringUtils;
 
 /**
  * Termination mode for the {@link JobCancellationHandler}.
@@ -41,6 +42,12 @@ public String convertValueToString(TerminationMode value) {
  return value.name().toLowerCase();
  }
 
+ @Override
+ public String getDescription() {
+ return "String value that specifies the termination mode. Supported values are: " +
+ StringUtils.toQuotedListString(TerminationMode.values()) + '.';
+ }
+
  /**
  * Supported termination modes.
  */

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/TriggerIdPathParameter.java

@@ -38,4 +38,10 @@ protected TriggerId convertFromString(String value) throws ConversionException {
  protected String convertToString(TriggerId value) {
  return value.toString();
  }
+
+ @Override
+ public String getDescription() {
+ return "32-character hexadecimal string that identifies an asynchronous operation trigger ID. " +
+ "The ID was returned then the operation was triggered.";
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointIdPathParameter.java

@@ -45,4 +45,9 @@ protected Long convertFromString(String value) throws ConversionException {
  protected String convertToString(Long value) {
  return value.toString();
  }
+
+ @Override
+ public String getDescription() {
+ return "Long value that identifies a checkpoint.";
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/SubtaskAttemptPathParameter.java

@@ -50,4 +50,9 @@ protected Integer convertFromString(String value) throws ConversionException {
  protected String convertToString(Integer value) {
  return value.toString();
  }
+
+ @Override
+ public String getDescription() {
+ return "Positive integer value that identifies an execution attempt.";
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/JobsFilterQueryParameter.java

@@ -45,4 +45,9 @@ public JobID convertStringToValue(String value) throws ConversionException {
  public String convertValueToString(JobID value) {
  return value.toString();
  }
+
+ @Override
+ public String getDescription() {
+ return "Comma-separated list of 32-character hexadecimal strings to select specific jobs.";
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/MetricsAggregationParameter.java

@@ -20,6 +20,7 @@
 
 import org.apache.flink.runtime.rest.messages.ConversionException;
 import org.apache.flink.runtime.rest.messages.MessageQueryParameter;
+import org.apache.flink.util.StringUtils;
 
 import java.util.Locale;
 
@@ -46,6 +47,12 @@ public String convertValueToString(AggregationMode value) {
  return value.name().toLowerCase();
  }
 
+ @Override
+ public String getDescription() {
+ return "Comma-separated list of aggregation modes which should be calculated. " +
+ "Available aggregations are: " + StringUtils.toQuotedListString(AggregationMode.values()) + '.';
+ }
+
  /**
  * The available aggregations.
  */

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/MetricsFilterParameter.java

@@ -45,4 +45,8 @@ public String convertValueToString(String value) {
  return value;
  }
 
+ @Override
+ public String getDescription() {
+ return "Comma-separated list of string values to select specific metrics.";
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/SubtasksFilterQueryParameter.java

@@ -38,4 +38,9 @@ public String convertStringToValue(String value) {
  public String convertValueToString(String value) {
  return value;
  }
+
+ @Override
+ public String getDescription() {
+ return "Comma-separated list of integer ranges (e.g. \"1,3,5-9\") to select specific subtasks.";
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/TaskManagersFilterQueryParameter.java

@@ -39,4 +39,9 @@ public ResourceID convertStringToValue(String value) {
  public String convertValueToString(ResourceID value) {
  return value.getResourceIdString();
  }
+
+ @Override
+ public String getDescription() {
+ return "Comma-separated list of 32-character hexadecimal strings to select specific task managers.";
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerIdPathParameter.java

@@ -41,4 +41,9 @@ protected ResourceID convertFromString(String value) {
  protected String convertToString(ResourceID value) {
  return value.getResourceIdString();
  }
+
+ @Override
+ public String getDescription() {
+ return "32-character hexadecimal string that identifies a task manager.";
+ }
 }

flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java

@@ -616,6 +616,11 @@ public JobID convertFromString(String value) {
  protected String convertToString(JobID value) {
  return value.toString();
  }
+
+ @Override
+ public String getDescription() {
+ return "correct JobID parameter";
+ }
  }
 
  static class FaultyJobIDPathParameter extends MessagePathParameter<JobID> {
@@ -633,6 +638,11 @@ protected JobID convertFromString(String value) throws ConversionException {
  protected String convertToString(JobID value) {
  return "foobar";
  }
+
+ @Override
+ public String getDescription() {
+ return "faulty JobID parameter";
+ }
  }
 
  static class JobIDQueryParameter extends MessageQueryParameter<JobID> {
@@ -649,6 +659,11 @@ public JobID convertStringToValue(String value) {
  public String convertValueToString(JobID value) {
  return value.toString();
  }
+
+ @Override
+ public String getDescription() {
+ return "query JobID parameter";
+ }
  }
 
  private static class TestUploadHandler extends AbstractRestHandler<RestfulGateway, EmptyRequestBody, EmptyResponseBody, EmptyMessageParameters> {

flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/util/HandlerRequestUtilsTest.java

@@ -111,5 +111,10 @@ public Boolean convertStringToValue(final String value) {
  public String convertValueToString(final Boolean value) {
  return value.toString();
  }
+
+ @Override
+ public String getDescription() {
+ return "boolean query parameter";
+ }
  }
 }

flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/MessageParametersTest.java

@@ -92,6 +92,11 @@ public JobID convertFromString(String value) {
  protected String convertToString(JobID value) {
  return value.toString();
  }
+
+ @Override
+ public String getDescription() {
+ return "path parameter";
+ }
  }
 
  private static class TestQueryParameter extends MessageQueryParameter<JobID> {
@@ -109,5 +114,10 @@ public JobID convertStringToValue(String value) {
  public String convertValueToString(JobID value) {
  return value.toString();
  }
+
+ @Override
+ public String getDescription() {
+ return "query parameter";
+ }
  }
 }