Update AggregatedMetricsResponseBody to extend from ArrayList

Author: nikhilsu

docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

@@ -1518,7 +1518,32 @@
           <details>
           <summary>Response</summary>
           <pre><code>{
-  "type" : "any"
+  "type" : "array",
+  "items" : {
+    "type" : "object",
+    "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:job:metrics:AggregatedMetric",
+    "properties" : {
+      "avg" : {
+        "type" : "number"
+      },
+      "id" : {
+        "type" : "string",
+        "required" : true
+      },
+      "max" : {
+        "type" : "number"
+      },
+      "min" : {
+        "type" : "number"
+      },
+      "skew" : {
+        "type" : "number"
+      },
+      "sum" : {
+        "type" : "number"
+      }
+    }
+  }
 }</code></pre>
         </label>
       </td>
@@ -5364,7 +5389,32 @@
           <details>
           <summary>Response</summary>
           <pre><code>{
-  "type" : "any"
+  "type" : "array",
+  "items" : {
+    "type" : "object",
+    "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:job:metrics:AggregatedMetric",
+    "properties" : {
+      "avg" : {
+        "type" : "number"
+      },
+      "id" : {
+        "type" : "string",
+        "required" : true
+      },
+      "max" : {
+        "type" : "number"
+      },
+      "min" : {
+        "type" : "number"
+      },
+      "skew" : {
+        "type" : "number"
+      },
+      "sum" : {
+        "type" : "number"
+      }
+    }
+  }
 }</code></pre>
         </label>
       </td>
@@ -6451,7 +6501,32 @@
           <details>
           <summary>Response</summary>
           <pre><code>{
-  "type" : "any"
+  "type" : "array",
+  "items" : {
+    "type" : "object",
+    "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:job:metrics:AggregatedMetric",
+    "properties" : {
+      "avg" : {
+        "type" : "number"
+      },
+      "id" : {
+        "type" : "string",
+        "required" : true
+      },
+      "max" : {
+        "type" : "number"
+      },
+      "min" : {
+        "type" : "number"
+      },
+      "skew" : {
+        "type" : "number"
+      },
+      "sum" : {
+        "type" : "number"
+      }
+    }
+  }
 }</code></pre>
         </label>
       </td>

docs/static/generated/rest_v1_dispatcher.yml

@@ -502,7 +502,9 @@ paths:
           content:
             application/json:
               schema:
-                $ref: "#/components/schemas/AggregatedMetricsResponseBody"
+                type: array
+                items:
+                  $ref: "#/components/schemas/AggregatedMetric"
   /jobs/overview:
     get:
       description: Returns an overview over all jobs.
@@ -1440,7 +1442,9 @@ paths:
           content:
             application/json:
               schema:
-                $ref: "#/components/schemas/AggregatedMetricsResponseBody"
+                type: array
+                items:
+                  $ref: "#/components/schemas/AggregatedMetric"
   /jobs/{jobid}/vertices/{vertexid}/subtasks/{subtaskindex}:
     get:
       description: Returns details of the current or latest execution attempt of a
@@ -1757,7 +1761,9 @@ paths:
           content:
             application/json:
               schema:
-                $ref: "#/components/schemas/AggregatedMetricsResponseBody"
+                type: array
+                items:
+                  $ref: "#/components/schemas/AggregatedMetric"
   /taskmanagers/{taskmanagerid}:
     get:
       description: Returns details for a task manager.
@@ -1841,8 +1847,32 @@ paths:
                 $ref: "#/components/schemas/ThreadDumpInfo"
 components:
   schemas:
-    AggregatedMetricsResponseBody:
+    AggregatedMetric:
+      required:
+      - id
       type: object
+      properties:
+        avg:
+          type: number
+          format: double
+        id:
+          type: string
+        max:
+          type: number
+          format: double
+        min:
+          type: number
+          format: double
+        skew:
+          type: number
+          format: double
+        sum:
+          type: number
+          format: double
+    AggregatedMetricsResponseBody:
+      type: array
+      items:
+        $ref: "#/components/schemas/AggregatedMetric"
     AggregatedTaskDetailsInfo:
       type: object
       properties:

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

