Collector: Add show task api & Improve the built-in Kafka Source plugin

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/task/impl/TaskApiServiceImpl.java

@@ -86,4 +86,11 @@ public Response dropTask(
         ? RuntimeService.task().get().dropTask(dropTaskRequest.getTaskId())
         : Response.serverError().entity("Task runtime is down").build();
   }
+
+  @Override
+  public Response showTask(final SecurityContext securityContext) {
+    return RuntimeService.task().isPresent()
+        ? RuntimeService.task().get().showTask()
+        : Response.serverError().entity("Task runtime is down").build();
+  }
 }

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/persistence/TaskPersistence.java

@@ -40,6 +40,9 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -222,4 +225,34 @@ public Optional<Map<Integer, ProgressIndex>> getTasksProgress(final String taskI
 
     return Optional.empty();
   }
+
+  public List<Map<String, Object>> showTasks() {
+    final List<Map<String, Object>> result = new ArrayList<>();
+    final String queryAllTaskSQL =
+        "SELECT task_id, task_state, source_attribute, processor_attribute, sink_attribute, create_time FROM task";
+
+    try (final Connection connection = getConnection();
+        final PreparedStatement statement = connection.prepareStatement(queryAllTaskSQL);
+        final ResultSet taskResultSet = statement.executeQuery()) {
+      while (taskResultSet.next()) {
+        final Map<String, Object> taskInfo = new LinkedHashMap<>();
+
+        taskInfo.put("taskId", taskResultSet.getString(1));
+        taskInfo.put("taskState", TaskStateEnum.values()[taskResultSet.getInt(2)].name());
+        taskInfo.put("sourceAttribute", SerializationUtil.deserialize(taskResultSet.getBytes(3)));
+        taskInfo.put(
+            "processorAttribute", SerializationUtil.deserialize(taskResultSet.getBytes(4)));
+        taskInfo.put("sinkAttribute", SerializationUtil.deserialize(taskResultSet.getBytes(5)));
+        taskInfo.put("createTime", taskResultSet.getTimestamp(6));
+
+        result.add(taskInfo);
+      }
+
+      return result;
+    } catch (final SQLException e) {
+      LOGGER.warn("Failed to show tasks because {}", e.getMessage());
+    }
+
+    return result;
+  }
 }

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/BuiltinPlugin.java

@@ -25,8 +25,8 @@
 import org.apache.iotdb.collector.plugin.builtin.sink.protocol.IoTDBDataRegionSyncConnector;
 import org.apache.iotdb.collector.plugin.builtin.source.HttpPullSource;
 import org.apache.iotdb.collector.plugin.builtin.source.HttpPushSource;
-import org.apache.iotdb.collector.plugin.builtin.source.IoTDBPushSource;
-import org.apache.iotdb.collector.plugin.builtin.source.KafkaSource;
+import org.apache.iotdb.collector.plugin.builtin.source.iotdb.IoTDBPushSource;
+import org.apache.iotdb.collector.plugin.builtin.source.kafka.KafkaSource;
 
 public enum BuiltinPlugin {
 

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/processor/SubscriptionProcessor.java

@@ -19,7 +19,7 @@
 
 package org.apache.iotdb.collector.plugin.builtin.processor;
 
-import org.apache.iotdb.collector.plugin.builtin.source.event.SubDemoEvent;
+import org.apache.iotdb.collector.plugin.builtin.source.iotdb.SubDemoEvent;
 import org.apache.iotdb.pipe.api.PipeProcessor;
 import org.apache.iotdb.pipe.api.collector.EventCollector;
 import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/constant/KafkaSourceConstant.java → iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/constant/SourceConstant.java

@@ -19,12 +19,9 @@
 
 package org.apache.iotdb.collector.plugin.builtin.source.constant;
 
-public class KafkaSourceConstant {
-  public static final String KAFKA_SOURCE_TOPIC_KEY = "topic";
-  public static final String KAFKA_SOURCE_URL_KEY = "url";
-  public static final String KAFKA_SOURCE_GROUP_ID_KEY = "group-id";
+public class SourceConstant {
 
-  public static final String KAFKA_SOURCE_TOPIC_VALUE = "my_topic";
-  public static final String KAFKA_SOURCE_URL_VALUE = "localhost:9092";
-  public static final String KAFKA_SOURCE_GROUP_ID_VALUE = "multi-thread-group";
+  public static final String REPORT_TIME_INTERVAL_KEY = "report-time-interval";
+
+  public static final String REPORT_TIME_INTERVAL_DEFAULT_VALUE = "60";
 }

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/IoTDBPushSource.java → iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/iotdb/IoTDBPushSource.java

@@ -17,11 +17,9 @@
  * under the License.
  */
 
-package org.apache.iotdb.collector.plugin.builtin.source;
+package org.apache.iotdb.collector.plugin.builtin.source.iotdb;
 
 import org.apache.iotdb.collector.plugin.api.PushSource;
-import org.apache.iotdb.collector.plugin.builtin.source.constant.IoTDBPushSourceConstant;
-import org.apache.iotdb.collector.plugin.builtin.source.event.SubDemoEvent;
 import org.apache.iotdb.collector.runtime.progress.ProgressIndex;
 import org.apache.iotdb.pipe.api.customizer.configuration.PipeSourceRuntimeConfiguration;
 import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/constant/IoTDBPushSourceConstant.java → iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/iotdb/IoTDBPushSourceConstant.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.iotdb.collector.plugin.builtin.source.constant;
+package org.apache.iotdb.collector.plugin.builtin.source.iotdb;
 
 public class IoTDBPushSourceConstant {
   public static final String HOST_KEY = "host";

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/event/SubDemoEvent.java → iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/iotdb/SubDemoEvent.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.iotdb.collector.plugin.builtin.source.event;
+package org.apache.iotdb.collector.plugin.builtin.source.iotdb;
 
 import org.apache.iotdb.collector.plugin.builtin.sink.event.PipeRawTabletInsertionEvent;
 import org.apache.iotdb.collector.plugin.builtin.source.event.common.PipeRowCollector;