[FLINK-16696][rest] Properly document async responses

docs/_includes/generated/rest_v1_dispatcher.html

@@ -2256,7 +2256,13 @@
  "id" : "urn:jsonschema:org:apache:flink:runtime:rest:handler:async:AsynchronousOperationResult",
  "properties" : {
  "operation" : {
- "type" : "any"
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:handler:async:AsynchronousOperationInfo",
+ "properties" : {
+ "failure-cause" : {
+ "type" : "any"
+ }
+ }
  },
  "status" : {
  "type" : "object",
@@ -2387,7 +2393,16 @@
  "id" : "urn:jsonschema:org:apache:flink:runtime:rest:handler:async:AsynchronousOperationResult",
  "properties" : {
  "operation" : {
- "type" : "any"
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:job:savepoints:SavepointInfo",
+ "properties" : {
+ "failure-cause" : {
+ "type" : "any"
+ },
+ "location" : {
+ "type" : "string"
+ }
+ }
  },
  "status" : {
  "type" : "object",
@@ -3760,7 +3775,13 @@
  "id" : "urn:jsonschema:org:apache:flink:runtime:rest:handler:async:AsynchronousOperationResult",
  "properties" : {
  "operation" : {
- "type" : "any"
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:handler:async:AsynchronousOperationInfo",
+ "properties" : {
+ "failure-cause" : {
+ "type" : "any"
+ }
+ }
  },
  "status" : {
  "type" : "object",

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

@@ -19,6 +19,8 @@
 package org.apache.flink.docs.rest;
 
 import org.apache.flink.runtime.rest.RestServerEndpoint;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationStatusMessageHeaders;
 import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
 import org.apache.flink.runtime.rest.messages.EmptyResponseBody;
 import org.apache.flink.runtime.rest.messages.MessageHeaders;
@@ -40,6 +42,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
@@ -129,11 +133,17 @@ private static void createHtmlFile(DocumentingRestEndpoint restEndpoint, RestAPI
  }
 
  private static String createHtmlEntry(MessageHeaders<?, ?, ?> spec) {
+ Class<?> nestedAsyncOperationResultClass = null;
+ if (spec instanceof AsynchronousOperationStatusMessageHeaders) {
+ nestedAsyncOperationResultClass = ((AsynchronousOperationStatusMessageHeaders<?, ?>) spec).getValueClass();
+ }
  String requestEntry = createMessageHtmlEntry(
  spec.getRequestClass(),
+ null,
  EmptyRequestBody.class);
  String responseEntry = createMessageHtmlEntry(
  spec.getResponseClass(),
+ nestedAsyncOperationResultClass,
  EmptyResponseBody.class);
 
  String pathParameterList = createPathParameterHtmlList(spec.getUnresolvedMessageParameters().getPathParameters());
@@ -237,13 +247,12 @@ private static String createQueryParameterHtmlList(Collection<MessageQueryParame
  return queryParameterList.toString();
  }
 
- private static String createMessageHtmlEntry(Class<?> messageClass, Class<?> emptyMessageClass) {
- JsonSchema schema;
- try {
- schema = schemaGen.generateSchema(messageClass);
- } catch (JsonProcessingException e) {
- LOG.error("Failed to generate message schema for class {}.", messageClass, e);
- throw new RuntimeException("Failed to generate message schema for class " + messageClass.getCanonicalName() + ".", e);
+ private static String createMessageHtmlEntry(Class<?> messageClass, @Nullable Class<?> nestedAsyncOperationResultClass, Class<?> emptyMessageClass) {
+ JsonSchema schema = generateSchema(messageClass);
+
+ if (nestedAsyncOperationResultClass != null) {
+ JsonSchema innerSchema = generateSchema(nestedAsyncOperationResultClass);
+ schema.asObjectSchema().getProperties().put(AsynchronousOperationResult.FIELD_NAME_OPERATION, innerSchema);
  }
 
  String json;
@@ -262,6 +271,15 @@ private static String createMessageHtmlEntry(Class<?> messageClass, Class<?> emp
  return json;
  }
 
+ private static JsonSchema generateSchema(Class<?> messageClass) {
+ try {
+ return schemaGen.generateSchema(messageClass);
+ } catch (JsonProcessingException e) {
+ LOG.error("Failed to generate message schema for class {}.", messageClass, e);
+ throw new RuntimeException("Failed to generate message schema for class " + messageClass.getCanonicalName() + ".", e);
+ }
+ }
+
  /**
  * Create character escapes for HTML when generating JSON request/response string.
  *

flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/async/AsynchronousOperationResult.java

@@ -36,7 +36,7 @@ public class AsynchronousOperationResult<V> implements AsynchronouslyCreatedReso
 
  private static final String FIELD_NAME_STATUS = "status";
 
- private static final String FIELD_NAME_OPERATION = "operation";
+ public static final String FIELD_NAME_OPERATION = "operation";
 
  @JsonProperty(FIELD_NAME_STATUS)
  private final QueueStatus queueStatus;

flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/async/AsynchronousOperationStatusMessageHeaders.java

@@ -38,7 +38,7 @@ public abstract class AsynchronousOperationStatusMessageHeaders<V, M extends Mes
  *
  * @return value class
  */
- protected abstract Class<V> getValueClass();
+ public abstract Class<V> getValueClass();
 
  @Override
  public Class<AsynchronousOperationResult<V>> getResponseClass() {

flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/rescaling/RescalingStatusHeaders.java

@@ -43,7 +43,7 @@ public class RescalingStatusHeaders extends
  private RescalingStatusHeaders() {}
 
  @Override
- protected Class<AsynchronousOperationInfo> getValueClass() {
+ public Class<AsynchronousOperationInfo> getValueClass() {
  return AsynchronousOperationInfo.class;
  }
 

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

@@ -69,7 +69,7 @@ public static SavepointDisposalStatusHeaders getInstance() {
  }
 
  @Override
- protected Class<AsynchronousOperationInfo> getValueClass() {
+ public Class<AsynchronousOperationInfo> getValueClass() {
  return AsynchronousOperationInfo.class;
  }
 

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

@@ -72,7 +72,7 @@ public static SavepointStatusHeaders getInstance() {
  }
 
  @Override
- protected Class<SavepointInfo> getValueClass() {
+ public Class<SavepointInfo> getValueClass() {
  return SavepointInfo.class;
  }
 

flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/async/AbstractAsynchronousOperationHandlersTest.java

@@ -297,7 +297,7 @@ private static final class TestingStatusMessageHeaders extends AsynchronousOpera
  private TestingStatusMessageHeaders() {}
 
  @Override
- protected Class<OperationResult> getValueClass() {
+ public Class<OperationResult> getValueClass() {
  return OperationResult.class;
  }