@@ -21,19 +21,11 @@
 import org.apache.flink.runtime.rest.messages.ResponseBody;
 
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.type.TypeReference;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationContext;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.std.StdDeserializer;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.std.StdSerializer;
 
-import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collection;
-import java.util.List;
+
+import static java.util.Objects.requireNonNull;
 
 /**
  * Response type for a collection of aggregated metrics.
@@ -46,62 +38,28 @@
  * [{"id": "metricName", "min": "1"}]
  * }</pre>
  *
- * @see AggregatedMetricsResponseBody.Serializer
- * @see AggregatedMetricsResponseBody.Deserializer
  * @see org.apache.flink.runtime.rest.handler.legacy.metrics.MetricStore
  */
-@JsonSerialize(using = AggregatedMetricsResponseBody.Serializer.class)
-@JsonDeserialize(using = AggregatedMetricsResponseBody.Deserializer.class)
-public class AggregatedMetricsResponseBody implements ResponseBody {
+public class AggregatedMetricsResponseBody extends ArrayList<AggregatedMetric>
+        implements ResponseBody {
 
-    private final Collection<AggregatedMetric> metrics;
+    private static final long serialVersionUID = -1170348873871206965L;
 
-    public AggregatedMetricsResponseBody(Collection<AggregatedMetric> metrics) {
+    // a default constructor is required for collection type marshalling
+    public AggregatedMetricsResponseBody() {}
 
-        this.metrics = metrics;
+    public AggregatedMetricsResponseBody(Collection<AggregatedMetric> metrics) {
+        super(requireNonNull(metrics, "metrics must not be null"));
     }
 
     @JsonIgnore
     public Collection<AggregatedMetric> getMetrics() {
-        return metrics;
-    }
-
-    /** JSON serializer for {@link AggregatedMetricsResponseBody}. */
-    public static class Serializer extends StdSerializer<AggregatedMetricsResponseBody> {
-
-        private static final long serialVersionUID = 1L;
-
-        protected Serializer() {
-            super(AggregatedMetricsResponseBody.class);
-        }
-
-        @Override
-        public void serialize(
-                AggregatedMetricsResponseBody metricCollectionResponseBody,
-                JsonGenerator jsonGenerator,
-                SerializerProvider serializerProvider)
-                throws IOException {
-
-            jsonGenerator.writeObject(metricCollectionResponseBody.getMetrics());
-        }
+        return this;
     }
 
-    /** JSON deserializer for {@link AggregatedMetricsResponseBody}. */
-    public static class Deserializer extends StdDeserializer<AggregatedMetricsResponseBody> {
-
-        private static final long serialVersionUID = 1L;
-
-        protected Deserializer() {
-            super(AggregatedMetricsResponseBody.class);
-        }
-
-        @Override
-        public AggregatedMetricsResponseBody deserialize(
-                JsonParser jsonParser, DeserializationContext deserializationContext)
-                throws IOException {
-
-            return new AggregatedMetricsResponseBody(
-                    jsonParser.readValueAs(new TypeReference<List<AggregatedMetric>>() {}));
-        }
+    @Override
+    @JsonIgnore
+    public boolean isEmpty() {
+        return super.isEmpty();
     }
 }

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

@@ -37,12 +37,14 @@
  * <pre>{@code
  * [{"id": "metricName", "value": "1"}]
  * }</pre>
+ *
+ * @see org.apache.flink.runtime.rest.handler.legacy.metrics.MetricStore
  */
 public final class MetricCollectionResponseBody extends ArrayList<Metric> implements ResponseBody {
 
     private static final long serialVersionUID = -1170348873871206965L;
 
-    // a default constructor is required for collection type marshaling
+    // a default constructor is required for collection type marshalling
     public MetricCollectionResponseBody() {}
 
     public MetricCollectionResponseBody(Collection<Metric> metrics) {