Make supervisor API similar to submit task API

docs/tutorials/tutorial-kafka.md

@@ -180,60 +180,62 @@ Paste in this spec and click `Submit`.
 ```json
 {
   "type": "kafka",
-  "dataSchema": {
-    "dataSource": "wikipedia",
-    "parser": {
-      "type": "string",
-      "parseSpec": {
-        "format": "json",
-        "timestampSpec": {
-          "column": "time",
-          "format": "auto"
-        },
-        "dimensionsSpec": {
-          "dimensions": [
-            "channel",
-            "cityName",
-            "comment",
-            "countryIsoCode",
-            "countryName",
-            "isAnonymous",
-            "isMinor",
-            "isNew",
-            "isRobot",
-            "isUnpatrolled",
-            "metroCode",
-            "namespace",
-            "page",
-            "regionIsoCode",
-            "regionName",
-            "user",
-            { "name": "added", "type": "long" },
-            { "name": "deleted", "type": "long" },
-            { "name": "delta", "type": "long" }
-          ]
+  "spec" : {
+    "dataSchema": {
+      "dataSource": "wikipedia",
+      "parser": {
+        "type": "string",
+        "parseSpec": {
+          "format": "json",
+          "timestampSpec": {
+            "column": "time",
+            "format": "auto"
+          },
+          "dimensionsSpec": {
+            "dimensions": [
+              "channel",
+              "cityName",
+              "comment",
+              "countryIsoCode",
+              "countryName",
+              "isAnonymous",
+              "isMinor",
+              "isNew",
+              "isRobot",
+              "isUnpatrolled",
+              "metroCode",
+              "namespace",
+              "page",
+              "regionIsoCode",
+              "regionName",
+              "user",
+              { "name": "added", "type": "long" },
+              { "name": "deleted", "type": "long" },
+              { "name": "delta", "type": "long" }
+            ]
+          }
         }
+      },
+      "metricsSpec" : [],
+      "granularitySpec": {
+        "type": "uniform",
+        "segmentGranularity": "DAY",
+        "queryGranularity": "NONE",
+        "rollup": false
       }
     },
-    "metricsSpec" : [],
-    "granularitySpec": {
-      "type": "uniform",
-      "segmentGranularity": "DAY",
-      "queryGranularity": "NONE",
-      "rollup": false
-    }
-  },
-  "tuningConfig": {
-    "type": "kafka",
-    "reportParseExceptions": false
-  },
-  "ioConfig": {
-    "topic": "wikipedia",
-    "replicas": 2,
-    "taskDuration": "PT10M",
-    "completionTimeout": "PT20M",
-    "consumerProperties": {
-      "bootstrap.servers": "localhost:9092"
+    "tuningConfig": {
+      "type": "kafka",
+      "reportParseExceptions": false
+    },
+    "ioConfig": {
+      "topic": "wikipedia",
+      "replicas": 2,
+      "taskDuration": "PT10M",
+      "completionTimeout": "PT20M",
+      "consumerProperties": {
+        "bootstrap.servers": "localhost:9092"
+      }
     }
   }
 }

examples/quickstart/tutorial/wikipedia-kafka-supervisor.json

@@ -1,59 +1,70 @@
 {
   "type": "kafka",
-  "dataSchema": {
-    "dataSource": "wikipedia",
-    "parser": {
-      "type": "string",
-      "parseSpec": {
-        "format": "json",
-        "timestampSpec": {
-          "column": "time",
-          "format": "auto"
-        },
-        "dimensionsSpec": {
-          "dimensions": [
-            "channel",
-            "cityName",
-            "comment",
-            "countryIsoCode",
-            "countryName",
-            "isAnonymous",
-            "isMinor",
-            "isNew",
-            "isRobot",
-            "isUnpatrolled",
-            "metroCode",
-            "namespace",
-            "page",
-            "regionIsoCode",
-            "regionName",
-            "user",
-            { "name": "added", "type": "long" },
-            { "name": "deleted", "type": "long" },
-            { "name": "delta", "type": "long" }
-          ]
+  "spec" : {
+    "dataSchema": {
+      "dataSource": "wikipedia",
+      "parser": {
+        "type": "string",
+        "parseSpec": {
+          "format": "json",
+          "timestampSpec": {
+            "column": "time",
+            "format": "auto"
+          },
+          "dimensionsSpec": {
+            "dimensions": [
+              "channel",
+              "cityName",
+              "comment",
+              "countryIsoCode",
+              "countryName",
+              "isAnonymous",
+              "isMinor",
+              "isNew",
+              "isRobot",
+              "isUnpatrolled",
+              "metroCode",
+              "namespace",
+              "page",
+              "regionIsoCode",
+              "regionName",
+              "user",
+              {
+                "name": "added",
+                "type": "long"
+              },
+              {
+                "name": "deleted",
+                "type": "long"
+              },
+              {
+                "name": "delta",
+                "type": "long"
+              }
+            ]
+          }
         }
+      },
+      "metricsSpec": [],
+      "granularitySpec": {
+        "type": "uniform",
+        "segmentGranularity": "DAY",
+        "queryGranularity": "NONE",
+        "rollup": false
       }
     },
-    "metricsSpec" : [],
-    "granularitySpec": {
-      "type": "uniform",
-      "segmentGranularity": "DAY",
-      "queryGranularity": "NONE",
-      "rollup": false
-    }
-  },
-  "tuningConfig": {
-    "type": "kafka",
-    "reportParseExceptions": false
-  },
-  "ioConfig": {
-    "topic": "wikipedia",
-    "replicas": 1,
-    "taskDuration": "PT10M",
-    "completionTimeout": "PT20M",
-    "consumerProperties": {
-      "bootstrap.servers": "localhost:9092"
+    "tuningConfig": {
+      "type": "kafka",
+      "reportParseExceptions": false
+    },
+    "ioConfig": {
+      "topic": "wikipedia",
+      "replicas": 1,
+      "taskDuration": "PT10M",
+      "completionTimeout": "PT20M",
+      "consumerProperties": {
+        "bootstrap.servers": "localhost:9092"
+      }
     }
   }
 }

extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorIngestionSpec.java

@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.kafka.supervisor;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorIngestionSpec;
+import org.apache.druid.segment.indexing.DataSchema;
+
+public class KafkaSupervisorIngestionSpec extends SeekableStreamSupervisorIngestionSpec
+{
+  private final DataSchema dataSchema;
+  private final KafkaSupervisorIOConfig ioConfig;
+  private final KafkaSupervisorTuningConfig tuningConfig;
+
+  @JsonCreator
+  public KafkaSupervisorIngestionSpec(
+      @JsonProperty("dataSchema") DataSchema dataSchema,
+      @JsonProperty("ioConfig") KafkaSupervisorIOConfig ioConfig,
+      @JsonProperty("tuningConfig") KafkaSupervisorTuningConfig tuningConfig
+  )
+  {
+    super(dataSchema, ioConfig, tuningConfig);
+    this.dataSchema = dataSchema;
+    this.ioConfig = ioConfig;
+    this.tuningConfig = tuningConfig == null ? KafkaSupervisorTuningConfig.defaultConfig() : tuningConfig;
+  }
+
+  @Override
+  @JsonProperty("dataSchema")
+  public DataSchema getDataSchema()
+  {
+    return dataSchema;
+  }
+
+  @Override
+  @JsonProperty("ioConfig")
+  public KafkaSupervisorIOConfig getIOConfig()
+  {
+    return ioConfig;
+  }
+
+  @Override
+  @JsonProperty("tuningConfig")
+  public KafkaSupervisorTuningConfig getTuningConfig()
+  {
+    return tuningConfig;
+  }
+}