You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by he...@apache.org on 2022/05/16 12:33:39 UTC
[incubator-inlong] branch master updated: [INLONG-4085][Manager] Change the inlong_group and inlong_stream table structure (#4214)
This is an automated email from the ASF dual-hosted git repository.
healchow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 236e5d9ba [INLONG-4085][Manager] Change the inlong_group and inlong_stream table structure (#4214)
236e5d9ba is described below
commit 236e5d9ba97487a19e698cee7198ddcccee283cd
Author: healchow <he...@gmail.com>
AuthorDate: Mon May 16 20:33:34 2022 +0800
[INLONG-4085][Manager] Change the inlong_group and inlong_stream table structure (#4214)
---
.../manager/client/AutoPush2HiveExample.java | 19 +-
.../inlong/manager/client/Binlog2KafkaExample.java | 23 +-
.../inlong/manager/client/File2HiveExample.java | 18 +-
.../inlong/manager/client/Kafka2HiveExample.java | 21 +-
.../inlong/manager/client/cli/pojo/GroupInfo.java | 1 -
.../inlong/manager/client/cli/pojo/StreamInfo.java | 2 +-
inlong-manager/manager-client/pom.xml | 4 +-
.../inlong/manager/client/api/InlongGroupConf.java | 19 +-
.../api/impl/DefaultInlongStreamBuilder.java | 4 +-
.../manager/client/api/impl/InlongStreamImpl.java | 2 +-
.../client/api/util/InlongGroupTransfer.java | 32 +-
.../manager/client/api/util/InlongParser.java | 10 +-
.../client/api/util/InlongStreamTransfer.java | 5 +-
.../inlong/manager/common/enums/Constant.java | 6 -
.../inlong/manager/common/enums/ErrorCodeEnum.java | 2 +-
.../pojo/group/InlongGroupApproveRequest.java | 11 +-
.../manager/common/pojo/group/InlongGroupInfo.java | 36 +-
.../common/pojo/group/InlongGroupListResponse.java | 3 -
.../common/pojo/group/InlongGroupMqExtBase.java | 6 +-
.../common/pojo/group/InlongGroupPageRequest.java | 4 +-
.../common/pojo/group/InlongGroupPulsarInfo.java | 2 +-
.../common/pojo/group/InlongGroupRequest.java | 38 +-
.../common/pojo/group/InlongGroupResponse.java | 51 +--
.../pojo/group/InlongGroupTopicResponse.java | 8 +-
.../inlong/manager/common/pojo/sink/SinkInfo.java | 3 +-
.../common/pojo/stream/InlongStreamInfo.java | 32 +-
.../pojo/stream/InlongStreamListResponse.java | 9 +-
.../pojo/stream/InlongStreamPageRequest.java | 3 -
.../common/pojo/stream/InlongStreamRequest.java | 18 +-
.../common/pojo/stream/InlongStreamResponse.java | 26 +-
.../pojo/stream/InlongStreamTopicResponse.java | 4 +-
.../common/pojo/stream/StreamBriefResponse.java | 3 -
.../manager/common/pojo/stream/StreamField.java | 60 +--
.../manager/dao/entity/InlongGroupEntity.java | 19 +-
.../manager/dao/entity/InlongStreamEntity.java | 12 +-
.../dao/mapper/InlongGroupEntityMapper.java | 16 +-
.../dao/mapper/InlongStreamEntityMapper.java | 22 +-
.../resources/mappers/InlongGroupEntityMapper.xml | 436 +++++----------------
.../resources/mappers/InlongStreamEntityMapper.xml | 403 +++++++++----------
.../resources/mappers/StreamSinkEntityMapper.xml | 3 +-
.../dao/mapper/InlongGroupEntityMapperTest.java | 10 +-
.../service/core/impl/ConsumptionServiceImpl.java | 18 +-
.../service/core/impl/InlongGroupServiceImpl.java | 23 +-
.../service/core/impl/InlongStreamServiceImpl.java | 8 +-
.../core/impl/ThirdPartyClusterServiceImpl.java | 15 +-
.../manager/service/core/mq/BlankMiddleware.java | 2 +-
.../manager/service/core/mq/MiddlewareFactory.java | 3 +-
.../manager/service/core/mq/PulsarMiddleware.java | 4 +-
.../manager/service/core/mq/TubeMiddleware.java | 6 +-
.../mq/CreatePulsarGroupForStreamTaskListener.java | 8 +-
.../service/mq/CreatePulsarGroupTaskListener.java | 6 +-
.../mq/CreatePulsarResourceTaskListener.java | 6 +-
.../mq/CreatePulsarTopicForStreamTaskListener.java | 6 +-
.../service/mq/CreateTubeGroupTaskListener.java | 2 +-
.../service/mq/CreateTubeTopicTaskListener.java | 2 +-
.../manager/service/mq/PulsarEventSelector.java | 2 +-
.../manager/service/mq/TubeEventSelector.java | 2 +-
.../service/sort/ZookeeperDisabledSelector.java | 4 +-
.../service/sort/ZookeeperEnabledSelector.java | 2 +-
.../manager/service/sort/util/DataFlowUtils.java | 2 +-
.../service/sort/util/NodeRelationShipUtils.java | 23 +-
.../manager/service/sort/util/SourceInfoUtils.java | 8 +-
.../listener/AbstractSourceOperateListener.java | 3 +-
.../ConsumptionCompleteProcessListener.java | 6 +-
.../stream/CreateStreamWorkflowDefinition.java | 12 +-
.../core/impl/InlongGroupProcessOperationTest.java | 2 +-
.../service/core/impl/InlongGroupServiceTest.java | 4 +-
.../manager/service/sort/DisableZkForSortTest.java | 4 +-
.../workflow/ServiceTaskListenerFactoryTest.java | 4 +-
.../service/workflow/WorkflowServiceImplTest.java | 31 +-
.../main/resources/sql/apache_inlong_manager.sql | 279 +++++--------
.../manager-web/sql/apache_inlong_manager.sql | 271 +++++--------
72 files changed, 863 insertions(+), 1311 deletions(-)
diff --git a/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/AutoPush2HiveExample.java b/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/AutoPush2HiveExample.java
index 88339ad2c..cd594a8ee 100644
--- a/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/AutoPush2HiveExample.java
+++ b/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/AutoPush2HiveExample.java
@@ -28,13 +28,13 @@ import org.apache.inlong.manager.client.api.InlongGroupContext;
import org.apache.inlong.manager.client.api.InlongStreamBuilder;
import org.apache.inlong.manager.client.api.InlongStreamConf;
import org.apache.inlong.manager.client.api.PulsarBaseConf;
-import org.apache.inlong.manager.common.pojo.stream.SinkField;
-import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.manager.client.api.auth.DefaultAuthentication;
import org.apache.inlong.manager.client.api.sink.HiveSink;
import org.apache.inlong.manager.client.api.source.AutoPushSource;
import org.apache.inlong.manager.common.enums.FieldType;
import org.apache.inlong.manager.common.enums.FileFormat;
+import org.apache.inlong.manager.common.pojo.stream.SinkField;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.shiro.util.Assert;
import org.junit.Test;
@@ -118,8 +118,8 @@ public class AutoPush2HiveExample {
InlongGroupConf inlongGroupConf = new InlongGroupConf();
inlongGroupConf.setGroupName(GROUP_NAME);
inlongGroupConf.setDescription(GROUP_NAME);
- inlongGroupConf.setProxyClusterId(1);
- //pulsar conf
+
+ // pulsar conf
PulsarBaseConf pulsarBaseConf = new PulsarBaseConf();
inlongGroupConf.setMqBaseConf(pulsarBaseConf);
pulsarBaseConf.setPulsarServiceUrl(PULSAR_SERVICE_URL);
@@ -128,14 +128,19 @@ public class AutoPush2HiveExample {
pulsarBaseConf.setEnableCreateResource(false);
pulsarBaseConf.setTenant(tenant);
- //flink conf
+ // flink conf
FlinkSortBaseConf sortBaseConf = new FlinkSortBaseConf();
inlongGroupConf.setSortBaseConf(sortBaseConf);
sortBaseConf.setServiceUrl(FLINK_URL);
Map<String, String> map = new HashMap<>(16);
sortBaseConf.setProperties(map);
- //enable zk
- inlongGroupConf.setZookeeperEnabled(false);
+
+ // set enable zk, create resource, lightweight mode, and cluster tag
+ inlongGroupConf.setEnableZookeeper(0);
+ inlongGroupConf.setEnableCreateResource(1);
+ inlongGroupConf.setLightweight(1);
+ inlongGroupConf.setInlongClusterTag("default_cluster");
+
inlongGroupConf.setDailyRecords(10000000L);
inlongGroupConf.setPeakRecords(100000L);
inlongGroupConf.setMaxLength(10000);
diff --git a/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/Binlog2KafkaExample.java b/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/Binlog2KafkaExample.java
index 2533135fd..ec340086d 100644
--- a/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/Binlog2KafkaExample.java
+++ b/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/Binlog2KafkaExample.java
@@ -19,7 +19,6 @@ package org.apache.inlong.manager.client;
import lombok.extern.slf4j.Slf4j;
import org.apache.inlong.manager.client.api.ClientConfiguration;
-import org.apache.inlong.manager.common.enums.DataFormat;
import org.apache.inlong.manager.client.api.DataSeparator;
import org.apache.inlong.manager.client.api.FlinkSortBaseConf;
import org.apache.inlong.manager.client.api.InlongClient;
@@ -32,12 +31,13 @@ import org.apache.inlong.manager.client.api.PulsarBaseConf;
import org.apache.inlong.manager.client.api.auth.DefaultAuthentication;
import org.apache.inlong.manager.client.api.sink.KafkaSink;
import org.apache.inlong.manager.client.api.source.MySQLBinlogSource;
+import org.apache.inlong.manager.common.enums.DataFormat;
import org.apache.inlong.manager.common.enums.MQType;
import org.apache.shiro.util.Assert;
import org.junit.Test;
import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
@@ -153,8 +153,7 @@ public class Binlog2KafkaExample {
InlongGroupConf inlongGroupConf = new InlongGroupConf();
inlongGroupConf.setGroupName(GROUP_NAME);
inlongGroupConf.setDescription(GROUP_NAME);
- inlongGroupConf.setProxyClusterId(1);
- //pulsar conf
+ // pulsar conf
PulsarBaseConf pulsarBaseConf = new PulsarBaseConf();
pulsarBaseConf.setType(MQType.PULSAR);
inlongGroupConf.setMqBaseConf(pulsarBaseConf);
@@ -163,14 +162,20 @@ public class Binlog2KafkaExample {
pulsarBaseConf.setNamespace("public");
pulsarBaseConf.setEnableCreateResource(false);
pulsarBaseConf.setTenant(tenant);
- //flink conf
+
+ // flink conf
FlinkSortBaseConf sortBaseConf = new FlinkSortBaseConf();
inlongGroupConf.setSortBaseConf(sortBaseConf);
sortBaseConf.setServiceUrl(FLINK_URL);
Map<String, String> map = new HashMap<>(16);
sortBaseConf.setProperties(map);
- //enable zk
- inlongGroupConf.setZookeeperEnabled(false);
+
+ // set enable zk, create resource, lightweight mode, and cluster tag
+ inlongGroupConf.setEnableZookeeper(0);
+ inlongGroupConf.setEnableCreateResource(1);
+ inlongGroupConf.setLightweight(1);
+ inlongGroupConf.setInlongClusterTag("default_cluster");
+
inlongGroupConf.setDailyRecords(10000000L);
inlongGroupConf.setPeakRecords(100000L);
inlongGroupConf.setMaxLength(10000);
@@ -191,7 +196,7 @@ public class Binlog2KafkaExample {
private MySQLBinlogSource createMysqlSource() {
MySQLBinlogSource mySQLBinlogSource = new MySQLBinlogSource();
- mySQLBinlogSource.setDbNames(Arrays.asList("{db.name}"));
+ mySQLBinlogSource.setDbNames(Collections.singletonList("{db.name}"));
mySQLBinlogSource.setHostname("{db.url}");
mySQLBinlogSource.setAuthentication(new DefaultAuthentication("root", "inlong"));
mySQLBinlogSource.setSourceName("{mysql.source.name}");
@@ -207,7 +212,7 @@ public class Binlog2KafkaExample {
kafkaSink.setNeedCreated(false);
kafkaSink.setSinkName("{kafka.sink.name}");
Map<String, Object> properties = new HashMap<>();
- //Not needed if kafka cluster is not set
+ // Not needed if kafka cluster is not set
properties.put("transaction.timeout.ms", 9000000);
kafkaSink.setProperties(properties);
return kafkaSink;
diff --git a/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/File2HiveExample.java b/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/File2HiveExample.java
index cf079271c..83187e3af 100644
--- a/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/File2HiveExample.java
+++ b/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/File2HiveExample.java
@@ -28,13 +28,13 @@ import org.apache.inlong.manager.client.api.InlongGroupContext;
import org.apache.inlong.manager.client.api.InlongStreamBuilder;
import org.apache.inlong.manager.client.api.InlongStreamConf;
import org.apache.inlong.manager.client.api.PulsarBaseConf;
-import org.apache.inlong.manager.common.pojo.stream.SinkField;
-import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.manager.client.api.auth.DefaultAuthentication;
import org.apache.inlong.manager.client.api.sink.HiveSink;
import org.apache.inlong.manager.client.api.source.AgentFileSource;
import org.apache.inlong.manager.common.enums.FieldType;
import org.apache.inlong.manager.common.enums.FileFormat;
+import org.apache.inlong.manager.common.pojo.stream.SinkField;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.shiro.util.Assert;
import org.junit.Test;
@@ -118,8 +118,7 @@ public class File2HiveExample {
InlongGroupConf inlongGroupConf = new InlongGroupConf();
inlongGroupConf.setGroupName(GROUP_NAME);
inlongGroupConf.setDescription(GROUP_NAME);
- inlongGroupConf.setProxyClusterId(1);
- //pulsar conf
+ // pulsar conf
PulsarBaseConf pulsarBaseConf = new PulsarBaseConf();
inlongGroupConf.setMqBaseConf(pulsarBaseConf);
pulsarBaseConf.setPulsarServiceUrl(PULSAR_SERVICE_URL);
@@ -128,14 +127,19 @@ public class File2HiveExample {
pulsarBaseConf.setEnableCreateResource(false);
pulsarBaseConf.setTenant(tenant);
- //flink conf
+ // flink conf
FlinkSortBaseConf sortBaseConf = new FlinkSortBaseConf();
inlongGroupConf.setSortBaseConf(sortBaseConf);
sortBaseConf.setServiceUrl(FLINK_URL);
Map<String, String> map = new HashMap<>(16);
sortBaseConf.setProperties(map);
- //enable zk
- inlongGroupConf.setZookeeperEnabled(false);
+
+ // set enable zk, create resource, lightweight mode, and cluster tag
+ inlongGroupConf.setEnableZookeeper(0);
+ inlongGroupConf.setEnableCreateResource(1);
+ inlongGroupConf.setLightweight(1);
+ inlongGroupConf.setInlongClusterTag("default_cluster");
+
inlongGroupConf.setDailyRecords(10000000L);
inlongGroupConf.setPeakRecords(100000L);
inlongGroupConf.setMaxLength(10000);
diff --git a/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/Kafka2HiveExample.java b/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/Kafka2HiveExample.java
index f0e1a254f..a3d9b4fc9 100644
--- a/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/Kafka2HiveExample.java
+++ b/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/Kafka2HiveExample.java
@@ -20,7 +20,6 @@ package org.apache.inlong.manager.client;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.inlong.manager.client.api.ClientConfiguration;
-import org.apache.inlong.manager.common.enums.DataFormat;
import org.apache.inlong.manager.client.api.DataSeparator;
import org.apache.inlong.manager.client.api.FlinkSortBaseConf;
import org.apache.inlong.manager.client.api.InlongClient;
@@ -30,13 +29,14 @@ import org.apache.inlong.manager.client.api.InlongGroupContext;
import org.apache.inlong.manager.client.api.InlongStreamBuilder;
import org.apache.inlong.manager.client.api.InlongStreamConf;
import org.apache.inlong.manager.client.api.PulsarBaseConf;
-import org.apache.inlong.manager.common.pojo.stream.SinkField;
-import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.inlong.manager.client.api.auth.DefaultAuthentication;
import org.apache.inlong.manager.client.api.sink.HiveSink;
import org.apache.inlong.manager.client.api.source.KafkaSource;
+import org.apache.inlong.manager.common.enums.DataFormat;
import org.apache.inlong.manager.common.enums.FieldType;
import org.apache.inlong.manager.common.enums.FileFormat;
+import org.apache.inlong.manager.common.pojo.stream.SinkField;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
import org.apache.shiro.util.Assert;
import org.junit.Test;
@@ -121,8 +121,8 @@ public class Kafka2HiveExample {
InlongGroupConf inlongGroupConf = new InlongGroupConf();
inlongGroupConf.setGroupName(GROUP_NAME);
inlongGroupConf.setDescription(GROUP_NAME);
- inlongGroupConf.setProxyClusterId(1);
- //pulsar conf
+
+ // pulsar conf
PulsarBaseConf pulsarBaseConf = new PulsarBaseConf();
inlongGroupConf.setMqBaseConf(pulsarBaseConf);
pulsarBaseConf.setPulsarServiceUrl(PULSAR_SERVICE_URL);
@@ -131,14 +131,19 @@ public class Kafka2HiveExample {
pulsarBaseConf.setEnableCreateResource(false);
pulsarBaseConf.setTenant(tenant);
- //flink conf
+ // flink conf
FlinkSortBaseConf sortBaseConf = new FlinkSortBaseConf();
inlongGroupConf.setSortBaseConf(sortBaseConf);
sortBaseConf.setServiceUrl(FLINK_URL);
Map<String, String> map = new HashMap<>(16);
sortBaseConf.setProperties(map);
- //enable zk
- inlongGroupConf.setZookeeperEnabled(false);
+
+ // set enable zk, create resource, lightweight mode, and cluster tag
+ inlongGroupConf.setEnableZookeeper(0);
+ inlongGroupConf.setEnableCreateResource(1);
+ inlongGroupConf.setLightweight(1);
+ inlongGroupConf.setInlongClusterTag("default_cluster");
+
inlongGroupConf.setDailyRecords(10000000L);
inlongGroupConf.setPeakRecords(100000L);
inlongGroupConf.setMaxLength(10000);
diff --git a/inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/pojo/GroupInfo.java b/inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/pojo/GroupInfo.java
index 37db1666d..c5e633e49 100644
--- a/inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/pojo/GroupInfo.java
+++ b/inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/pojo/GroupInfo.java
@@ -31,7 +31,6 @@ public class GroupInfo {
private Integer id;
private String inlongGroupId;
private String name;
- private String cnName;
private String status;
private Date modifyTime;
diff --git a/inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/pojo/StreamInfo.java b/inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/pojo/StreamInfo.java
index 5a59950af..fc1c28201 100644
--- a/inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/pojo/StreamInfo.java
+++ b/inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/pojo/StreamInfo.java
@@ -29,7 +29,7 @@ public class StreamInfo {
private Integer id;
private String name;
- private String mqResourceObj;
+ private String mqResource;
private String dataType;
private String dataEncoding;
private String dataSeparator;
diff --git a/inlong-manager/manager-client/pom.xml b/inlong-manager/manager-client/pom.xml
index 0b893d37e..8f56edf78 100644
--- a/inlong-manager/manager-client/pom.xml
+++ b/inlong-manager/manager-client/pom.xml
@@ -15,8 +15,8 @@
~ See the License for the specific language governing permissions and
~ limitations under the License.
-->
-<project xmlns="http://maven.apache.org/POM/4.0.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongGroupConf.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongGroupConf.java
index e96174cc5..2664d1bbe 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongGroupConf.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongGroupConf.java
@@ -35,9 +35,6 @@ public class InlongGroupConf {
@ApiModelProperty(value = "Group name", required = true)
private String groupName;
- @ApiModelProperty("Chinese display name")
- private String cnName;
-
@ApiModelProperty("Group description")
private String description;
@@ -59,12 +56,16 @@ public class InlongGroupConf {
@ApiModelProperty("The operator of stream group, default : admin")
private String operator = "admin";
- @ApiModelProperty("Need zookeeper support")
- private boolean zookeeperEnabled = true;
+ @ApiModelProperty(value = "Whether to enable zookeeper? 0: disable, 1: enable")
+ private Integer enableZookeeper;
+
+ @ApiModelProperty(value = "Whether to enable zookeeper? 0: disable, 1: enable")
+ private Integer enableCreateResource;
+
+ @ApiModelProperty(value = "Whether to use lightweight mode, 0: false, 1: true")
+ private Integer lightweight;
- @ApiModelProperty("Data proxy cluster id")
- private Integer proxyClusterId;
+ @ApiModelProperty("Inlong cluster tag")
+ private String inlongClusterTag;
- @ApiModelProperty("Use lightweight group")
- private boolean lightweight = false;
}
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/DefaultInlongStreamBuilder.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/DefaultInlongStreamBuilder.java
index 3fcbe2d30..565ac0745 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/DefaultInlongStreamBuilder.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/DefaultInlongStreamBuilder.java
@@ -120,7 +120,7 @@ public class DefaultInlongStreamBuilder extends InlongStreamBuilder {
public InlongStream init() {
InlongStreamInfo streamInfo = streamContext.getStreamInfo();
StreamPipeline streamPipeline = inlongStream.createPipeline();
- streamInfo.setTempView(GsonUtil.toJson(streamPipeline));
+ streamInfo.setExtParams(GsonUtil.toJson(streamPipeline));
Double streamIndex = managerClient.createStreamInfo(streamInfo);
streamInfo.setId(streamIndex.intValue());
//Create source and update index
@@ -145,7 +145,7 @@ public class DefaultInlongStreamBuilder extends InlongStreamBuilder {
public InlongStream initOrUpdate() {
InlongStreamInfo dataStreamInfo = streamContext.getStreamInfo();
StreamPipeline streamPipeline = inlongStream.createPipeline();
- dataStreamInfo.setTempView(GsonUtil.toJson(streamPipeline));
+ dataStreamInfo.setExtParams(GsonUtil.toJson(streamPipeline));
Boolean isExist = managerClient.isStreamExists(dataStreamInfo);
if (isExist) {
Pair<Boolean, String> updateMsg = managerClient.updateStreamInfo(dataStreamInfo);
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongStreamImpl.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongStreamImpl.java
index ff2d5e312..eb2c99565 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongStreamImpl.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongStreamImpl.java
@@ -306,7 +306,7 @@ public class InlongStreamImpl extends InlongStream {
}
streamInfo.setFieldList(InlongStreamTransfer.createStreamFields(this.streamFields, streamInfo));
StreamPipeline streamPipeline = createPipeline();
- streamInfo.setTempView(GsonUtil.toJson(streamPipeline));
+ streamInfo.setExtParams(GsonUtil.toJson(streamPipeline));
Pair<Boolean, String> updateMsg = managerClient.updateStreamInfo(streamInfo);
if (!updateMsg.getKey()) {
throw new RuntimeException(String.format("Update data stream failed:%s", updateMsg.getValue()));
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongGroupTransfer.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongGroupTransfer.java
index 2930ae512..c6bb1cc02 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongGroupTransfer.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongGroupTransfer.java
@@ -55,14 +55,13 @@ public class InlongGroupTransfer {
InlongGroupConf inlongGroupConf = new InlongGroupConf();
inlongGroupConf.setGroupName(inlongGroupResponse.getName());
inlongGroupConf.setDescription(inlongGroupResponse.getDescription());
- inlongGroupConf.setCnName(inlongGroupResponse.getCnName());
- inlongGroupConf.setLightweight(isLightGroup(inlongGroupResponse));
- inlongGroupConf.setZookeeperEnabled(inlongGroupResponse.getZookeeperEnabled() == 1);
+ inlongGroupConf.setLightweight(isLightGroup(inlongGroupResponse) ? 1 : 0);
+ inlongGroupConf.setEnableZookeeper(inlongGroupResponse.getEnableZookeeper());
inlongGroupConf.setDailyRecords(Long.valueOf(inlongGroupResponse.getDailyRecords()));
inlongGroupConf.setPeakRecords(Long.valueOf(inlongGroupResponse.getPeakRecords()));
inlongGroupConf.setMqBaseConf(parseMqBaseConf(inlongGroupResponse));
inlongGroupConf.setSortBaseConf(parseSortBaseConf(inlongGroupResponse));
- inlongGroupConf.setProxyClusterId(inlongGroupResponse.getProxyClusterId());
+ inlongGroupConf.setInlongClusterTag(inlongGroupResponse.getInlongClusterTag());
return inlongGroupConf;
}
@@ -82,10 +81,10 @@ public class InlongGroupTransfer {
public static MQBaseConf parseMqBaseConf(InlongGroupResponse inlongGroupResponse) {
InlongGroupMqExtBase mqExtBase = inlongGroupResponse.getMqExtInfo();
- if (null == mqExtBase || StringUtils.isBlank(mqExtBase.getMiddlewareType())) {
+ if (null == mqExtBase || StringUtils.isBlank(mqExtBase.getMqType())) {
return null;
}
- String middleWare = mqExtBase.getMiddlewareType();
+ String middleWare = mqExtBase.getMqType();
MQType mqType = MQType.forType(middleWare);
switch (mqType) {
case NONE:
@@ -122,7 +121,7 @@ public class InlongGroupTransfer {
case USER_DEFINED:
return parseUdf(groupExtInfos);
default:
- throw new IllegalArgumentException(String.format("Unsupport sort type=%s for Inlong", sortType));
+ throw new IllegalArgumentException(String.format("Unsupported sort type=%s for Inlong", sortType));
}
}
@@ -160,7 +159,7 @@ public class InlongGroupTransfer {
private static PulsarBaseConf parsePulsarConf(InlongGroupResponse groupResponse) {
PulsarBaseConf pulsarBaseConf = new PulsarBaseConf();
- pulsarBaseConf.setNamespace(groupResponse.getMqResourceObj());
+ pulsarBaseConf.setNamespace(groupResponse.getMqResource());
InlongGroupPulsarInfo pulsarInfo = (InlongGroupPulsarInfo) groupResponse.getMqExtInfo();
pulsarBaseConf.setAckQuorum(pulsarInfo.getAckQuorum());
pulsarBaseConf.setWriteQuorum(pulsarInfo.getWriteQuorum());
@@ -187,7 +186,7 @@ public class InlongGroupTransfer {
private static TubeBaseConf parseTubeConf(InlongGroupResponse groupResponse) {
TubeBaseConf tubeBaseConf = new TubeBaseConf();
- tubeBaseConf.setGroupName(groupResponse.getMqResourceObj());
+ tubeBaseConf.setGroupName(groupResponse.getMqResource());
List<InlongGroupExtInfo> groupExtInfos = groupResponse.getExtList();
for (InlongGroupExtInfo extInfo : groupExtInfos) {
if (extInfo.getKeyName().equals(InlongGroupSettings.TUBE_CLUSTER_ID)) {
@@ -207,19 +206,18 @@ public class InlongGroupTransfer {
InlongGroupInfo groupInfo = new InlongGroupInfo();
AssertUtil.hasLength(groupConf.getGroupName(), "GroupName should not be empty");
groupInfo.setName(groupConf.getGroupName());
- groupInfo.setCnName(groupConf.getCnName());
groupInfo.setInlongGroupId("b_" + groupConf.getGroupName());
groupInfo.setDescription(groupConf.getDescription());
- groupInfo.setZookeeperEnabled(groupConf.isZookeeperEnabled() ? 1 : 0);
+ groupInfo.setEnableZookeeper(groupConf.getEnableZookeeper());
groupInfo.setDailyRecords(groupConf.getDailyRecords().intValue());
groupInfo.setPeakRecords(groupConf.getPeakRecords().intValue());
groupInfo.setMaxLength(groupConf.getMaxLength());
- groupInfo.setProxyClusterId(groupConf.getProxyClusterId());
+ groupInfo.setInlongClusterTag(groupConf.getInlongClusterTag());
MQBaseConf mqConf = groupConf.getMqBaseConf();
MQType mqType = MQType.NONE;
if (null != mqConf) {
mqType = mqConf.getType();
- groupInfo.setMiddlewareType(mqType.name());
+ groupInfo.setMqType(mqType.name());
}
groupInfo.setInCharges(groupConf.getOperator());
groupInfo.setExtList(Lists.newArrayList());
@@ -227,7 +225,7 @@ public class InlongGroupTransfer {
setGroupMode(groupConf, groupInfo);
if (mqType == MQType.PULSAR || mqType == MQType.TDMQ_PULSAR) {
PulsarBaseConf pulsarBaseConf = (PulsarBaseConf) mqConf;
- groupInfo.setMqResourceObj(pulsarBaseConf.getNamespace());
+ groupInfo.setMqResource(pulsarBaseConf.getNamespace());
InlongGroupPulsarInfo pulsarInfo = createPulsarInfo(pulsarBaseConf);
groupInfo.setMqExtInfo(pulsarInfo);
List<InlongGroupExtInfo> extInfos = createPulsarExtInfo(pulsarBaseConf);
@@ -236,7 +234,7 @@ public class InlongGroupTransfer {
} else if (mqType == MQType.TUBE) {
TubeBaseConf tubeBaseConf = (TubeBaseConf) mqConf;
List<InlongGroupExtInfo> extInfos = createTubeExtInfo(tubeBaseConf);
- groupInfo.setMqResourceObj(tubeBaseConf.getGroupName());
+ groupInfo.setMqResource(tubeBaseConf.getGroupName());
groupInfo.getExtList().addAll(extInfos);
groupInfo.setTopicPartitionNum(tubeBaseConf.getTopicPartitionNum());
}
@@ -261,7 +259,7 @@ public class InlongGroupTransfer {
InlongGroupExtInfo modeInfo = new InlongGroupExtInfo();
modeInfo.setInlongGroupId(inlongGroupId);
modeInfo.setKeyName(InlongGroupSettings.GROUP_MODE);
- if (inlongGroupConf.isLightweight()) {
+ if (inlongGroupConf.getLightweight() == 1) {
modeInfo.setKeyValue(GroupMode.LIGHT.getMode());
} else {
modeInfo.setKeyValue(GroupMode.NORMAL.getMode());
@@ -271,7 +269,7 @@ public class InlongGroupTransfer {
public static InlongGroupPulsarInfo createPulsarInfo(PulsarBaseConf pulsarBaseConf) {
InlongGroupPulsarInfo pulsarInfo = new InlongGroupPulsarInfo();
- pulsarInfo.setMiddlewareType(pulsarBaseConf.getType().name());
+ pulsarInfo.setMqType(pulsarBaseConf.getType().name());
pulsarInfo.setEnsemble(pulsarBaseConf.getEnsemble());
pulsarInfo.setAckQuorum(pulsarBaseConf.getAckQuorum());
pulsarInfo.setEnableCreateResource(pulsarBaseConf.isEnableCreateResource() ? 1 : 0);
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongParser.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongParser.java
index eb547dacc..a3d319726 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongParser.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongParser.java
@@ -66,7 +66,7 @@ public class InlongParser {
public static final String GROUP_INFO = "groupInfo";
public static final String MQ_EXT_INFO = "mqExtInfo";
- public static final String MIDDLEWARE_TYPE = "middlewareType";
+ public static final String MQ_TYPE = "mqType";
public static final String SINK_INFO = "sinkInfo";
public static final String SOURCE_INFO = "sourceInfo";
public static final String SINK_TYPE = "sinkType";
@@ -89,8 +89,8 @@ public class InlongParser {
JsonObject groupJson = GsonUtil.fromJson(GsonUtil.toJson(data), JsonObject.class);
InlongGroupResponse inlongGroupResponse = GsonUtil.fromJson(GsonUtil.toJson(data), InlongGroupResponse.class);
JsonObject mqExtInfo = groupJson.getAsJsonObject(MQ_EXT_INFO);
- if (mqExtInfo != null && mqExtInfo.get(MIDDLEWARE_TYPE) != null) {
- MQType mqType = MQType.forType(mqExtInfo.get(MIDDLEWARE_TYPE).getAsString());
+ if (mqExtInfo != null && mqExtInfo.get(MQ_TYPE) != null) {
+ MQType mqType = MQType.forType(mqExtInfo.get(MQ_TYPE).getAsString());
if (mqType == MQType.PULSAR || mqType == MQType.TDMQ_PULSAR) {
InlongGroupPulsarInfo pulsarInfo = GsonUtil.fromJson(mqExtInfo.toString(), InlongGroupPulsarInfo.class);
inlongGroupResponse.setMqExtInfo(pulsarInfo);
@@ -250,8 +250,8 @@ public class InlongParser {
InlongGroupApproveRequest groupApproveInfo = GsonUtil.fromJson(groupJson.toString(),
InlongGroupApproveRequest.class);
JsonObject mqExtInfo = groupJson.getAsJsonObject(MQ_EXT_INFO);
- if (mqExtInfo != null && mqExtInfo.get(MIDDLEWARE_TYPE) != null) {
- MQType mqType = MQType.forType(mqExtInfo.get(MIDDLEWARE_TYPE).getAsString());
+ if (mqExtInfo != null && mqExtInfo.get(MQ_TYPE) != null) {
+ MQType mqType = MQType.forType(mqExtInfo.get(MQ_TYPE).getAsString());
if (mqType == MQType.PULSAR || mqType == MQType.TDMQ_PULSAR) {
InlongGroupPulsarInfo pulsarInfo = GsonUtil.fromJson(mqExtInfo.toString(),
InlongGroupPulsarInfo.class);
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamTransfer.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamTransfer.java
index ed4192927..78a72c322 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamTransfer.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/util/InlongStreamTransfer.java
@@ -40,9 +40,9 @@ public class InlongStreamTransfer {
dataStreamInfo.setName(streamConf.getName());
dataStreamInfo.setDataEncoding(streamConf.getCharset().name());
if (StringUtils.isEmpty(streamConf.getTopic())) {
- dataStreamInfo.setMqResourceObj(streamId);
+ dataStreamInfo.setMqResource(streamId);
} else {
- dataStreamInfo.setMqResourceObj(streamConf.getTopic());
+ dataStreamInfo.setMqResource(streamConf.getTopic());
}
dataStreamInfo.setSyncSend(streamConf.isStrictlyOrdered() ? 1 : 0);
dataStreamInfo.setDataSeparator(String.valueOf(streamConf.getDataSeparator().getAsciiCode()));
@@ -51,7 +51,6 @@ public class InlongStreamTransfer {
dataStreamInfo.setDailyRecords(streamConf.getDailyRecords());
dataStreamInfo.setDailyStorage(streamConf.getDailyStorage());
dataStreamInfo.setPeakRecords(streamConf.getPeakRecords());
- dataStreamInfo.setHavePredefinedFields(0);
if (CollectionUtils.isNotEmpty(streamConf.getStreamFields())) {
dataStreamInfo.setFieldList(createStreamFields(streamConf.getStreamFields(), dataStreamInfo));
}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/Constant.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/Constant.java
index d10093326..ae1bdd401 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/Constant.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/Constant.java
@@ -28,16 +28,10 @@ public class Constant {
public static final String DATA_SOURCE_DB = "DB";
- public static final String DATA_SOURCE_FILE = "FILE";
-
- public static final String DATA_SOURCE_AUTO_PUSH = "AUTO_PUSH";
-
public static final String DATA_TYPE_TEXT = "TEXT";
public static final String DATA_TYPE_KEY_VALUE = "KEY-VALUE";
- public static final String SCHEMA_M0_DAY = "m0_day";
-
public static final String CLUSTER_TUBE = "TUBE";
public static final String CLUSTER_PULSAR = "PULSAR";
public static final String CLUSTER_TDMQ_PULSAR = "TDMQ_PULSAR";
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ErrorCodeEnum.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ErrorCodeEnum.java
index 3ec8491a2..8a6a8be3a 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ErrorCodeEnum.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ErrorCodeEnum.java
@@ -48,7 +48,7 @@ public enum ErrorCodeEnum {
OPT_NOT_ALLOWED_BY_STATUS(1021,
"The current inlong group status does not allow adding/modifying/deleting related info"),
- MIDDLEWARE_TYPE_NOT_SUPPORTED(1021, "MIDDLEWARE_TYPE_NOT_SUPPORTED"),
+ MQ_TYPE_NOT_SUPPORTED(1021, "MIDDLEWARE_TYPE_NOT_SUPPORTED"),
CLUSTER_NOT_FOUND(1101, "Cluster information does not exist"),
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupApproveRequest.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupApproveRequest.java
index f447518cd..3f6845127 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupApproveRequest.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupApproveRequest.java
@@ -34,14 +34,11 @@ public class InlongGroupApproveRequest {
@ApiModelProperty(value = "Inlong group id", required = true)
private String inlongGroupId;
- @ApiModelProperty(value = "Middleware Type")
- private String middlewareType;
+ @ApiModelProperty(value = "MQ Type")
+ private String mqType;
- @ApiModelProperty(value = "MQ resource object, for Tube, it's Topic")
- private String mqResourceObj;
-
- @ApiModelProperty(value = "Data type name")
- private String schemaName;
+ @ApiModelProperty(value = "MQ resource, for Tube, it is Topic, for Pulsar, it is Namespace")
+ private String mqResource;
@ApiModelProperty(value = "The partition num of Pulsar topic, between 1-20")
private Integer topicPartitionNum;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupInfo.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupInfo.java
index 70e58ad0c..8519a54d5 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupInfo.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupInfo.java
@@ -49,15 +49,12 @@ public class InlongGroupInfo {
@ApiModelProperty(value = "Inlong group name", required = true)
private String name;
- @ApiModelProperty(value = "Chinese display name")
- private String cnName;
-
@ApiModelProperty(value = "Inlong group description")
private String description;
- @NotNull(message = "middlewareType cannot be null")
- @ApiModelProperty(value = "Middleware type, high throughput: TUBE, high consistency: PULSAR")
- private String middlewareType;
+ @NotNull(message = "mqType cannot be null")
+ @ApiModelProperty(value = "MQ type, high throughput: TUBE, high consistency: PULSAR")
+ private String mqType;
@ApiModelProperty(value = "Queue model of Pulsar, parallel: multiple partitions, high throughput, out-of-order "
+ "messages; serial: single partition, low throughput, and orderly messages")
@@ -66,9 +63,9 @@ public class InlongGroupInfo {
@ApiModelProperty(value = "The number of partitions of Pulsar Topic, 1-20")
private Integer topicPartitionNum;
- @ApiModelProperty(value = "MQ resource object, in inlong group",
+ @ApiModelProperty(value = "MQ resource, in inlong group",
notes = "Tube corresponds to Topic, Pulsar corresponds to Namespace")
- private String mqResourceObj;
+ private String mqResource;
@ApiModelProperty(value = "Tube master URL")
private String tubeMaster;
@@ -79,11 +76,20 @@ public class InlongGroupInfo {
@ApiModelProperty(value = "Pulsar service URL")
private String pulsarServiceUrl;
- @ApiModelProperty(value = "Need zookeeper support, 0 false 1 true")
- private Integer zookeeperEnabled;
+ @ApiModelProperty(value = "Whether to enable zookeeper? 0: disable, 1: enable")
+ private Integer enableZookeeper = 0;
+
+ @ApiModelProperty(value = "Whether to enable zookeeper? 0: disable, 1: enable")
+ private Integer enableCreateResource = 1;
+
+ @ApiModelProperty(value = "Whether to use lightweight mode, 0: false, 1: true")
+ private Integer lightweight = 1;
- @ApiModelProperty(value = "Data type name")
- private String schemaName;
+ @ApiModelProperty(value = "Inlong cluster tag, which links to inlong_cluster table")
+ private String inlongClusterTag;
+
+ @ApiModelProperty(value = "Extended params, will be saved as JSON string, such as queue_module, partition_num")
+ private String extParams;
@ApiModelProperty(value = "Number of access items per day, unit: 10,000 items per day")
private Integer dailyRecords;
@@ -115,12 +121,6 @@ public class InlongGroupInfo {
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
private Date modifyTime;
- @ApiModelProperty(value = "Temporary view, string in JSON format")
- private String tempView;
-
- @ApiModelProperty(value = "data proxy cluster id")
- private Integer proxyClusterId;
-
@ApiModelProperty(value = "Inlong group Extension properties")
private List<InlongGroupExtInfo> extList;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupListResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupListResponse.java
index a9dd552bc..00747890f 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupListResponse.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupListResponse.java
@@ -42,9 +42,6 @@ public class InlongGroupListResponse {
@ApiModelProperty(value = "Inlong group name")
private String name;
- @ApiModelProperty(value = "Chinese display name")
- private String cnName;
-
@ApiModelProperty(value = "Name of responsible person, separated by commas")
private String inCharges;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupMqExtBase.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupMqExtBase.java
index 3ad4b1c0f..697b26110 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupMqExtBase.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupMqExtBase.java
@@ -29,7 +29,7 @@ import lombok.Data;
*/
@Data
@ApiModel("Extended inlong group info of different MQs")
-@JsonTypeInfo(use = Id.NAME, visible = true, property = "middlewareType", defaultImpl = InlongGroupMqExtBase.class)
+@JsonTypeInfo(use = Id.NAME, visible = true, property = "mqType", defaultImpl = InlongGroupMqExtBase.class)
@JsonSubTypes({
@JsonSubTypes.Type(value = InlongGroupPulsarInfo.class, name = "PULSAR"),
@JsonSubTypes.Type(value = InlongGroupPulsarInfo.class, name = "TDMQ_PULSAR")
@@ -45,8 +45,8 @@ public class InlongGroupMqExtBase {
@ApiModelProperty(value = "is deleted? 0: deleted, 1: not deleted")
private Integer isDeleted = 0;
- @ApiModelProperty(value = "Middleware type of stream sink, high throughput: TUBE, high consistency : PULSAR")
- private String middlewareType;
+ @ApiModelProperty(value = "MQ type, high throughput: TUBE, high consistency : PULSAR")
+ private String mqType;
@ApiModelProperty(value = "Whether to create mq resource, 0: not create, 1: create; default is 1")
private Integer enableCreateResource = 1;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupPageRequest.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupPageRequest.java
index 3eb9ca6ff..3706208d2 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupPageRequest.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupPageRequest.java
@@ -42,8 +42,8 @@ public class InlongGroupPageRequest extends PageRequest {
@ApiModelProperty(value = "Inlong group id list")
private List<String> groupIdList;
- @ApiModelProperty(value = "MQ resource object")
- private String middlewareType;
+ @ApiModelProperty(value = "MQ type")
+ private String mqType;
@ApiModelProperty(value = "Group status")
private Integer status;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupPulsarInfo.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupPulsarInfo.java
index 766381503..d249ab445 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupPulsarInfo.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupPulsarInfo.java
@@ -34,7 +34,7 @@ import org.apache.inlong.manager.common.enums.MQType;
public class InlongGroupPulsarInfo extends InlongGroupMqExtBase {
public InlongGroupPulsarInfo() {
- this.setMiddlewareType(MQType.PULSAR.getType());
+ this.setMqType(MQType.PULSAR.getType());
}
@ApiModelProperty(value = "Tenant name of Inlong group")
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupRequest.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupRequest.java
index eb2db53e3..dc579add7 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupRequest.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupRequest.java
@@ -43,15 +43,12 @@ public class InlongGroupRequest {
@ApiModelProperty(value = "Inlong group name", required = true)
private String name;
- @ApiModelProperty(value = "Chinese display name")
- private String cnName;
-
@ApiModelProperty(value = "Inlong group description")
private String description;
- @NotNull(message = "middlewareType cannot be null")
- @ApiModelProperty(value = "Middleware type, high throughput: TUBE, high consistency: PULSAR")
- private String middlewareType;
+ @NotNull(message = "mqType cannot be null")
+ @ApiModelProperty(value = "MQ type, high throughput: TUBE, high consistency: PULSAR")
+ private String mqType;
@ApiModelProperty(value = "Queue model of Pulsar, parallel: multiple partitions, high throughput, out-of-order "
+ "messages; serial: single partition, low throughput, and orderly messages")
@@ -62,9 +59,9 @@ public class InlongGroupRequest {
@Builder.Default
private Integer topicPartitionNum = 3;
- @ApiModelProperty(value = "MQ resource object, in inlong group",
+ @ApiModelProperty(value = "MQ resource, in inlong group",
notes = "Tube corresponds to Topic, Pulsar corresponds to Namespace")
- private String mqResourceObj;
+ private String mqResource;
@ApiModelProperty(value = "Tube master URL")
private String tubeMaster;
@@ -75,12 +72,23 @@ public class InlongGroupRequest {
@ApiModelProperty(value = "Pulsar service URL")
private String pulsarServiceUrl;
- @ApiModelProperty(value = "Whether zookeeper enabled? 0: disabled, 1: enabled")
+ @ApiModelProperty(value = "Whether to enable zookeeper? 0: disable, 1: enable")
+ @Builder.Default
+ private Integer enableZookeeper = 0;
+
+ @ApiModelProperty(value = "Whether to enable zookeeper? 0: disable, 1: enable")
+ @Builder.Default
+ private Integer enableCreateResource = 1;
+
+ @ApiModelProperty(value = "Whether to use lightweight mode, 0: false, 1: true")
@Builder.Default
- private Integer zookeeperEnabled = 0;
+ private Integer lightweight = 1;
+
+ @ApiModelProperty(value = "Inlong cluster tag, which links to inlong_cluster table")
+ private String inlongClusterTag;
- @ApiModelProperty(value = "Data type name")
- private String schemaName;
+ @ApiModelProperty(value = "Extended params, will be saved as JSON string, such as queue_module, partition_num")
+ private String extParams;
@ApiModelProperty(value = "Number of access items per day, unit: 10,000 items per day")
private Integer dailyRecords;
@@ -103,12 +111,6 @@ public class InlongGroupRequest {
@ApiModelProperty(value = "Name of creator")
private String creator;
- @ApiModelProperty(value = "Temporary view, string in JSON format")
- private String tempView;
-
- @ApiModelProperty(value = "data proxy cluster id")
- private Integer proxyClusterId;
-
@ApiModelProperty(value = "Inlong group Extension properties")
private List<InlongGroupExtInfo> extList;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupResponse.java
index 1fc5b82b2..9add4ebb3 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupResponse.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupResponse.java
@@ -48,26 +48,16 @@ public class InlongGroupResponse {
@ApiModelProperty(value = "Inlong group name", required = true)
private String name;
- @ApiModelProperty(value = "Chinese display name")
- private String cnName;
-
@ApiModelProperty(value = "Inlong group description")
private String description;
- @NotNull(message = "middlewareType cannot be null")
- @ApiModelProperty(value = "Middleware type, high throughput: TUBE, high consistency: PULSAR")
- private String middlewareType;
+ @NotNull(message = "mqType cannot be null")
+ @ApiModelProperty(value = "MQ type, high throughput: TUBE, high consistency: PULSAR")
+ private String mqType;
- @ApiModelProperty(value = "Queue model of Pulsar, parallel: multiple partitions, high throughput, out-of-order "
- + "messages; serial: single partition, low throughput, and orderly messages")
- private String queueModule;
-
- @ApiModelProperty(value = "The number of partitions of Pulsar Topic, 1-20")
- private Integer topicPartitionNum;
-
- @ApiModelProperty(value = "MQ resource object, in inlong group",
+ @ApiModelProperty(value = "MQ resource, in inlong group",
notes = "Tube corresponds to Topic, Pulsar corresponds to Namespace")
- private String mqResourceObj;
+ private String mqResource;
@ApiModelProperty(value = "Tube master URL")
private String tubeMaster;
@@ -81,8 +71,17 @@ public class InlongGroupResponse {
@ApiModelProperty(value = "Whether zookeeper enabled? 0: disabled, 1: enabled")
private Integer zookeeperEnabled;
- @ApiModelProperty(value = "Data type name")
- private String schemaName;
+ @ApiModelProperty(value = "Whether to enable zookeeper? 0: disable, 1: enable")
+ private Integer enableZookeeper;
+
+ @ApiModelProperty(value = "Whether to enable zookeeper? 0: disable, 1: enable")
+ private Integer enableCreateResource;
+
+ @ApiModelProperty(value = "Whether to use lightweight mode, 0: false, 1: true")
+ private Integer lightweight;
+
+ @ApiModelProperty(value = "Inlong cluster tag, which links to inlong_cluster table")
+ private String inlongClusterTag;
@ApiModelProperty(value = "Number of access items per day, unit: 10,000 items per day")
private Integer dailyRecords;
@@ -96,6 +95,16 @@ public class InlongGroupResponse {
@ApiModelProperty(value = "The maximum length of a single piece of data, unit: Byte")
private Integer maxLength;
+ @ApiModelProperty(value = "Extended params, will be saved as JSON string, such as queue_module, partition_num")
+ private String extParams;
+ /* @ApiModelProperty(value = "Queue model of Pulsar, parallel: multiple partitions, high throughput, out-of-order "
+ + "messages; serial: single partition, low throughput, and orderly messages")
+ private String queueModule;
+
+ @ApiModelProperty(value = "The number of partitions of Pulsar Topic, 1-20")
+ private Integer topicPartitionNum;*/
+
+
@ApiModelProperty(value = "Name of responsible person, separated by commas")
private String inCharges;
@@ -104,6 +113,8 @@ public class InlongGroupResponse {
private Integer status;
+ private Integer previousStatus;
+
private String creator;
private String modifier;
@@ -114,12 +125,6 @@ public class InlongGroupResponse {
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
private Date modifyTime;
- @ApiModelProperty(value = "Temporary view, string in JSON format")
- private String tempView;
-
- @ApiModelProperty(value = "data proxy cluster Id")
- private Integer proxyClusterId;
-
@ApiModelProperty(value = "Inlong group Extension properties")
private List<InlongGroupExtInfo> extList;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupTopicResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupTopicResponse.java
index d602049ed..f42945b98 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupTopicResponse.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/group/InlongGroupTopicResponse.java
@@ -34,11 +34,11 @@ public class InlongGroupTopicResponse {
@ApiModelProperty(value = "Inlong group id", required = true)
private String inlongGroupId;
- @ApiModelProperty(value = "Middleware type, high throughput: TUBE, high consistency: PULSAR")
- private String middlewareType;
+ @ApiModelProperty(value = "MQ type, high throughput: TUBE, high consistency: PULSAR")
+ private String mqType;
- @ApiModelProperty(value = "Tube topic name, or Pulsar namespace name")
- private String mqResourceObj;
+ @ApiModelProperty(value = "MQ resource, Tube topic name, or Pulsar namespace name")
+ private String mqResource;
@ApiModelProperty(value = "Topic list, Tube corresponds to inlong group, there is only 1 topic, "
+ "Pulsar corresponds to inlong stream, there are multiple topics")
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkInfo.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkInfo.java
index 78f2f5501..73f520f9f 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkInfo.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/sink/SinkInfo.java
@@ -39,8 +39,7 @@ public class SinkInfo {
private String creator;
// Inlong stream info
- private String mqResourceObj;
- private String dataSourceType;
+ private String mqResource;
private String dataType;
private String sourceSeparator; // Source separator configured in the stream info
private String dataEscapeChar;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamInfo.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamInfo.java
index ef76978c6..6b26034dc 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamInfo.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamInfo.java
@@ -36,24 +36,21 @@ public class InlongStreamInfo {
@ApiModelProperty(value = "Primary key")
private Integer id;
- @ApiModelProperty(value = "Inlong stream name", required = true)
- private String name;
-
- @ApiModelProperty(value = "Inlong stream description")
- private String description;
-
@ApiModelProperty(value = "Inlong group id")
private String inlongGroupId;
@ApiModelProperty(value = "Inlong stream id")
private String inlongStreamId;
- @ApiModelProperty(value = "MQ resource object, in inlong group",
- notes = "Tube corresponds to Topic, Pulsar corresponds to Namespace")
- private String mqResourceObj;
+ @ApiModelProperty(value = "Inlong stream name", required = true)
+ private String name;
- @ApiModelProperty(value = "Data storage period, unit: day (required when dataSourceType=AUTO_PUSH)")
- private Integer storagePeriod;
+ @ApiModelProperty(value = "Inlong stream description")
+ private String description;
+
+ @ApiModelProperty(value = "MQ resource, in inlong group",
+ notes = "Tube corresponds to Topic, Pulsar corresponds to Namespace")
+ private String mqResource;
@ApiModelProperty(value = "Data type, including: TEXT, KV, etc.")
private String dataType;
@@ -67,9 +64,6 @@ public class InlongStreamInfo {
@ApiModelProperty(value = "Data field escape symbol, stored as ASCII code")
private String dataEscapeChar;
- @ApiModelProperty(value = "(File and DB access) Whether there are predefined fields, 0: no, 1: yes")
- private Integer havePredefinedFields;
-
@ApiModelProperty(value = "Whether to send synchronously, 0: no, 1: yes",
notes = "Each task under this stream sends data synchronously, "
+ "which will affect the throughput of data collection, please choose carefully")
@@ -87,8 +81,11 @@ public class InlongStreamInfo {
@ApiModelProperty(value = "The maximum length of a single piece of data, unit: Byte")
private Integer maxLength;
- @ApiModelProperty(value = "Names of responsible persons, separated by commas")
- private String inCharges;
+ @ApiModelProperty(value = "Data storage period, unit: day")
+ private Integer storagePeriod;
+
+ @ApiModelProperty(value = "Extended params, will be saved as JSON string")
+ private String extParams;
@ApiModelProperty(value = "Status")
private Integer status;
@@ -109,9 +106,6 @@ public class InlongStreamInfo {
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
private Date modifyTime;
- @ApiModelProperty(value = "Temporary view, string in JSON format")
- private String tempView;
-
@ApiModelProperty(value = "Field list")
private List<InlongStreamFieldInfo> fieldList;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamListResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamListResponse.java
index f5942f852..2ea380d4d 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamListResponse.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamListResponse.java
@@ -35,18 +35,15 @@ public class InlongStreamListResponse {
@ApiModelProperty(value = "Primary key")
private Integer id;
- @ApiModelProperty(value = "Inlong stream id")
- private String inlongStreamId;
-
@ApiModelProperty(value = "Inlong group id")
private String inlongGroupId;
+ @ApiModelProperty(value = "Inlong stream id")
+ private String inlongStreamId;
+
@ApiModelProperty(value = "Inlong stream name")
private String name;
- @ApiModelProperty(value = "Source type, FILE/DB/AUTO_PUSH")
- private String dataSourceType;
-
@ApiModelProperty(value = "Data type, TEXT, KEY-VALUE, PB, BON")
private String dataType;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamPageRequest.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamPageRequest.java
index 88e62eddd..8c30286e6 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamPageRequest.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamPageRequest.java
@@ -37,9 +37,6 @@ public class InlongStreamPageRequest extends PageRequest {
@ApiModelProperty(value = "Keyword, can be stream id or name")
private String keyword;
- @ApiModelProperty(value = "Source type, including: FILE, DB, AUTO_PUSH (DATA_PROXY_SDK, HTTP)")
- private String dataSourceType;
-
@ApiModelProperty(value = "Sink type to be created (which has no inlong stream of this type)")
private String sinkType;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamRequest.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamRequest.java
index 12bc2e662..4fe9f923d 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamRequest.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamRequest.java
@@ -45,12 +45,9 @@ public class InlongStreamRequest {
@ApiModelProperty(value = "Inlong stream description")
private String description;
- @ApiModelProperty(value = "MQ resource object, in inlong group",
+ @ApiModelProperty(value = "MQ resource, in inlong group",
notes = "Tube corresponds to Topic, Pulsar corresponds to Namespace")
- private String mqResourceObj;
-
- @ApiModelProperty(value = "Data storage period, unit: day (required when dataSourceType=AUTO_PUSH)")
- private Integer storagePeriod;
+ private String mqResource;
@ApiModelProperty(value = "Data type, including: TEXT, KV, etc.")
private String dataType;
@@ -64,9 +61,6 @@ public class InlongStreamRequest {
@ApiModelProperty(value = "Data field escape symbol, stored as ASCII code")
private String dataEscapeChar;
- @ApiModelProperty(value = "(File and DB access) Whether there are predefined fields, 0: no, 1: yes")
- private Integer havePredefinedFields;
-
@ApiModelProperty(value = "Whether to send synchronously, 0: no, 1: yes",
notes = "Each task under this stream sends data synchronously, "
+ "which will affect the throughput of data collection, please choose carefully")
@@ -84,11 +78,11 @@ public class InlongStreamRequest {
@ApiModelProperty(value = "The maximum length of a single piece of data, unit: Byte")
private Integer maxLength;
- @ApiModelProperty(value = "Names of responsible persons, separated by commas")
- private String inCharges;
+ @ApiModelProperty(value = "Data storage period, unit: day")
+ private Integer storagePeriod;
- @ApiModelProperty(value = "StreamPipeline snapshot of stream,, string in JSON format")
- private String tempView;
+ @ApiModelProperty(value = "Extended params, will be saved as JSON string")
+ private String extParams;
@ApiModelProperty(value = "Field list")
private List<InlongStreamFieldInfo> fieldList;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamResponse.java
index c2dcc511f..9dc694fa7 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamResponse.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamResponse.java
@@ -41,24 +41,21 @@ public class InlongStreamResponse {
@ApiModelProperty(value = "Primary key")
private Integer id;
- @ApiModelProperty(value = "Inlong stream name", required = true)
- private String name;
-
@ApiModelProperty(value = "Inlong group id")
private String inlongGroupId;
@ApiModelProperty(value = "Inlong stream id")
private String inlongStreamId;
+ @ApiModelProperty(value = "Inlong stream name", required = true)
+ private String name;
+
@ApiModelProperty(value = "Inlong stream description")
private String description;
- @ApiModelProperty(value = "MQ resource object, in inlong group",
+ @ApiModelProperty(value = "MQ resource, in inlong group",
notes = "Tube corresponds to Topic, Pulsar corresponds to Namespace")
- private String mqResourceObj;
-
- @ApiModelProperty(value = "Data storage period, unit: day (required when dataSourceType=AUTO_PUSH)")
- private Integer storagePeriod;
+ private String mqResource;
@ApiModelProperty(value = "Data type, including: TEXT, KV, etc.")
private String dataType;
@@ -72,9 +69,6 @@ public class InlongStreamResponse {
@ApiModelProperty(value = "Data field escape symbol, stored as ASCII code")
private String dataEscapeChar;
- @ApiModelProperty(value = "(File and DB access) Whether there are predefined fields, 0: no, 1: yes")
- private Integer havePredefinedFields;
-
@ApiModelProperty(value = "Whether to send synchronously, 0: no, 1: yes",
notes = "Each task under this stream sends data synchronously, "
+ "which will affect the throughput of data collection, please choose carefully")
@@ -92,8 +86,11 @@ public class InlongStreamResponse {
@ApiModelProperty(value = "The maximum length of a single piece of data, unit: Byte")
private Integer maxLength;
- @ApiModelProperty(value = "Names of responsible persons, separated by commas")
- private String inCharges;
+ @ApiModelProperty(value = "Data storage period, unit: day")
+ private Integer storagePeriod;
+
+ @ApiModelProperty(value = "Extended params, will be saved as JSON string")
+ private String extParams;
@ApiModelProperty(value = "Status")
private Integer status;
@@ -115,9 +112,6 @@ public class InlongStreamResponse {
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
private Date modifyTime;
- @ApiModelProperty(value = "Temporary view, string in JSON format")
- private String tempView;
-
@ApiModelProperty(value = "Field list")
private List<InlongStreamFieldInfo> fieldList;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamTopicResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamTopicResponse.java
index 1c9d1daa0..1c5ca4156 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamTopicResponse.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/InlongStreamTopicResponse.java
@@ -31,7 +31,7 @@ public class InlongStreamTopicResponse {
@ApiModelProperty(value = "InLong stream ID")
private String inlongStreamId;
- @ApiModelProperty(value = "Message queue type, the inlong stream id corresponds to the topic of Pulsar one-to-one")
- private String mqResourceObj;
+ @ApiModelProperty(value = "MQ resource, the inlong stream id corresponds to the topic of Pulsar one-to-one")
+ private String mqResource;
}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/StreamBriefResponse.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/StreamBriefResponse.java
index 0d7c3b6f4..3dd260f33 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/StreamBriefResponse.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/StreamBriefResponse.java
@@ -45,9 +45,6 @@ public class StreamBriefResponse {
@ApiModelProperty(value = "Inlong stream name")
private String name;
- @ApiModelProperty(value = "Source type, support: FILE/DB/AUTO_PUSH")
- private String dataSourceType;
-
@ApiModelProperty(value = "Sink brief list")
private List<SinkBriefResponse> sinkList;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/StreamField.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/StreamField.java
index 69b331fba..6f1e31895 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/StreamField.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/stream/StreamField.java
@@ -34,10 +34,40 @@ import org.apache.inlong.manager.common.enums.MetaFieldType;
@ApiModel("Stream field configuration")
public class StreamField {
- public static final StreamField PROCESSING_TIME = new StreamField(100, FieldType.BIGINT,
+ public static final StreamField PROCESSING_TIME = new StreamField(
+ 100,
+ FieldType.BIGINT,
MetaFieldType.PROCESSING_TIME.getName(),
null, null, 1);
+ @ApiModelProperty("Field index")
+ private Integer id;
+
+ @ApiModelProperty(value = "Field type", required = true)
+ private FieldType fieldType;
+
+ @ApiModelProperty(value = "Field name", required = true)
+ private String fieldName;
+
+ @ApiModelProperty(value = "Field comment")
+ private String fieldComment;
+
+ @ApiModelProperty(value = "Field value for constants")
+ private String fieldValue;
+
+ @ApiModelProperty("Is this field a meta field, 0: no, 1: yes")
+ private Integer isMetaField = 0;
+
+ @ApiModelProperty("Field format, including: MICROSECONDS, MILLISECONDS, SECONDS, SQL, ISO_8601"
+ + " and custom such as 'yyyy-MM-dd HH:mm:ss'. This is mainly used for time format")
+ private String fieldFormat;
+
+ @ApiModelProperty("Origin node name which stream fields belong")
+ private String originNodeName;
+
+ @ApiModelProperty("Origin field name before transform operation")
+ private String originFieldName;
+
public StreamField(int index, FieldType fieldType, String fieldName, String fieldComment, String fieldValue) {
this.id = index;
this.fieldType = fieldType;
@@ -67,32 +97,4 @@ public class StreamField {
this.originFieldName = originFieldName;
}
- @ApiModelProperty("Field index")
- private Integer id;
-
- @ApiModelProperty(value = "Field type", required = true)
- private FieldType fieldType;
-
- @ApiModelProperty(value = "Field name", required = true)
- private String fieldName;
-
- @ApiModelProperty(value = "Field comment")
- private String fieldComment;
-
- @ApiModelProperty(value = "Field value for constants")
- private String fieldValue;
-
- @ApiModelProperty("Is this field a meta field, 0: no, 1: yes")
- private Integer isMetaField = 0;
-
- @ApiModelProperty("Field format, including: MICROSECONDS, MILLISECONDS, SECONDS, SQL, ISO_8601"
- + " and custom such as 'yyyy-MM-dd HH:mm:ss'. This is mainly used for time format")
- private String fieldFormat;
-
- @ApiModelProperty("Origin Node name which stream field belongs")
- private String originNodeName;
-
- @ApiModelProperty("Origin field name before transform operation")
- private String originFieldName;
-
}
diff --git a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/InlongGroupEntity.java b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/InlongGroupEntity.java
index bf4e1c5c1..88687ad1f 100644
--- a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/InlongGroupEntity.java
+++ b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/InlongGroupEntity.java
@@ -32,27 +32,28 @@ public class InlongGroupEntity implements Serializable {
private Integer id;
private String inlongGroupId;
private String name;
- private String cnName;
private String description;
- private String middlewareType;
- private String queueModule;
- private Integer topicPartitionNum;
- private String mqResourceObj;
+ private String mqType;
+ private String mqResource;
private Integer dailyRecords;
private Integer dailyStorage;
private Integer peakRecords;
- private Integer zookeeperEnabled;
private Integer maxLength;
- private String schemaName;
+
+ private String enableCreateResource;
+ private Integer enableZookeeper;
+ private Integer lightweight;
+ private Integer inlongClusterTag;
+
+ private String extParams;
private String inCharges;
private String followers;
private Integer status;
+ private Integer previousStatus;
private Integer isDeleted;
private String creator;
private String modifier;
private Date createTime;
private Date modifyTime;
- private String tempView;
- private Integer proxyClusterId;
}
\ No newline at end of file
diff --git a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/InlongStreamEntity.java b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/InlongStreamEntity.java
index 0150c9387..a01ee3152 100644
--- a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/InlongStreamEntity.java
+++ b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/InlongStreamEntity.java
@@ -30,25 +30,24 @@ public class InlongStreamEntity implements Serializable {
private static final long serialVersionUID = 1L;
private Integer id;
- private String inlongStreamId;
private String inlongGroupId;
+ private String inlongStreamId;
private String name;
private String description;
- private String mqResourceObj;
- private String dataSourceType;
- private Integer storagePeriod;
+ private String mqResource;
+
private String dataType;
private String dataEncoding;
private String dataSeparator;
private String dataEscapeChar;
- private Integer havePredefinedFields;
private Integer syncSend;
private Integer dailyRecords;
private Integer dailyStorage;
private Integer peakRecords;
private Integer maxLength;
- private String inCharges;
+ private Integer storagePeriod;
+ private String extParams;
private Integer status;
private Integer previousStatus;
@@ -57,6 +56,5 @@ public class InlongStreamEntity implements Serializable {
private String modifier;
private Date createTime;
private Date modifyTime;
- private String tempView;
}
\ No newline at end of file
diff --git a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/InlongGroupEntityMapper.java b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/InlongGroupEntityMapper.java
index c2b7e1092..904f452d8 100644
--- a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/InlongGroupEntityMapper.java
+++ b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/InlongGroupEntityMapper.java
@@ -18,7 +18,6 @@
package org.apache.inlong.manager.dao.mapper;
import org.apache.ibatis.annotations.Param;
-import org.apache.inlong.common.pojo.dataproxy.DataProxyConfig;
import org.apache.inlong.manager.common.pojo.group.InlongGroupPageRequest;
import org.apache.inlong.manager.dao.entity.InlongGroupEntity;
import org.springframework.stereotype.Repository;
@@ -29,12 +28,8 @@ import java.util.Map;
@Repository
public interface InlongGroupEntityMapper {
- int deleteByPrimaryKey(Integer id);
-
int insert(InlongGroupEntity record);
- int insertSelective(InlongGroupEntity record);
-
InlongGroupEntity selectByPrimaryKey(Integer id);
List<Map<String, Object>> countGroupByUser(@Param(value = "username") String username);
@@ -49,20 +44,15 @@ public interface InlongGroupEntityMapper {
List<InlongGroupEntity> selectAll(Integer status);
- /**
- * get all config with inlong group status of 130, that is, config successful
- */
- List<DataProxyConfig> selectDataProxyConfig();
-
List<String> selectGroupIdByProxyId(Integer proxyClusterId);
- int updateByPrimaryKeySelective(InlongGroupEntity record);
+ int updateByPrimaryKey(InlongGroupEntity record);
int updateByIdentifierSelective(InlongGroupEntity record);
- int updateByPrimaryKey(InlongGroupEntity record);
-
int updateStatus(@Param("groupId") String groupId, @Param("status") Integer status,
@Param("modifier") String modifier);
+ int deleteByPrimaryKey(Integer id);
+
}
\ No newline at end of file
diff --git a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/InlongStreamEntityMapper.java b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/InlongStreamEntityMapper.java
index b76d7e28c..5eb0b0889 100644
--- a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/InlongStreamEntityMapper.java
+++ b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/InlongStreamEntityMapper.java
@@ -28,16 +28,10 @@ import java.util.List;
@Repository
public interface InlongStreamEntityMapper {
- int deleteByPrimaryKey(Integer id);
-
int insert(InlongStreamEntity record);
int insertSelective(InlongStreamEntity record);
- InlongStreamEntity selectByPrimaryKey(Integer id);
-
- int updateByPrimaryKey(InlongStreamEntity record);
-
InlongStreamEntity selectByIdentifier(@Param("groupId") String groupId, @Param("streamId") String streamId);
Integer selectExistByIdentifier(@Param("groupId") String groupId, @Param("streamId") String streamId);
@@ -56,12 +50,7 @@ public interface InlongStreamEntityMapper {
List<InlongStreamTopicResponse> selectTopicList(@Param("groupId") String groupId);
- /**
- * Physically delete all inlong streams of the specified inlong group id
- *
- * @return rows deleted
- */
- int deleteAllByGroupId(@Param("groupId") String groupId);
+ int updateByPrimaryKey(InlongStreamEntity record);
int updateByIdentifierSelective(InlongStreamEntity streamEntity);
@@ -73,4 +62,13 @@ public interface InlongStreamEntityMapper {
*/
void logicDeleteDlqOrRlq(String groupId, String streamId, String operator);
+ int deleteByPrimaryKey(Integer id);
+
+ /**
+ * Physically delete all inlong streams of the specified inlong group id
+ *
+ * @return rows deleted
+ */
+ int deleteAllByGroupId(@Param("groupId") String groupId);
+
}
\ No newline at end of file
diff --git a/inlong-manager/manager-dao/src/main/resources/mappers/InlongGroupEntityMapper.xml b/inlong-manager/manager-dao/src/main/resources/mappers/InlongGroupEntityMapper.xml
index 32018f650..666d102ba 100644
--- a/inlong-manager/manager-dao/src/main/resources/mappers/InlongGroupEntityMapper.xml
+++ b/inlong-manager/manager-dao/src/main/resources/mappers/InlongGroupEntityMapper.xml
@@ -24,28 +24,28 @@
<id column="id" jdbcType="INTEGER" property="id"/>
<result column="inlong_group_id" jdbcType="VARCHAR" property="inlongGroupId"/>
<result column="name" jdbcType="VARCHAR" property="name"/>
- <result column="cn_name" jdbcType="VARCHAR" property="cnName"/>
<result column="description" jdbcType="VARCHAR" property="description"/>
- <result column="middleware_type" jdbcType="VARCHAR" property="middlewareType"/>
- <result column="queue_module" jdbcType="VARCHAR" property="queueModule"/>
- <result column="topic_partition_num" jdbcType="INTEGER" property="topicPartitionNum"/>
- <result column="mq_resource_obj" jdbcType="VARCHAR" property="mqResourceObj"/>
+ <result column="mq_type" jdbcType="VARCHAR" property="mqType"/>
+ <result column="mq_resource" jdbcType="VARCHAR" property="mqResource"/>
<result column="daily_records" jdbcType="INTEGER" property="dailyRecords"/>
<result column="daily_storage" jdbcType="INTEGER" property="dailyStorage"/>
<result column="peak_records" jdbcType="INTEGER" property="peakRecords"/>
<result column="max_length" jdbcType="INTEGER" property="maxLength"/>
- <result column="schema_name" jdbcType="VARCHAR" property="schemaName"/>
+ <result column="enable_zookeeper" jdbcType="INTEGER" property="enableZookeeper"/>
+ <result column="enable_create_resource" jdbcType="INTEGER" property="enableCreateResource"/>
+ <result column="lightweight" jdbcType="INTEGER" property="lightweight"/>
+ <result column="inlong_cluster_tag" jdbcType="VARCHAR" property="inlongClusterTag"/>
+ <result column="ext_params" jdbcType="LONGVARCHAR" property="extParams"/>
+
<result column="in_charges" jdbcType="VARCHAR" property="inCharges"/>
<result column="followers" jdbcType="VARCHAR" property="followers"/>
<result column="status" jdbcType="INTEGER" property="status"/>
+ <result column="previous_status" jdbcType="INTEGER" property="previousStatus"/>
<result column="is_deleted" jdbcType="INTEGER" property="isDeleted"/>
<result column="creator" jdbcType="VARCHAR" property="creator"/>
<result column="modifier" jdbcType="VARCHAR" property="modifier"/>
<result column="create_time" jdbcType="TIMESTAMP" property="createTime"/>
<result column="modify_time" jdbcType="TIMESTAMP" property="modifyTime"/>
- <result column="temp_view" jdbcType="LONGVARCHAR" property="tempView"/>
- <result column="zookeeper_enabled" jdbcType="INTEGER" property="zookeeperEnabled"/>
- <result column="proxy_cluster_id" jdbcType="INTEGER" property="proxyClusterId"/>
</resultMap>
<resultMap id="statusCountMap" type="java.util.Map">
@@ -54,11 +54,40 @@
</resultMap>
<sql id="Base_Column_List">
- id
- , inlong_group_id, name, cn_name, description, middleware_type, queue_module, topic_partition_num,
- mq_resource_obj, daily_records, daily_storage, peak_records, max_length, schema_name, in_charges, followers,
- status, is_deleted, creator, modifier, create_time, modify_time, temp_view, zookeeper_enabled, proxy_cluster_id
+ id, inlong_group_id, name, description, mq_type, mq_resource,
+ daily_records, daily_storage, peak_records, max_length,
+ enable_zookeeper, enable_create_resource, lightweight, inlong_cluster_tag, ext_params,
+ in_charges, followers, status, previous_status, is_deleted, creator, modifier, create_time, modify_time
</sql>
+
+ <insert id="insert" useGeneratedKeys="true" keyProperty="id"
+ parameterType="org.apache.inlong.manager.dao.entity.InlongGroupEntity">
+ insert into inlong_group (id, inlong_group_id,
+ name, description,
+ mq_type, mq_resource,
+ daily_records, daily_storage,
+ peak_records, max_length,
+ enable_zookeeper, enable_create_resource,
+ lightweight, inlong_cluster_tag,
+ ext_params, in_charges,
+ followers, status,
+ previous_status, is_deleted,
+ creator, modifier,
+ create_time, modify_time)
+ values (#{id,jdbcType=INTEGER}, #{inlongGroupId,jdbcType=VARCHAR},
+ #{name,jdbcType=VARCHAR}, #{description,jdbcType=VARCHAR},
+ #{mqType,jdbcType=VARCHAR}, #{mqResource,jdbcType=VARCHAR},
+ #{dailyRecords,jdbcType=INTEGER}, #{dailyStorage,jdbcType=INTEGER},
+ #{peakRecords,jdbcType=INTEGER}, #{maxLength,jdbcType=INTEGER},
+ #{enableZookeeper,jdbcType=INTEGER}, #{enableCreateResource,jdbcType=INTEGER},
+ #{lightweight,jdbcType=INTEGER}, #{inlongClusterTag,jdbcType=VARCHAR},
+ #{extParams,jdbcType=LONGVARCHAR}, #{inCharges,jdbcType=VARCHAR},
+ #{followers,jdbcType=VARCHAR}, #{status,jdbcType=INTEGER},
+ #{previousStatus,jdbcType=INTEGER}, #{isDeleted,jdbcType=INTEGER},
+ #{creator,jdbcType=VARCHAR}, #{modifier,jdbcType=VARCHAR},
+ #{createTime,jdbcType=TIMESTAMP}, #{modifyTime,jdbcType=TIMESTAMP})
+ </insert>
+
<select id="selectByPrimaryKey" parameterType="java.lang.Integer" resultMap="BaseResultMap">
select
<include refid="Base_Column_List"/>
@@ -127,8 +156,8 @@
#{name}
</foreach>
</if>
- <if test="middlewareType != null and middlewareType != ''">
- and middleware_type = #{middlewareType, jdbcType=VARCHAR}
+ <if test="mqType != null and mqType != ''">
+ and mq_type = #{mqType, jdbcType=VARCHAR}
</if>
<if test="status != null and status != ''">
and status = #{status, jdbcType=INTEGER}
@@ -142,230 +171,55 @@
</where>
order by modify_time desc
</select>
- <select id="selectDataProxyConfig" resultType="org.apache.inlong.common.pojo.dataproxy.DataProxyConfig">
- select biz.inlong_group_id as inlongGroupId, biz.mq_resource_obj as topic, d.sort_type as m
- from inlong_group biz,
- data_schema d
- where biz.status = 130
- and biz.is_deleted = 0
- and biz.schema_name = d.name
- </select>
<select id="selectGroupIdByProxyId" parameterType="java.lang.Integer" resultType="java.lang.String">
select inlong_group_id
from inlong_group
- where proxy_cluster_id = #{proxyClusterId,jdbcType=INTEGER}
+ where inlong_cluster_tag = #{inlongClusterTag,jdbcType=VARCHAR}
</select>
- <delete id="deleteByPrimaryKey" parameterType="java.lang.Integer">
- delete
- from inlong_group
+ <update id="updateByPrimaryKey" parameterType="org.apache.inlong.manager.dao.entity.InlongGroupEntity">
+ update inlong_group
+ set inlong_group_id = #{inlongGroupId,jdbcType=VARCHAR},
+ name = #{name,jdbcType=VARCHAR},
+ description = #{description,jdbcType=VARCHAR},
+ mq_type = #{mqType,jdbcType=VARCHAR},
+ mq_resource = #{mqResource,jdbcType=VARCHAR},
+ daily_records = #{dailyRecords,jdbcType=INTEGER},
+ daily_storage = #{dailyStorage,jdbcType=INTEGER},
+ peak_records = #{peakRecords,jdbcType=INTEGER},
+ max_length = #{maxLength,jdbcType=INTEGER},
+
+ enable_zookeeper = #{enableZookeeper,jdbcType=INTEGER},
+ enable_create_resource = #{enableCreateResource,jdbcType=INTEGER},
+ lightweight = #{lightweight,jdbcType=INTEGER},
+ inlong_cluster_tag = #{inlongClusterTag,jdbcType=VARCHAR},
+
+ ext_params = #{extParams,jdbcType=LONGVARCHAR},
+ in_charges = #{inCharges,jdbcType=VARCHAR},
+ followers = #{followers,jdbcType=VARCHAR},
+ status = #{status,jdbcType=INTEGER},
+ previous_status = #{previous_status,jdbcType=INTEGER},
+ is_deleted = #{isDeleted,jdbcType=INTEGER},
+ creator = #{creator,jdbcType=VARCHAR},
+ modifier = #{modifier,jdbcType=VARCHAR},
+ create_time = #{createTime,jdbcType=TIMESTAMP},
+ modify_time = #{modifyTime,jdbcType=TIMESTAMP}
where id = #{id,jdbcType=INTEGER}
- </delete>
- <insert id="insert" useGeneratedKeys="true" keyProperty="id"
- parameterType="org.apache.inlong.manager.dao.entity.InlongGroupEntity">
- insert into inlong_group (id, inlong_group_id, name,
- cn_name, description, middleware_type,
- queue_module, topic_partition_num,
- mq_resource_obj, daily_records, daily_storage,
- peak_records, max_length, schema_name,
- in_charges, followers, status,
- is_deleted, creator, modifier,
- create_time, modify_time, temp_view,
- zookeeper_enabled, proxy_cluster_id)
- values (#{id,jdbcType=INTEGER}, #{inlongGroupId,jdbcType=VARCHAR}, #{name,jdbcType=VARCHAR},
- #{cnName,jdbcType=VARCHAR}, #{description,jdbcType=VARCHAR}, #{middlewareType,jdbcType=VARCHAR},
- #{queueModule,jdbcType=VARCHAR}, #{topicPartitionNum,jdbcType=INTEGER},
- #{mqResourceObj,jdbcType=VARCHAR}, #{dailyRecords,jdbcType=INTEGER}, #{dailyStorage,jdbcType=INTEGER},
- #{peakRecords,jdbcType=INTEGER}, #{maxLength,jdbcType=INTEGER}, #{schemaName,jdbcType=VARCHAR},
- #{inCharges,jdbcType=VARCHAR}, #{followers,jdbcType=VARCHAR}, #{status,jdbcType=INTEGER},
- #{isDeleted,jdbcType=INTEGER}, #{creator,jdbcType=VARCHAR}, #{modifier,jdbcType=VARCHAR},
- #{createTime,jdbcType=TIMESTAMP}, #{modifyTime,jdbcType=TIMESTAMP}, #{tempView,jdbcType=LONGVARCHAR},
- #{zookeeperEnabled,jdbcType=INTEGER}, #{proxyClusterId,jdbcType=INTEGER})
- </insert>
- <insert id="insertSelective" useGeneratedKeys="true" keyProperty="id"
- parameterType="org.apache.inlong.manager.dao.entity.InlongGroupEntity">
- insert into inlong_group
- <trim prefix="(" suffix=")" suffixOverrides=",">
- <if test="id != null">
- id,
- </if>
- <if test="inlongGroupId != null">
- inlong_group_id,
- </if>
- <if test="name != null">
- name,
- </if>
- <if test="cnName != null">
- cn_name,
- </if>
- <if test="description != null">
- description,
- </if>
- <if test="middlewareType != null">
- middleware_type,
- </if>
- <if test="queueModule != null">
- queue_module,
- </if>
- <if test="topicPartitionNum != null">
- topic_partition_num,
- </if>
- <if test="mqResourceObj != null">
- mq_resource_obj,
- </if>
- <if test="dailyRecords != null">
- daily_records,
- </if>
- <if test="dailyStorage != null">
- daily_storage,
- </if>
- <if test="peakRecords != null">
- peak_records,
- </if>
- <if test="maxLength != null">
- max_length,
- </if>
- <if test="schemaName != null">
- schema_name,
- </if>
- <if test="inCharges != null">
- in_charges,
- </if>
- <if test="followers != null">
- followers,
- </if>
- <if test="status != null">
- status,
- </if>
- <if test="isDeleted != null">
- is_deleted,
- </if>
- <if test="creator != null">
- creator,
- </if>
- <if test="modifier != null">
- modifier,
- </if>
- <if test="createTime != null">
- create_time,
- </if>
- <if test="modifyTime != null">
- modify_time,
- </if>
- <if test="tempView != null">
- temp_view,
- </if>
- <if test="zookeeperEnabled != null">
- zookeeper_enabled,
- </if>
- <if test="proxyClusterId != null">
- proxy_cluster_id,
- </if>
- </trim>
- <trim prefix="values (" suffix=")" suffixOverrides=",">
- <if test="id != null">
- #{id,jdbcType=INTEGER},
- </if>
- <if test="inlongGroupId != null">
- #{inlongGroupId,jdbcType=VARCHAR},
- </if>
- <if test="name != null">
- #{name,jdbcType=VARCHAR},
- </if>
- <if test="cnName != null">
- #{cnName,jdbcType=VARCHAR},
- </if>
- <if test="description != null">
- #{description,jdbcType=VARCHAR},
- </if>
- <if test="middlewareType != null">
- #{middlewareType,jdbcType=VARCHAR},
- </if>
- <if test="queueModule != null">
- #{queueModule,jdbcType=VARCHAR},
- </if>
- <if test="topicPartitionNum != null">
- #{topicPartitionNum,jdbcType=INTEGER},
- </if>
- <if test="mqResourceObj != null">
- #{mqResourceObj,jdbcType=VARCHAR},
- </if>
- <if test="dailyRecords != null">
- #{dailyRecords,jdbcType=INTEGER},
- </if>
- <if test="dailyStorage != null">
- #{dailyStorage,jdbcType=INTEGER},
- </if>
- <if test="peakRecords != null">
- #{peakRecords,jdbcType=INTEGER},
- </if>
- <if test="maxLength != null">
- #{maxLength,jdbcType=INTEGER},
- </if>
- <if test="schemaName != null">
- #{schemaName,jdbcType=VARCHAR},
- </if>
- <if test="inCharges != null">
- #{inCharges,jdbcType=VARCHAR},
- </if>
- <if test="followers != null">
- #{followers,jdbcType=VARCHAR},
- </if>
- <if test="status != null">
- #{status,jdbcType=INTEGER},
- </if>
- <if test="isDeleted != null">
- #{isDeleted,jdbcType=INTEGER},
- </if>
- <if test="creator != null">
- #{creator,jdbcType=VARCHAR},
- </if>
- <if test="modifier != null">
- #{modifier,jdbcType=VARCHAR},
- </if>
- <if test="createTime != null">
- #{createTime,jdbcType=TIMESTAMP},
- </if>
- <if test="modifyTime != null">
- #{modifyTime,jdbcType=TIMESTAMP},
- </if>
- <if test="tempView != null">
- #{tempView,jdbcType=LONGVARCHAR},
- </if>
- <if test="zookeeperEnabled != null">
- #{zookeeperEnabled,jdbcType=INTEGER},
- </if>
- <if test="proxyClusterId != null">
- #{proxyClusterId,jdbcType=INTEGER},
- </if>
- </trim>
- </insert>
- <update id="updateByPrimaryKeySelective" parameterType="org.apache.inlong.manager.dao.entity.InlongGroupEntity">
+ </update>
+ <update id="updateByIdentifierSelective" parameterType="org.apache.inlong.manager.dao.entity.InlongGroupEntity">
update inlong_group
<set>
- <if test="inlongGroupId != null">
- inlong_group_id = #{inlongGroupId,jdbcType=VARCHAR},
- </if>
<if test="name != null">
name = #{name,jdbcType=VARCHAR},
</if>
- <if test="cnName != null">
- cn_name = #{cnName,jdbcType=VARCHAR},
- </if>
<if test="description != null">
description = #{description,jdbcType=VARCHAR},
</if>
- <if test="middlewareType != null">
- middleware_type = #{middlewareType,jdbcType=VARCHAR},
+ <if test="mqType != null">
+ mq_type = #{mqType,jdbcType=VARCHAR},
</if>
- <if test="queueModule != null">
- queue_module = #{queueModule,jdbcType=VARCHAR},
- </if>
- <if test="topicPartitionNum != null">
- topic_partition_num = #{topicPartitionNum,jdbcType=INTEGER},
- </if>
- <if test="mqResourceObj != null">
- mq_resource_obj = #{mqResourceObj,jdbcType=VARCHAR},
+ <if test="mqResource != null">
+ mq_resource = #{mqResource,jdbcType=VARCHAR},
</if>
<if test="dailyRecords != null">
daily_records = #{dailyRecords,jdbcType=INTEGER},
@@ -379,83 +233,20 @@
<if test="maxLength != null">
max_length = #{maxLength,jdbcType=INTEGER},
</if>
- <if test="schemaName != null">
- schema_name = #{schemaName,jdbcType=VARCHAR},
+ <if test="enableZookeeper != null">
+ enable_zookeeper = #{enableZookeeper,jdbcType=INTEGER},
</if>
- <if test="inCharges != null">
- in_charges = #{inCharges,jdbcType=VARCHAR},
+ <if test="enableCreateResource != null">
+ enable_create_resource = #{enableCreateResource,jdbcType=INTEGER},
</if>
- <if test="followers != null">
- followers = #{followers,jdbcType=VARCHAR},
- </if>
- <if test="status != null">
- status = #{status,jdbcType=INTEGER},
- </if>
- <if test="isDeleted != null">
- is_deleted = #{isDeleted,jdbcType=INTEGER},
- </if>
- <if test="creator != null">
- creator = #{creator,jdbcType=VARCHAR},
+ <if test="lightweight != null">
+ lightweight = #{lightweight,jdbcType=INTEGER},
</if>
- <if test="modifier != null">
- modifier = #{modifier,jdbcType=VARCHAR},
- </if>
- <if test="createTime != null">
- create_time = #{createTime,jdbcType=TIMESTAMP},
- </if>
- <if test="modifyTime != null">
- modify_time = #{modifyTime,jdbcType=TIMESTAMP},
- </if>
- <if test="tempView != null">
- temp_view = #{tempView,jdbcType=LONGVARCHAR},
+ <if test="inlongClusterTag != null">
+ inlong_cluster_tag = #{inlongClusterTag,jdbcType=VARCHAR},
</if>
- <if test="zookeeperEnabled != null">
- zookeeper_enabled = #{zookeeperEnabled,jdbcType=INTEGER},
- </if>
- <if test="proxyClusterId != null">
- proxy_cluster_id = #{proxyClusterId,jdbcType=INTEGER},
- </if>
- </set>
- where id = #{id,jdbcType=INTEGER}
- </update>
- <update id="updateByIdentifierSelective" parameterType="org.apache.inlong.manager.dao.entity.InlongGroupEntity">
- update inlong_group
- <set>
- <if test="name != null">
- name = #{name,jdbcType=VARCHAR},
- </if>
- <if test="cnName != null">
- cn_name = #{cnName,jdbcType=VARCHAR},
- </if>
- <if test="description != null">
- description = #{description,jdbcType=VARCHAR},
- </if>
- <if test="middlewareType != null">
- middleware_type = #{middlewareType,jdbcType=VARCHAR},
- </if>
- <if test="queueModule != null">
- queue_module = #{queueModule,jdbcType=VARCHAR},
- </if>
- <if test="topicPartitionNum != null">
- topic_partition_num = #{topicPartitionNum,jdbcType=INTEGER},
- </if>
- <if test="mqResourceObj != null">
- mq_resource_obj = #{mqResourceObj,jdbcType=VARCHAR},
- </if>
- <if test="dailyRecords != null">
- daily_records = #{dailyRecords,jdbcType=INTEGER},
- </if>
- <if test="dailyStorage != null">
- daily_storage = #{dailyStorage,jdbcType=INTEGER},
- </if>
- <if test="peakRecords != null">
- peak_records = #{peakRecords,jdbcType=INTEGER},
- </if>
- <if test="maxLength != null">
- max_length = #{maxLength,jdbcType=INTEGER},
- </if>
- <if test="schemaName != null">
- schema_name = #{schemaName,jdbcType=VARCHAR},
+ <if test="extParams != null">
+ ext_params = #{extParams,jdbcType=LONGVARCHAR},
</if>
<if test="inCharges != null">
in_charges = #{inCharges,jdbcType=VARCHAR},
@@ -466,6 +257,9 @@
<if test="status != null">
status = #{status,jdbcType=INTEGER},
</if>
+ <if test="previousStatus != null">
+ previous_status = #{previousStatus,jdbcType=INTEGER},
+ </if>
<if test="isDeleted != null">
is_deleted = #{isDeleted,jdbcType=INTEGER},
</if>
@@ -475,54 +269,22 @@
<if test="modifier != null">
modifier = #{modifier,jdbcType=VARCHAR},
</if>
- <if test="tempView != null">
- temp_view = #{tempView,jdbcType=LONGVARCHAR},
- </if>
- <if test="zookeeperEnabled != null">
- zookeeper_enabled = #{zookeeperEnabled,jdbcType=INTEGER},
- </if>
- <if test="proxyClusterId != null">
- proxy_cluster_id = #{proxyClusterId,jdbcType=INTEGER},
- </if>
</set>
where inlong_group_id = #{inlongGroupId, jdbcType=VARCHAR}
and is_deleted = 0
</update>
-
- <update id="updateByPrimaryKey" parameterType="org.apache.inlong.manager.dao.entity.InlongGroupEntity">
- update inlong_group
- set inlong_group_id = #{inlongGroupId,jdbcType=VARCHAR},
- name = #{name,jdbcType=VARCHAR},
- cn_name = #{cnName,jdbcType=VARCHAR},
- description = #{description,jdbcType=VARCHAR},
- middleware_type = #{middlewareType,jdbcType=VARCHAR},
- queue_module = #{queueModule,jdbcType=VARCHAR},
- topic_partition_num = #{topicPartitionNum,jdbcType=INTEGER},
- mq_resource_obj = #{mqResourceObj,jdbcType=VARCHAR},
- daily_records = #{dailyRecords,jdbcType=INTEGER},
- daily_storage = #{dailyStorage,jdbcType=INTEGER},
- peak_records = #{peakRecords,jdbcType=INTEGER},
- max_length = #{maxLength,jdbcType=INTEGER},
- schema_name = #{schemaName,jdbcType=VARCHAR},
- in_charges = #{inCharges,jdbcType=VARCHAR},
- followers = #{followers,jdbcType=VARCHAR},
- status = #{status,jdbcType=INTEGER},
- is_deleted = #{isDeleted,jdbcType=INTEGER},
- creator = #{creator,jdbcType=VARCHAR},
- modifier = #{modifier,jdbcType=VARCHAR},
- create_time = #{createTime,jdbcType=TIMESTAMP},
- modify_time = #{modifyTime,jdbcType=TIMESTAMP},
- temp_view = #{tempView,jdbcType=LONGVARCHAR},
- zookeeper_enabled = #{zookeeperEnabled,jdbcType=INTEGER},
- proxy_cluster_id = #{proxyClusterId,jdbcType=INTEGER}
- where id = #{id,jdbcType=INTEGER}
- </update>
<update id="updateStatus">
update inlong_group
- set status = #{status,jdbcType=INTEGER},
- modifier = #{modifier,jdbcType=VARCHAR}
+ set previous_status = status,
+ status = #{status, jdbcType=INTEGER},
+ modifier = #{modifier, jdbcType=VARCHAR}
where inlong_group_id = #{groupId, jdbcType=VARCHAR}
and is_deleted = 0
</update>
+ <delete id="deleteByPrimaryKey" parameterType="java.lang.Integer">
+ delete
+ from inlong_group
+ where id = #{id,jdbcType=INTEGER}
+ </delete>
</mapper>
\ No newline at end of file
diff --git a/inlong-manager/manager-dao/src/main/resources/mappers/InlongStreamEntityMapper.xml b/inlong-manager/manager-dao/src/main/resources/mappers/InlongStreamEntityMapper.xml
index ee64e4f3d..a54d87374 100644
--- a/inlong-manager/manager-dao/src/main/resources/mappers/InlongStreamEntityMapper.xml
+++ b/inlong-manager/manager-dao/src/main/resources/mappers/InlongStreamEntityMapper.xml
@@ -22,24 +22,25 @@
<mapper namespace="org.apache.inlong.manager.dao.mapper.InlongStreamEntityMapper">
<resultMap id="BaseResultMap" type="org.apache.inlong.manager.dao.entity.InlongStreamEntity">
<id column="id" jdbcType="INTEGER" property="id"/>
- <result column="inlong_stream_id" jdbcType="VARCHAR" property="inlongStreamId"/>
<result column="inlong_group_id" jdbcType="VARCHAR" property="inlongGroupId"/>
+ <result column="inlong_stream_id" jdbcType="VARCHAR" property="inlongStreamId"/>
<result column="name" jdbcType="VARCHAR" property="name"/>
<result column="description" jdbcType="VARCHAR" property="description"/>
- <result column="mq_resource_obj" jdbcType="VARCHAR" property="mqResourceObj"/>
- <result column="data_source_type" jdbcType="VARCHAR" property="dataSourceType"/>
- <result column="storage_period" jdbcType="INTEGER" property="storagePeriod"/>
+ <result column="mq_resource" jdbcType="VARCHAR" property="mqResource"/>
+
<result column="data_type" jdbcType="VARCHAR" property="dataType"/>
<result column="data_encoding" jdbcType="VARCHAR" property="dataEncoding"/>
<result column="data_separator" jdbcType="VARCHAR" property="dataSeparator"/>
<result column="data_escape_char" jdbcType="VARCHAR" property="dataEscapeChar"/>
- <result column="have_predefined_fields" jdbcType="INTEGER" property="havePredefinedFields"/>
<result column="sync_send" jdbcType="INTEGER" property="syncSend"/>
+
<result column="daily_records" jdbcType="INTEGER" property="dailyRecords"/>
<result column="daily_storage" jdbcType="INTEGER" property="dailyStorage"/>
<result column="peak_records" jdbcType="INTEGER" property="peakRecords"/>
<result column="max_length" jdbcType="INTEGER" property="maxLength"/>
- <result column="in_charges" jdbcType="VARCHAR" property="inCharges"/>
+ <result column="storage_period" jdbcType="INTEGER" property="storagePeriod"/>
+ <result column="ext_params" jdbcType="LONGVARCHAR" property="extParams"/>
+
<result column="status" jdbcType="INTEGER" property="status"/>
<result column="previous_status" jdbcType="INTEGER" property="previousStatus"/>
<result column="is_deleted" jdbcType="INTEGER" property="isDeleted"/>
@@ -47,141 +48,33 @@
<result column="modifier" jdbcType="VARCHAR" property="modifier"/>
<result column="create_time" jdbcType="TIMESTAMP" property="createTime"/>
<result column="modify_time" jdbcType="TIMESTAMP" property="modifyTime"/>
- <result column="temp_view" jdbcType="LONGVARCHAR" property="tempView"/>
</resultMap>
<sql id="Base_Column_List">
- id
- , inlong_stream_id, inlong_group_id, name, description, mq_resource_obj, data_source_type,
- storage_period, data_type, data_encoding, data_separator, data_escape_char, have_predefined_fields,
- sync_send, daily_records, daily_storage, peak_records, max_length, in_charges, status, previous_status,
- is_deleted, creator, modifier, create_time, modify_time, temp_view
+ id, inlong_group_id, inlong_stream_id, name, description, mq_resource,
+ data_type, data_encoding, data_separator, data_escape_char, sync_send,
+ daily_records, daily_storage, peak_records, max_length, storage_period, ext_params,
+ status, previous_status, is_deleted, creator, modifier, create_time, modify_time
</sql>
- <select id="selectByPrimaryKey" parameterType="java.lang.Integer" resultMap="BaseResultMap">
- select
- <include refid="Base_Column_List"/>
- from inlong_stream
- where id = #{id,jdbcType=INTEGER}
- </select>
- <select id="selectByIdentifier" resultType="org.apache.inlong.manager.dao.entity.InlongStreamEntity">
- select
- <include refid="Base_Column_List"/>
- from inlong_stream
- <where>
- is_deleted = 0
- <if test="groupId != null and groupId != ''">
- and inlong_group_id = #{groupId, jdbcType=VARCHAR}
- </if>
- <if test="streamId != null and streamId != ''">
- and inlong_stream_id = #{streamId, jdbcType=VARCHAR}
- </if>
- </where>
- </select>
- <select id="selectExistByIdentifier" resultType="java.lang.Integer">
- select count(1)
- from inlong_stream
- where inlong_group_id = #{groupId, jdbcType=VARCHAR}
- and inlong_stream_id = #{streamId, jdbcType=VARCHAR}
- and is_deleted = 0
- </select>
- <select id="selectByCondition" resultMap="BaseResultMap"
- parameterType="org.apache.inlong.manager.common.pojo.stream.InlongStreamPageRequest">
- select
- distinct stream.id, stream.inlong_group_id, stream.inlong_stream_id, stream.name,
- stream.description, stream.mq_resource_obj, stream.data_source_type, stream.storage_period, stream.data_type,
- stream.data_encoding, stream.data_separator, stream.data_escape_char, stream.have_predefined_fields,
- stream.sync_send,
- stream.daily_records, stream.daily_storage, stream.peak_records, stream.max_length, stream.in_charges,
- stream.status, stream.creator, stream.modifier, stream.create_time, stream.modify_time
- from inlong_stream stream, inlong_group biz, workflow_approver approver
- <where>
- stream.is_deleted = 0
- and approver.process_name = "NEW_GROUP_PROCESS" and approver.is_deleted = 0
- and stream.inlong_group_id = biz.inlong_group_id and biz.is_deleted = 0
- <if test="request.inlongGroupId != null and request.inlongGroupId != ''">
- and stream.inlong_group_id = #{request.inlongGroupId, jdbcType=VARCHAR}
- </if>
- <if test="request.isAdminRole == false">
- and (
- find_in_set(#{request.currentUser, jdbcType=VARCHAR}, biz.in_charges)
- or find_in_set(#{request.currentUser, jdbcType=VARCHAR}, biz.followers)
- or find_in_set(#{request.currentUser, jdbcType=VARCHAR}, approver.approvers)
- )
- </if>
- <if test="request.dataSourceType != null and request.dataSourceType != ''">
- and stream.data_source_type = #{request.dataSourceType, jdbcType=VARCHAR}
- </if>
- <if test="request.keyword != null and request.keyword != ''">
- and (
- stream.inlong_stream_id like CONCAT('%', #{request.keyword}, '%')
- or stream.name like CONCAT('%', #{request.keyword}, '%')
- )
- </if>
- <if test="request.status != null and request.status != ''">
- and stream.status = #{request.status, jdbcType=INTEGER}
- </if>
- </where>
- order by stream.modify_time desc
- </select>
- <select id="selectByGroupId" resultType="org.apache.inlong.manager.dao.entity.InlongStreamEntity">
- select
- <include refid="Base_Column_List"/>
- from inlong_stream
- where inlong_group_id = #{groupId, jdbcType=VARCHAR}
- and is_deleted = 0
- </select>
- <select id="selectCountByGroupId" resultType="java.lang.Integer">
- select count(1)
- from inlong_stream
- <where>
- inlong_group_id = #{groupId,jdbcType=VARCHAR}
- and is_deleted = 0
- </where>
- </select>
- <select id="selectTopicList" resultType="org.apache.inlong.manager.common.pojo.stream.InlongStreamTopicResponse">
- select inlong_stream_id, mq_resource_obj
- from inlong_stream
- <where>
- inlong_group_id = #{groupId, jdbcType=VARCHAR}
- and is_deleted = 0
- </where>
- </select>
-
- <delete id="deleteByPrimaryKey" parameterType="java.lang.Integer">
- delete
- from inlong_stream
- where id = #{id,jdbcType=INTEGER}
- </delete>
- <delete id="deleteAllByGroupId">
- delete
- from inlong_stream
- where inlong_group_id = #{groupId, jdbcType=VARCHAR}
- and is_deleted = 0
- </delete>
<insert id="insert" useGeneratedKeys="true" keyProperty="id"
parameterType="org.apache.inlong.manager.dao.entity.InlongStreamEntity">
- insert into inlong_stream (id, inlong_stream_id, inlong_group_id,
- name, description, mq_resource_obj,
- data_source_type, storage_period, data_type,
- data_encoding, data_separator,
- data_escape_char, have_predefined_fields, sync_send,
- daily_records, daily_storage,
- peak_records, max_length,
- in_charges, status, previous_status,
- is_deleted, creator, modifier,
- create_time, modify_time, temp_view)
- values (#{id,jdbcType=INTEGER}, #{inlongStreamId,jdbcType=VARCHAR}, #{inlongGroupId,jdbcType=VARCHAR},
- #{name,jdbcType=VARCHAR}, #{description,jdbcType=VARCHAR}, #{mqResourceObj,jdbcType=VARCHAR},
- #{dataSourceType,jdbcType=VARCHAR}, #{storagePeriod,jdbcType=INTEGER}, #{dataType,jdbcType=VARCHAR},
- #{dataEncoding,jdbcType=VARCHAR}, #{dataSeparator,jdbcType=VARCHAR},
- #{dataEscapeChar,jdbcType=VARCHAR}, #{havePredefinedFields,jdbcType=INTEGER},
- #{syncSend,jdbcType=INTEGER},
- #{dailyRecords,jdbcType=INTEGER}, #{dailyStorage,jdbcType=INTEGER},
- #{peakRecords,jdbcType=INTEGER}, #{maxLength,jdbcType=INTEGER},
- #{inCharges,jdbcType=VARCHAR}, #{status,jdbcType=INTEGER}, #{previousStatus,jdbcType=INTEGER},
- #{isDeleted,jdbcType=INTEGER}, #{creator,jdbcType=VARCHAR}, #{modifier,jdbcType=VARCHAR},
- #{createTime,jdbcType=TIMESTAMP}, #{modifyTime,jdbcType=TIMESTAMP}, #{tempView,jdbcType=LONGVARCHAR})
+ insert into inlong_stream (id, inlong_group_id, inlong_stream_id,
+ name, description, mq_resource,
+ data_type, data_encoding, data_separator,
+ data_escape_char, sync_send, daily_records,
+ daily_storage, peak_records, max_length,
+ storage_period, ext_params, status,
+ previous_status, is_deleted, creator,
+ modifier, create_time, modify_time)
+ values (#{id,jdbcType=INTEGER}, #{inlongGroupId,jdbcType=VARCHAR}, #{inlongStreamId,jdbcType=VARCHAR},
+ #{name,jdbcType=VARCHAR}, #{description,jdbcType=VARCHAR}, #{mqResource,jdbcType=VARCHAR},
+ #{dataType,jdbcType=VARCHAR}, #{dataEncoding,jdbcType=VARCHAR}, #{dataSeparator,jdbcType=VARCHAR},
+ #{dataEscapeChar,jdbcType=VARCHAR}, #{syncSend,jdbcType=INTEGER}, #{dailyRecords,jdbcType=INTEGER},
+ #{dailyStorage,jdbcType=INTEGER}, #{peakRecords,jdbcType=INTEGER}, #{maxLength,jdbcType=INTEGER},
+ #{storagePeriod,jdbcType=INTEGER}, #{extParams,jdbcType=LONGVARCHAR}, #{status,jdbcType=INTEGER},
+ #{previousStatus,jdbcType=INTEGER}, #{isDeleted,jdbcType=INTEGER}, #{creator,jdbcType=VARCHAR},
+ #{modifier,jdbcType=VARCHAR}, #{createTime,jdbcType=TIMESTAMP}, #{modifyTime,jdbcType=TIMESTAMP})
</insert>
<insert id="insertSelective" useGeneratedKeys="true" keyProperty="id"
parameterType="org.apache.inlong.manager.dao.entity.InlongStreamEntity">
@@ -190,26 +83,20 @@
<if test="id != null">
id,
</if>
- <if test="inlongStreamId != null">
- inlong_stream_id,
- </if>
<if test="inlongGroupId != null">
inlong_group_id,
</if>
+ <if test="inlongStreamId != null">
+ inlong_stream_id,
+ </if>
<if test="name != null">
name,
</if>
<if test="description != null">
description,
</if>
- <if test="mqResourceObj != null">
- mq_resource_obj,
- </if>
- <if test="dataSourceType != null">
- data_source_type,
- </if>
- <if test="storagePeriod != null">
- storage_period,
+ <if test="mqResource != null">
+ mq_resource,
</if>
<if test="dataType != null">
data_type,
@@ -223,9 +110,6 @@
<if test="dataEscapeChar != null">
data_escape_char,
</if>
- <if test="havePredefinedFields != null">
- have_predefined_fields,
- </if>
<if test="syncSend != null">
sync_send,
</if>
@@ -241,8 +125,11 @@
<if test="maxLength != null">
max_length,
</if>
- <if test="inCharges != null">
- in_charges,
+ <if test="storagePeriod != null">
+ storage_period,
+ </if>
+ <if test="extParams != null">
+ ext_params,
</if>
<if test="status != null">
status,
@@ -265,34 +152,25 @@
<if test="modifyTime != null">
modify_time,
</if>
- <if test="tempView != null">
- temp_view,
- </if>
</trim>
<trim prefix="values (" suffix=")" suffixOverrides=",">
<if test="id != null">
#{id,jdbcType=INTEGER},
</if>
- <if test="inlongStreamId != null">
- #{inlongStreamId,jdbcType=VARCHAR},
- </if>
<if test="inlongGroupId != null">
#{inlongGroupId,jdbcType=VARCHAR},
</if>
+ <if test="inlongStreamId != null">
+ #{inlongStreamId,jdbcType=VARCHAR},
+ </if>
<if test="name != null">
#{name,jdbcType=VARCHAR},
</if>
<if test="description != null">
#{description,jdbcType=VARCHAR},
</if>
- <if test="mqResourceObj != null">
- #{mqResourceObj,jdbcType=VARCHAR},
- </if>
- <if test="dataSourceType != null">
- #{dataSourceType,jdbcType=VARCHAR},
- </if>
- <if test="storagePeriod != null">
- #{storagePeriod,jdbcType=INTEGER},
+ <if test="mqResource != null">
+ #{mqResource,jdbcType=VARCHAR},
</if>
<if test="dataType != null">
#{dataType,jdbcType=VARCHAR},
@@ -306,9 +184,6 @@
<if test="dataEscapeChar != null">
#{dataEscapeChar,jdbcType=VARCHAR},
</if>
- <if test="havePredefinedFields != null">
- #{havePredefinedFields,jdbcType=INTEGER},
- </if>
<if test="syncSend != null">
#{syncSend,jdbcType=INTEGER},
</if>
@@ -324,8 +199,11 @@
<if test="maxLength != null">
#{maxLength,jdbcType=INTEGER},
</if>
- <if test="inCharges != null">
- #{inCharges,jdbcType=VARCHAR},
+ <if test="storagePeriod != null">
+ #{storagePeriod,jdbcType=INTEGER},
+ </if>
+ <if test="extParams != null">
+ #{extParams,jdbcType=LONGVARCHAR},
</if>
<if test="status != null">
#{status,jdbcType=INTEGER},
@@ -348,65 +226,133 @@
<if test="modifyTime != null">
#{modifyTime,jdbcType=TIMESTAMP},
</if>
- <if test="tempView != null">
- #{tempView,jdbcType=LONGVARCHAR},
- </if>
</trim>
</insert>
+ <select id="selectByIdentifier" resultType="org.apache.inlong.manager.dao.entity.InlongStreamEntity">
+ select
+ <include refid="Base_Column_List"/>
+ from inlong_stream
+ <where>
+ is_deleted = 0
+ <if test="groupId != null and groupId != ''">
+ and inlong_group_id = #{groupId, jdbcType=VARCHAR}
+ </if>
+ <if test="streamId != null and streamId != ''">
+ and inlong_stream_id = #{streamId, jdbcType=VARCHAR}
+ </if>
+ </where>
+ </select>
+ <select id="selectExistByIdentifier" resultType="java.lang.Integer">
+ select count(1)
+ from inlong_stream
+ where inlong_group_id = #{groupId, jdbcType=VARCHAR}
+ and inlong_stream_id = #{streamId, jdbcType=VARCHAR}
+ and is_deleted = 0
+ </select>
+ <select id="selectByCondition" resultMap="BaseResultMap"
+ parameterType="org.apache.inlong.manager.common.pojo.stream.InlongStreamPageRequest">
+ select
+ distinct stream.id, stream.inlong_group_id, stream.inlong_stream_id, stream.name,
+ stream.description, stream.mq_resource, stream.data_type, stream.data_encoding,
+ stream.data_separator, stream.data_escape_char, stream.sync_send, stream.daily_records,
+ stream.daily_storage, stream.peak_records, stream.max_length, stream.storage_period,
+ stream.status, stream.creator, stream.modifier, stream.create_time, stream.modify_time
+ from inlong_stream stream, inlong_group group, workflow_approver approver
+ <where>
+ stream.is_deleted = 0
+ and approver.process_name = "NEW_GROUP_PROCESS" and approver.is_deleted = 0
+ and stream.inlong_group_id = group.inlong_group_id and group.is_deleted = 0
+ <if test="request.inlongGroupId != null and request.inlongGroupId != ''">
+ and stream.inlong_group_id = #{request.inlongGroupId, jdbcType=VARCHAR}
+ </if>
+ <if test="request.isAdminRole == false">
+ and (
+ find_in_set(#{request.currentUser, jdbcType=VARCHAR}, group.in_charges)
+ or find_in_set(#{request.currentUser, jdbcType=VARCHAR}, group.followers)
+ or find_in_set(#{request.currentUser, jdbcType=VARCHAR}, approver.approvers)
+ )
+ </if>
+ <if test="request.keyword != null and request.keyword != ''">
+ and (
+ stream.inlong_stream_id like CONCAT('%', #{request.keyword}, '%')
+ or stream.name like CONCAT('%', #{request.keyword}, '%')
+ )
+ </if>
+ <if test="request.status != null and request.status != ''">
+ and stream.status = #{request.status, jdbcType=INTEGER}
+ </if>
+ </where>
+ order by stream.modify_time desc
+ </select>
+ <select id="selectByGroupId" resultType="org.apache.inlong.manager.dao.entity.InlongStreamEntity">
+ select
+ <include refid="Base_Column_List"/>
+ from inlong_stream
+ where inlong_group_id = #{groupId, jdbcType=VARCHAR}
+ and is_deleted = 0
+ </select>
+ <select id="selectCountByGroupId" resultType="java.lang.Integer">
+ select count(1)
+ from inlong_stream
+ <where>
+ inlong_group_id = #{groupId,jdbcType=VARCHAR}
+ and is_deleted = 0
+ </where>
+ </select>
+ <select id="selectTopicList" resultType="org.apache.inlong.manager.common.pojo.stream.InlongStreamTopicResponse">
+ select inlong_stream_id, mq_resource
+ from inlong_stream
+ <where>
+ inlong_group_id = #{groupId, jdbcType=VARCHAR}
+ and is_deleted = 0
+ </where>
+ </select>
+
<update id="updateByPrimaryKey" parameterType="org.apache.inlong.manager.dao.entity.InlongStreamEntity">
update inlong_stream
- set inlong_stream_id = #{inlongStreamId,jdbcType=VARCHAR},
- inlong_group_id = #{inlongGroupId,jdbcType=VARCHAR},
- name = #{name,jdbcType=VARCHAR},
- description = #{description,jdbcType=VARCHAR},
- mq_resource_obj = #{mqResourceObj,jdbcType=VARCHAR},
- data_source_type = #{dataSourceType,jdbcType=VARCHAR},
- storage_period = #{storagePeriod,jdbcType=INTEGER},
- data_type = #{dataType,jdbcType=VARCHAR},
- data_encoding = #{dataEncoding,jdbcType=VARCHAR},
- data_separator = #{dataSeparator,jdbcType=VARCHAR},
- data_escape_char = #{dataEscapeChar,jdbcType=VARCHAR},
- have_predefined_fields = #{havePredefinedFields,jdbcType=INTEGER},
- sync_send = #{syncSend,jdbcType=INTEGER},
- daily_records = #{dailyRecords,jdbcType=INTEGER},
- daily_storage = #{dailyStorage,jdbcType=INTEGER},
- peak_records = #{peakRecords,jdbcType=INTEGER},
- max_length = #{maxLength,jdbcType=INTEGER},
- in_charges = #{inCharges,jdbcType=VARCHAR},
- status = #{status,jdbcType=INTEGER},
- previous_status = #{previousStatus,jdbcType=INTEGER},
- is_deleted = #{isDeleted,jdbcType=INTEGER},
- creator = #{creator,jdbcType=VARCHAR},
- modifier = #{modifier,jdbcType=VARCHAR},
- create_time = #{createTime,jdbcType=TIMESTAMP},
- modify_time = #{modifyTime,jdbcType=TIMESTAMP},
- temp_view = #{tempView,jdbcType=LONGVARCHAR}
+ set inlong_group_id = #{inlongGroupId,jdbcType=VARCHAR},
+ inlong_stream_id = #{inlongStreamId,jdbcType=VARCHAR},
+ name = #{name,jdbcType=VARCHAR},
+ description = #{description,jdbcType=VARCHAR},
+ mq_resource = #{mqResource,jdbcType=VARCHAR},
+ data_type = #{dataType,jdbcType=VARCHAR},
+ data_encoding = #{dataEncoding,jdbcType=VARCHAR},
+ data_separator = #{dataSeparator,jdbcType=VARCHAR},
+ data_escape_char = #{dataEscapeChar,jdbcType=VARCHAR},
+ sync_send = #{syncSend,jdbcType=INTEGER},
+ daily_records = #{dailyRecords,jdbcType=INTEGER},
+ daily_storage = #{dailyStorage,jdbcType=INTEGER},
+ peak_records = #{peakRecords,jdbcType=INTEGER},
+ max_length = #{maxLength,jdbcType=INTEGER},
+ storage_period = #{storagePeriod,jdbcType=INTEGER},
+ ext_params = #{extParams,jdbcType=LONGVARCHAR},
+ status = #{status,jdbcType=INTEGER},
+ previous_status = #{previousStatus,jdbcType=INTEGER},
+ is_deleted = #{isDeleted,jdbcType=INTEGER},
+ creator = #{creator,jdbcType=VARCHAR},
+ modifier = #{modifier,jdbcType=VARCHAR},
+ create_time = #{createTime,jdbcType=TIMESTAMP},
+ modify_time = #{modifyTime,jdbcType=TIMESTAMP}
where id = #{id,jdbcType=INTEGER}
</update>
<update id="updateByIdentifierSelective" parameterType="org.apache.inlong.manager.dao.entity.InlongStreamEntity">
update inlong_stream
<set>
- <if test="inlongStreamId != null">
- inlong_stream_id = #{inlongStreamId,jdbcType=VARCHAR},
- </if>
<if test="inlongGroupId != null">
inlong_group_id = #{inlongGroupId,jdbcType=VARCHAR},
</if>
+ <if test="inlongStreamId != null">
+ inlong_stream_id = #{inlongStreamId,jdbcType=VARCHAR},
+ </if>
<if test="name != null">
name = #{name,jdbcType=VARCHAR},
</if>
<if test="description != null">
description = #{description,jdbcType=VARCHAR},
</if>
- <if test="mqResourceObj != null">
- mq_resource_obj = #{mqResourceObj,jdbcType=VARCHAR},
- </if>
- <if test="dataSourceType != null">
- data_source_type = #{dataSourceType,jdbcType=VARCHAR},
- </if>
- <if test="storagePeriod != null">
- storage_period = #{storagePeriod,jdbcType=INTEGER},
+ <if test="mqResource != null">
+ mq_resource = #{mqResource,jdbcType=VARCHAR},
</if>
<if test="dataType != null">
data_type = #{dataType,jdbcType=VARCHAR},
@@ -420,9 +366,6 @@
<if test="dataEscapeChar != null">
data_escape_char = #{dataEscapeChar,jdbcType=VARCHAR},
</if>
- <if test="havePredefinedFields != null">
- have_predefined_fields = #{havePredefinedFields,jdbcType=INTEGER},
- </if>
<if test="syncSend != null">
sync_send = #{syncSend,jdbcType=INTEGER},
</if>
@@ -438,8 +381,11 @@
<if test="maxLength != null">
max_length= #{maxLength,jdbcType=INTEGER},
</if>
- <if test="inCharges != null">
- in_charges = #{inCharges,jdbcType=VARCHAR},
+ <if test="storagePeriod != null">
+ storage_period = #{storagePeriod,jdbcType=INTEGER},
+ </if>
+ <if test="extParams != null">
+ ext_params = #{extParams,jdbcType=LONGVARCHAR},
</if>
<if test="status != null">
status = #{status,jdbcType=INTEGER},
@@ -462,9 +408,6 @@
<if test="modifyTime != null">
modify_time = #{modifyTime,jdbcType=TIMESTAMP},
</if>
- <if test="tempView != null">
- temp_view = #{tempView,jdbcType=LONGVARCHAR},
- </if>
</set>
where inlong_group_id = #{inlongGroupId, jdbcType=VARCHAR}
and inlong_stream_id = #{inlongStreamId, jdbcType=VARCHAR}
@@ -472,14 +415,13 @@
</update>
<update id="updateStatusByIdentifier">
update inlong_stream
- set status = #{status, jdbcType=INTEGER}, modifier = #{modifier, jdbcType=VARCHAR}
- <where>
- inlong_group_id = #{groupId, jdbcType=VARCHAR}
- <if test="streamId != null and streamId != ''">
- and inlong_stream_id = #{streamId, jdbcType=VARCHAR}
- </if>
- and is_deleted = 0
- </where>
+ set status = #{status, jdbcType=INTEGER},
+ modifier = #{modifier, jdbcType=VARCHAR}
+ where inlong_group_id = #{groupId, jdbcType=VARCHAR}
+ <if test="streamId != null and streamId != ''">
+ and inlong_stream_id = #{streamId, jdbcType=VARCHAR}
+ </if>
+ and is_deleted = 0
</update>
<update id="logicDeleteDlqOrRlq">
update inlong_stream
@@ -490,4 +432,15 @@
and is_deleted = 0
</update>
+ <delete id="deleteByPrimaryKey" parameterType="java.lang.Integer">
+ delete
+ from inlong_stream
+ where id = #{id,jdbcType=INTEGER}
+ </delete>
+ <delete id="deleteAllByGroupId">
+ delete
+ from inlong_stream
+ where inlong_group_id = #{groupId, jdbcType=VARCHAR}
+ and is_deleted = 0
+ </delete>
</mapper>
\ No newline at end of file
diff --git a/inlong-manager/manager-dao/src/main/resources/mappers/StreamSinkEntityMapper.xml b/inlong-manager/manager-dao/src/main/resources/mappers/StreamSinkEntityMapper.xml
index 5430e7fb3..c0b233a26 100644
--- a/inlong-manager/manager-dao/src/main/resources/mappers/StreamSinkEntityMapper.xml
+++ b/inlong-manager/manager-dao/src/main/resources/mappers/StreamSinkEntityMapper.xml
@@ -189,8 +189,7 @@
sink.status,
sink.creator,
sink.sink_name,
- stream.mq_resource_obj,
- stream.data_source_type,
+ stream.mq_resource,
stream.data_type,
stream.data_separator as sourceSeparator,
stream.data_escape_char
diff --git a/inlong-manager/manager-dao/src/test/java/org/apache/inlong/manager/dao/mapper/InlongGroupEntityMapperTest.java b/inlong-manager/manager-dao/src/test/java/org/apache/inlong/manager/dao/mapper/InlongGroupEntityMapperTest.java
index 2f82b0624..efbaf18b7 100644
--- a/inlong-manager/manager-dao/src/test/java/org/apache/inlong/manager/dao/mapper/InlongGroupEntityMapperTest.java
+++ b/inlong-manager/manager-dao/src/test/java/org/apache/inlong/manager/dao/mapper/InlongGroupEntityMapperTest.java
@@ -41,14 +41,6 @@ public class InlongGroupEntityMapperTest extends DaoBaseTest {
Assert.assertNull(groupEntityMapper.selectByGroupId(entity.getInlongGroupId()));
}
- @Test
- public void insertSelective() {
- InlongGroupEntity entity = createHeartbeatEntity();
- groupEntityMapper.insertSelective(entity);
- InlongGroupEntity queryResult = groupEntityMapper.selectByGroupId(entity.getInlongGroupId());
- Assert.assertNotNull(queryResult);
- }
-
@Test
public void selectByPrimaryKey() {
InlongGroupEntity entity = createHeartbeatEntity();
@@ -59,7 +51,7 @@ public class InlongGroupEntityMapperTest extends DaoBaseTest {
private InlongGroupEntity createHeartbeatEntity() {
InlongGroupEntity entity = new InlongGroupEntity();
entity.setInlongGroupId("test_group");
- entity.setMqResourceObj("test_group");
+ entity.setMqResource("test_group");
entity.setInCharges("admin");
entity.setCreator("admin");
entity.setCreateTime(new Date());
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/ConsumptionServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/ConsumptionServiceImpl.java
index cc5bc9256..476d633d6 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/ConsumptionServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/ConsumptionServiceImpl.java
@@ -308,7 +308,7 @@ public class ConsumptionServiceImpl implements ConsumptionService {
private String getFullPulsarTopic(String groupId, String topic) {
InlongGroupEntity inlongGroupEntity = groupMapper.selectByGroupId(groupId);
String tenant = clusterBean.getDefaultTenant();
- String namespace = inlongGroupEntity.getMqResourceObj();
+ String namespace = inlongGroupEntity.getMqResource();
return "persistent://" + tenant + "/" + namespace + "/" + topic;
}
@@ -342,8 +342,8 @@ public class ConsumptionServiceImpl implements ConsumptionService {
}
@Override
- public void saveSortConsumption(InlongGroupInfo bizInfo, String topic, String consumerGroup) {
- String groupId = bizInfo.getInlongGroupId();
+ public void saveSortConsumption(InlongGroupInfo groupInfo, String topic, String consumerGroup) {
+ String groupId = groupInfo.getInlongGroupId();
ConsumptionEntity exists = consumptionMapper.selectConsumptionExists(groupId, topic, consumerGroup);
if (exists != null) {
log.warn("consumption with groupId={}, topic={}, consumer group={} already exists, skip to create",
@@ -352,19 +352,19 @@ public class ConsumptionServiceImpl implements ConsumptionService {
}
log.debug("begin to save consumption, groupId={}, topic={}, consumer group={}", groupId, topic, consumerGroup);
- MQType mqType = MQType.forType(bizInfo.getMiddlewareType());
+ MQType mqType = MQType.forType(groupInfo.getMqType());
ConsumptionEntity entity = new ConsumptionEntity();
entity.setInlongGroupId(groupId);
entity.setMiddlewareType(mqType.getType());
entity.setTopic(topic);
entity.setConsumerGroupId(consumerGroup);
entity.setConsumerGroupName(consumerGroup);
- entity.setInCharges(bizInfo.getInCharges());
+ entity.setInCharges(groupInfo.getInCharges());
entity.setFilterEnabled(0);
entity.setStatus(ConsumptionStatus.APPROVED.getStatus());
entity.setIsDeleted(GlobalConstants.UN_DELETED);
- entity.setCreator(bizInfo.getCreator());
+ entity.setCreator(groupInfo.getCreator());
entity.setCreateTime(new Date());
consumptionMapper.insert(entity);
@@ -441,9 +441,9 @@ public class ConsumptionServiceImpl implements ConsumptionService {
Preconditions.checkNotNull(topicVO, "inlong group not exist: " + groupId);
// Tube’s topic is the inlong group level, one inlong group, one Tube topic
- MQType mqType = MQType.forType(topicVO.getMiddlewareType());
+ MQType mqType = MQType.forType(topicVO.getMqType());
if (mqType == MQType.TUBE) {
- String bizTopic = topicVO.getMqResourceObj();
+ String bizTopic = topicVO.getMqResource();
Preconditions.checkTrue(bizTopic == null || bizTopic.equals(info.getTopic()),
"topic [" + info.getTopic() + "] not belong to inlong group " + groupId);
} else if (mqType == MQType.PULSAR || mqType == MQType.TDMQ_PULSAR) {
@@ -452,7 +452,7 @@ public class ConsumptionServiceImpl implements ConsumptionService {
List<InlongStreamTopicResponse> dsTopicList = topicVO.getDsTopicList();
if (dsTopicList != null && dsTopicList.size() > 0) {
Set<String> topicSet = new HashSet<>(Arrays.asList(info.getTopic().split(",")));
- dsTopicList.forEach(ds -> topicSet.remove(ds.getMqResourceObj()));
+ dsTopicList.forEach(ds -> topicSet.remove(ds.getMqResource()));
Preconditions.checkEmpty(topicSet, "topic [" + topicSet + "] not belong to inlong group " + groupId);
}
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongGroupServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongGroupServiceImpl.java
index 24be9d000..27b55bd38 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongGroupServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongGroupServiceImpl.java
@@ -24,7 +24,6 @@ import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
-import org.apache.inlong.manager.common.enums.Constant;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.GlobalConstants;
import org.apache.inlong.manager.common.enums.GroupStatus;
@@ -110,11 +109,9 @@ public class InlongGroupServiceImpl implements InlongGroupService {
// Processing inlong group and extended information
InlongGroupEntity entity = CommonBeanUtils.copyProperties(groupInfo, InlongGroupEntity::new);
entity.setInlongGroupId(groupId);
- if (StringUtils.isEmpty(entity.getMqResourceObj())) {
- entity.setMqResourceObj(groupId);
+ if (StringUtils.isEmpty(entity.getMqResource())) {
+ entity.setMqResource(groupId);
}
- // Only M0 is currently supported
- entity.setSchemaName(Constant.SCHEMA_M0_DAY);
// After saving, the status is set to [GROUP_WAIT_SUBMIT]
entity.setStatus(GroupStatus.TO_BE_SUBMIT.getCode());
@@ -126,10 +123,10 @@ public class InlongGroupServiceImpl implements InlongGroupService {
entity.setModifier(operator);
}
entity.setCreateTime(new Date());
- groupMapper.insertSelective(entity);
+ groupMapper.insert(entity);
this.saveOrUpdateExt(groupId, groupInfo.getExtList());
// Saving MQ information.
- Middleware mqMiddleware = groupMqFactory.getMqMiddleware(MQType.forType(groupInfo.getMiddlewareType()));
+ Middleware mqMiddleware = groupMqFactory.getMqMiddleware(MQType.forType(groupInfo.getMqType()));
if (groupInfo.getMqExtInfo() != null && StringUtils.isBlank(groupInfo.getMqExtInfo().getInlongGroupId())) {
groupInfo.getMqExtInfo().setInlongGroupId(groupId);
}
@@ -155,7 +152,7 @@ public class InlongGroupServiceImpl implements InlongGroupService {
groupInfo.setExtList(extInfoList);
// If the middleware is Pulsar, we need to encapsulate Pulsar related data
- MQType mqType = MQType.forType(entity.getMiddlewareType());
+ MQType mqType = MQType.forType(entity.getMqType());
Middleware mq = groupMqFactory.getMqMiddleware(mqType);
groupInfo.setMqExtInfo(mq.get(groupId));
@@ -227,7 +224,7 @@ public class InlongGroupServiceImpl implements InlongGroupService {
this.saveOrUpdateExt(groupId, groupRequest.getExtList());
// Update the MQ info
- MQType mqType = MQType.forType(groupRequest.getMiddlewareType());
+ MQType mqType = MQType.forType(groupRequest.getMqType());
Middleware mqMiddleware = groupMqFactory.getMqMiddleware(mqType);
InlongGroupMqExtBase mqExtInfo = groupRequest.getMqExtInfo();
if (mqExtInfo != null && StringUtils.isBlank(mqExtInfo.getInlongGroupId())) {
@@ -336,7 +333,7 @@ public class InlongGroupServiceImpl implements InlongGroupService {
// To logically delete the associated extension table
groupExtMapper.logicDeleteAllByGroupId(groupId);
- groupMqFactory.getMqMiddleware(MQType.forType(entity.getMiddlewareType())).delete(groupId);
+ groupMqFactory.getMqMiddleware(MQType.forType(entity.getMqType())).delete(groupId);
LOGGER.info("success to delete inlong group and inlong group ext property for groupId={}", groupId);
return true;
}
@@ -377,7 +374,7 @@ public class InlongGroupServiceImpl implements InlongGroupService {
public InlongGroupTopicResponse getTopic(String groupId) {
LOGGER.debug("begin to get topic by groupId={}", groupId);
InlongGroupInfo groupInfo = this.get(groupId);
- MQType mqType = MQType.forType(groupInfo.getMiddlewareType());
+ MQType mqType = MQType.forType(groupInfo.getMqType());
Middleware mqMiddleware = groupMqFactory.getMqMiddleware(mqType);
return mqMiddleware.getTopic(groupInfo);
}
@@ -391,8 +388,8 @@ public class InlongGroupServiceImpl implements InlongGroupService {
Preconditions.checkNotNull(approveInfo, "InlongGroupApproveRequest is empty");
String groupId = approveInfo.getInlongGroupId();
Preconditions.checkNotNull(groupId, ErrorCodeEnum.GROUP_ID_IS_EMPTY.getMessage());
- String middlewareType = approveInfo.getMiddlewareType();
- Preconditions.checkNotNull(middlewareType, "Middleware type is empty");
+ String mqType = approveInfo.getMqType();
+ Preconditions.checkNotNull(mqType, "MQ type cannot by empty");
// Update status to [GROUP_APPROVE_PASSED]
// If you need to change inlong group info after approve, just do in here
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongStreamServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongStreamServiceImpl.java
index 8e1d935c4..d276a0a43 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongStreamServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/InlongStreamServiceImpl.java
@@ -106,8 +106,8 @@ public class InlongStreamServiceImpl implements InlongStreamService {
LOGGER.error("inlong stream id [{}] has already exists", streamId);
throw new BusinessException(ErrorCodeEnum.STREAM_ID_DUPLICATE);
}
- if (StringUtils.isEmpty(request.getMqResourceObj())) {
- request.setMqResourceObj(streamId);
+ if (StringUtils.isEmpty(request.getMqResource())) {
+ request.setMqResource(streamId);
}
// Processing inlong stream
InlongStreamEntity streamEntity = CommonBeanUtils.copyProperties(request, InlongStreamEntity::new);
@@ -536,7 +536,7 @@ public class InlongStreamServiceImpl implements InlongStreamService {
InlongStreamEntity streamEntity = new InlongStreamEntity();
streamEntity.setInlongGroupId(groupId);
streamEntity.setInlongStreamId(topicName);
- streamEntity.setMqResourceObj(topicName);
+ streamEntity.setMqResource(topicName);
streamEntity.setDescription("This is DLQ / RLQ topic created by SYSTEM");
streamEntity.setDailyRecords(1000);
streamEntity.setDailyStorage(1000);
@@ -648,7 +648,7 @@ public class InlongStreamServiceImpl implements InlongStreamService {
}
/**
- * Check that groupId, streamId, and dataSourceType are not allowed to be modified
+ * Check that groupId, streamId are not allowed to be modified
*/
private void checkUpdatedFields(InlongStreamEntity streamEntity, InlongStreamRequest request) {
String newGroupId = request.getInlongGroupId();
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/ThirdPartyClusterServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/ThirdPartyClusterServiceImpl.java
index 1b581d14b..a01c9d1ee 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/ThirdPartyClusterServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/ThirdPartyClusterServiceImpl.java
@@ -69,6 +69,7 @@ import java.util.Map;
@Service
public class ThirdPartyClusterServiceImpl implements ThirdPartyClusterService {
+ public static final String SCHEMA_M0_DAY = "m0_day";
private static final Logger LOGGER = LoggerFactory.getLogger(ThirdPartyClusterServiceImpl.class);
private static final Gson GSON = new Gson();
@@ -253,18 +254,18 @@ public class ThirdPartyClusterServiceImpl implements ThirdPartyClusterService {
List<DataProxyConfig> configList = new ArrayList<>();
for (InlongGroupEntity groupEntity : groupEntityList) {
String groupId = groupEntity.getInlongGroupId();
- String bizResource = groupEntity.getMqResourceObj();
+ String bizResource = groupEntity.getMqResource();
DataProxyConfig config = new DataProxyConfig();
- config.setM(groupEntity.getSchemaName());
- MQType mqType = MQType.forType(groupEntity.getMiddlewareType());
+ config.setM(SCHEMA_M0_DAY);
+ MQType mqType = MQType.forType(groupEntity.getMqType());
if (mqType == MQType.TUBE) {
config.setInlongGroupId(groupId);
config.setTopic(bizResource);
} else if (mqType == MQType.PULSAR || mqType == MQType.TDMQ_PULSAR) {
List<InlongStreamEntity> streamList = streamMapper.selectByGroupId(groupId);
for (InlongStreamEntity stream : streamList) {
- String topic = stream.getMqResourceObj();
+ String topic = stream.getMqResource();
String streamId = stream.getInlongStreamId();
config.setInlongGroupId(groupId + "/" + streamId);
config.setTopic("persistent://" + clusterBean.getDefaultTenant() + "/" + bizResource + "/" + topic);
@@ -298,21 +299,21 @@ public class ThirdPartyClusterServiceImpl implements ThirdPartyClusterService {
// third-party-cluster type
String mqType = "";
if (!groupEntityList.isEmpty()) {
- mqType = groupEntityList.get(0).getMiddlewareType();
+ mqType = groupEntityList.get(0).getMqType();
}
// Get topic list by group id
List<DataProxyConfig> topicList = new ArrayList<>();
for (InlongGroupEntity groupEntity : groupEntityList) {
final String groupId = groupEntity.getInlongGroupId();
- final String mqResource = groupEntity.getMqResourceObj();
+ final String mqResource = groupEntity.getMqResource();
MQType type = MQType.forType(mqType);
if (type == MQType.PULSAR || type == MQType.TDMQ_PULSAR) {
List<InlongStreamEntity> streamList = streamMapper.selectByGroupId(groupId);
for (InlongStreamEntity stream : streamList) {
DataProxyConfig topicConfig = new DataProxyConfig();
String streamId = stream.getInlongStreamId();
- String topic = stream.getMqResourceObj();
+ String topic = stream.getMqResource();
String tenant = clusterBean.getDefaultTenant();
InlongGroupPulsarEntity pulsarEntity = pulsarEntityMapper.selectByGroupId(groupId);
if (pulsarEntity != null && StringUtils.isNotEmpty(pulsarEntity.getTenant())) {
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/mq/BlankMiddleware.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/mq/BlankMiddleware.java
index fdc2c1bde..81848356e 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/mq/BlankMiddleware.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/mq/BlankMiddleware.java
@@ -45,7 +45,7 @@ public class BlankMiddleware implements Middleware {
@Override
public InlongGroupMqExtBase get(String groupId) {
InlongGroupMqExtBase mqExtBase = new InlongGroupMqExtBase();
- mqExtBase.setMiddlewareType(MQType.NONE.getType());
+ mqExtBase.setMqType(MQType.NONE.getType());
mqExtBase.setInlongGroupId(groupId);
return mqExtBase;
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/mq/MiddlewareFactory.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/mq/MiddlewareFactory.java
index 112c30a8d..c67acd0ee 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/mq/MiddlewareFactory.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/mq/MiddlewareFactory.java
@@ -39,7 +39,6 @@ public class MiddlewareFactory {
* Getting InLong Group Mq middle ware implementation.
*
* @param type MQ TYPE.
- * @return
*/
public Middleware getMqMiddleware(MQType type) {
for (Middleware middleware : middlewares) {
@@ -47,7 +46,7 @@ public class MiddlewareFactory {
return middleware;
}
}
- throw new BusinessException(ErrorCodeEnum.MIDDLEWARE_TYPE_NOT_SUPPORTED,
+ throw new BusinessException(ErrorCodeEnum.MQ_TYPE_NOT_SUPPORTED,
"Current version of InLong not support middleware type of MQ:" + type.name());
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/mq/PulsarMiddleware.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/mq/PulsarMiddleware.java
index 0e7b1c6e2..276ea7ba3 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/mq/PulsarMiddleware.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/mq/PulsarMiddleware.java
@@ -97,7 +97,7 @@ public class PulsarMiddleware implements Middleware {
Preconditions.checkNotNull(pulsarEntity, "Pulsar info not found by the groupId=" + groupId);
InlongGroupPulsarInfo pulsarInfo = CommonBeanUtils.copyProperties(pulsarEntity,
InlongGroupPulsarInfo::new);
- pulsarInfo.setMiddlewareType(type().name());
+ pulsarInfo.setMqType(type().name());
return pulsarInfo;
}
@@ -126,7 +126,7 @@ public class PulsarMiddleware implements Middleware {
topicVO.setPulsarAdminUrl(commonOperateService.getSpecifiedParam(InlongGroupSettings.PULSAR_ADMIN_URL));
topicVO.setPulsarServiceUrl(commonOperateService.getSpecifiedParam(InlongGroupSettings.PULSAR_SERVICE_URL));
topicVO.setInlongGroupId(groupInfo.getInlongGroupId());
- topicVO.setMiddlewareType(type().name());
+ topicVO.setMqType(type().name());
return topicVO;
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/mq/TubeMiddleware.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/mq/TubeMiddleware.java
index 5533d354b..5a7dc6833 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/mq/TubeMiddleware.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/mq/TubeMiddleware.java
@@ -51,7 +51,7 @@ public class TubeMiddleware implements Middleware {
@Override
public InlongGroupMqExtBase get(String groupId) {
InlongGroupMqExtBase mqExtBase = new InlongGroupMqExtBase();
- mqExtBase.setMiddlewareType(MQType.TUBE.getType());
+ mqExtBase.setMqType(MQType.TUBE.getType());
mqExtBase.setInlongGroupId(groupId);
return mqExtBase;
}
@@ -65,10 +65,10 @@ public class TubeMiddleware implements Middleware {
@Override
public InlongGroupTopicResponse getTopic(InlongGroupInfo groupInfo) {
InlongGroupTopicResponse topicVO = new InlongGroupTopicResponse();
- topicVO.setMqResourceObj(groupInfo.getMqResourceObj());
+ topicVO.setMqResource(groupInfo.getMqResource());
topicVO.setTubeMasterUrl(commonOperateService.getSpecifiedParam(InlongGroupSettings.TUBE_MASTER_URL));
topicVO.setInlongGroupId(groupInfo.getInlongGroupId());
- topicVO.setMiddlewareType(type().name());
+ topicVO.setMqType(type().name());
return topicVO;
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreatePulsarGroupForStreamTaskListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreatePulsarGroupForStreamTaskListener.java
index 90e230023..3732d9b22 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreatePulsarGroupForStreamTaskListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreatePulsarGroupForStreamTaskListener.java
@@ -86,7 +86,7 @@ public class CreatePulsarGroupForStreamTaskListener implements QueueOperateListe
log.warn("inlong stream is empty for group={}, stream={}, skip to create pulsar group", groupId, streamId);
return ListenerResult.success();
}
- PulsarClusterInfo globalCluster = commonOperateService.getPulsarClusterInfo(groupInfo.getMiddlewareType());
+ PulsarClusterInfo globalCluster = commonOperateService.getPulsarClusterInfo(groupInfo.getMqType());
try (PulsarAdmin globalPulsarAdmin = PulsarUtils.getPulsarAdmin(globalCluster)) {
// Query data sink info based on groupId and streamId
List<String> sinkTypeList = sinkService.getSinkTypeList(groupId, streamId);
@@ -98,14 +98,14 @@ public class CreatePulsarGroupForStreamTaskListener implements QueueOperateListe
PulsarTopicBean topicBean = new PulsarTopicBean();
topicBean.setTenant(clusterBean.getDefaultTenant());
- topicBean.setNamespace(groupInfo.getMqResourceObj());
- String topic = streamEntity.getMqResourceObj();
+ topicBean.setNamespace(groupInfo.getMqResource());
+ String topic = streamEntity.getMqResource();
topicBean.setTopicName(topic);
List<String> pulsarClusters = PulsarUtils.getPulsarClusters(globalPulsarAdmin);
// Create a subscription in the Pulsar cluster (cross-region), you need to ensure that the Topic exists
String tenant = clusterBean.getDefaultTenant();
- String namespace = groupInfo.getMqResourceObj();
+ String namespace = groupInfo.getMqResource();
for (String cluster : pulsarClusters) {
String serviceUrl = PulsarUtils.getServiceUrl(globalPulsarAdmin, cluster);
PulsarClusterInfo pulsarClusterInfo = PulsarClusterInfo.builder()
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreatePulsarGroupTaskListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreatePulsarGroupTaskListener.java
index 87b67bbd2..2947aaa08 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreatePulsarGroupTaskListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreatePulsarGroupTaskListener.java
@@ -82,16 +82,16 @@ public class CreatePulsarGroupTaskListener implements QueueOperateListener {
log.warn("inlong stream is empty for groupId={}, skip to create pulsar subscription", groupId);
return ListenerResult.success();
}
- PulsarClusterInfo globalCluster = commonOperateService.getPulsarClusterInfo(groupInfo.getMiddlewareType());
+ PulsarClusterInfo globalCluster = commonOperateService.getPulsarClusterInfo(groupInfo.getMqType());
try (PulsarAdmin globalPulsarAdmin = PulsarUtils.getPulsarAdmin(globalCluster)) {
String tenant = clusterBean.getDefaultTenant();
- String namespace = groupInfo.getMqResourceObj();
+ String namespace = groupInfo.getMqResource();
for (InlongStreamEntity streamEntity : streamEntities) {
PulsarTopicBean topicBean = new PulsarTopicBean();
topicBean.setTenant(tenant);
topicBean.setNamespace(namespace);
- String topic = streamEntity.getMqResourceObj();
+ String topic = streamEntity.getMqResource();
topicBean.setTopicName(topic);
List<String> pulsarClusters = PulsarUtils.getPulsarClusters(globalPulsarAdmin);
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreatePulsarResourceTaskListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreatePulsarResourceTaskListener.java
index e9181dd34..159f07c6a 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreatePulsarResourceTaskListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreatePulsarResourceTaskListener.java
@@ -77,7 +77,7 @@ public class CreatePulsarResourceTaskListener implements QueueOperateListener {
if (groupInfo == null) {
throw new WorkflowListenerException("inlong group or pulsar cluster not found for groupId=" + groupId);
}
- PulsarClusterInfo globalCluster = commonOperateService.getPulsarClusterInfo(groupInfo.getMiddlewareType());
+ PulsarClusterInfo globalCluster = commonOperateService.getPulsarClusterInfo(groupInfo.getMqType());
try (PulsarAdmin globalPulsarAdmin = PulsarUtils.getPulsarAdmin(globalCluster)) {
List<String> pulsarClusters = PulsarUtils.getPulsarClusters(globalPulsarAdmin);
for (String cluster : pulsarClusters) {
@@ -102,7 +102,7 @@ public class CreatePulsarResourceTaskListener implements QueueOperateListener {
String groupId = groupInfo.getInlongGroupId();
log.info("begin to create pulsar resource for groupId={} in cluster={}", groupId, pulsarClusterInfo);
- String namespace = groupInfo.getMqResourceObj();
+ String namespace = groupInfo.getMqResource();
Preconditions.checkNotNull(namespace, "pulsar namespace cannot be empty for groupId=" + groupId);
String queueModule = groupInfo.getQueueModule();
Preconditions.checkNotNull(queueModule, "queue module cannot be empty for groupId=" + groupId);
@@ -123,7 +123,7 @@ public class CreatePulsarResourceTaskListener implements QueueOperateListener {
.tenant(tenant).namespace(namespace).numPartitions(partitionNum).queueModule(queueModule).build();
for (InlongStreamTopicResponse topicVO : streamTopicList) {
- topicBean.setTopicName(topicVO.getMqResourceObj());
+ topicBean.setTopicName(topicVO.getMqResource());
pulsarOptService.createTopic(pulsarAdmin, topicBean);
}
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreatePulsarTopicForStreamTaskListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreatePulsarTopicForStreamTaskListener.java
index cd1797658..278c8525d 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreatePulsarTopicForStreamTaskListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreatePulsarTopicForStreamTaskListener.java
@@ -76,14 +76,14 @@ public class CreatePulsarTopicForStreamTaskListener implements QueueOperateListe
}
log.info("begin to create pulsar topic for groupId={}, streamId={}", groupId, streamId);
- PulsarClusterInfo globalCluster = commonOperateService.getPulsarClusterInfo(groupInfo.getMiddlewareType());
+ PulsarClusterInfo globalCluster = commonOperateService.getPulsarClusterInfo(groupInfo.getMqType());
try (PulsarAdmin globalPulsarAdmin = PulsarUtils.getPulsarAdmin(globalCluster)) {
List<String> pulsarClusters = PulsarUtils.getPulsarClusters(globalPulsarAdmin);
for (String cluster : pulsarClusters) {
String serviceUrl = PulsarUtils.getServiceUrl(globalPulsarAdmin, cluster);
PulsarClusterInfo pulsarClusterInfo = PulsarClusterInfo.builder()
.token(globalCluster.getToken()).adminUrl(serviceUrl).build();
- String pulsarTopic = streamEntity.getMqResourceObj();
+ String pulsarTopic = streamEntity.getMqResource();
this.createTopic(groupInfo, pulsarTopic, pulsarClusterInfo);
}
} catch (Exception e) {
@@ -107,7 +107,7 @@ public class CreatePulsarTopicForStreamTaskListener implements QueueOperateListe
try (PulsarAdmin pulsarAdmin = PulsarUtils.getPulsarAdmin(pulsarClusterInfo)) {
PulsarTopicBean topicBean = PulsarTopicBean.builder()
.tenant(clusterBean.getDefaultTenant())
- .namespace(bizInfo.getMqResourceObj())
+ .namespace(bizInfo.getMqResource())
.topicName(pulsarTopic)
.numPartitions(partition)
.queueModule(bizInfo.getQueueModule())
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreateTubeGroupTaskListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreateTubeGroupTaskListener.java
index 3f59958ce..b88e52b97 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreateTubeGroupTaskListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreateTubeGroupTaskListener.java
@@ -66,7 +66,7 @@ public class CreateTubeGroupTaskListener implements QueueOperateListener {
log.info("try to create consumer group for groupId {}", groupId);
InlongGroupInfo groupInfo = groupService.get(groupId);
- String topicName = groupInfo.getMqResourceObj();
+ String topicName = groupInfo.getMqResource();
int clusterId = Integer.parseInt(commonOperateService.getSpecifiedParam(InlongGroupSettings.TUBE_CLUSTER_ID));
QueryTubeTopicRequest queryTubeTopicRequest = QueryTubeTopicRequest.builder()
.topicName(topicName).clusterId(clusterId)
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreateTubeTopicTaskListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreateTubeTopicTaskListener.java
index c7a796c1f..db7e82da5 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreateTubeTopicTaskListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/CreateTubeTopicTaskListener.java
@@ -58,7 +58,7 @@ public class CreateTubeTopicTaskListener implements QueueOperateListener {
try {
InlongGroupInfo groupInfo = groupService.get(groupId);
- String topicName = groupInfo.getMqResourceObj();
+ String topicName = groupInfo.getMqResource();
AddTubeMqTopicRequest request = new AddTubeMqTopicRequest();
request.setUser("inlong-manager");
AddTubeMqTopicRequest.AddTopicTasksBean tasksBean = new AddTubeMqTopicRequest.AddTopicTasksBean();
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/PulsarEventSelector.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/PulsarEventSelector.java
index dcf7c2bf5..e56426aca 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/PulsarEventSelector.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/PulsarEventSelector.java
@@ -41,7 +41,7 @@ public class PulsarEventSelector implements EventSelector {
GroupResourceProcessForm form = (GroupResourceProcessForm) processForm;
String groupId = form.getInlongGroupId();
- MQType mqType = MQType.forType(form.getGroupInfo().getMiddlewareType());
+ MQType mqType = MQType.forType(form.getGroupInfo().getMqType());
if (mqType == MQType.PULSAR || mqType == MQType.TDMQ_PULSAR) {
InlongGroupPulsarInfo pulsarInfo = (InlongGroupPulsarInfo) form.getGroupInfo().getMqExtInfo();
boolean enable = GlobalConstants.ENABLE_CREATE_RESOURCE.equals(pulsarInfo.getEnableCreateResource());
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/TubeEventSelector.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/TubeEventSelector.java
index ef1fbd431..235afd314 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/TubeEventSelector.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/mq/TubeEventSelector.java
@@ -38,7 +38,7 @@ public class TubeEventSelector implements EventSelector {
}
GroupResourceProcessForm form = (GroupResourceProcessForm) processForm;
String groupId = form.getInlongGroupId();
- MQType mqType = MQType.forType(form.getGroupInfo().getMiddlewareType());
+ MQType mqType = MQType.forType(form.getGroupInfo().getMqType());
if (mqType == MQType.TUBE) {
log.info("need to create tube resource for groupId [{}]", groupId);
return true;
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/ZookeeperDisabledSelector.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/ZookeeperDisabledSelector.java
index 1657a6d8b..75ad6d937 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/ZookeeperDisabledSelector.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/ZookeeperDisabledSelector.java
@@ -41,8 +41,8 @@ public class ZookeeperDisabledSelector implements EventSelector {
GroupResourceProcessForm groupResourceForm = (GroupResourceProcessForm) processForm;
InlongGroupInfo groupInfo = groupResourceForm.getGroupInfo();
- boolean enable = groupInfo.getZookeeperEnabled() == 0
- && MQType.forType(groupInfo.getMiddlewareType()) != MQType.NONE;
+ boolean enable = groupInfo.getEnableZookeeper() == 0
+ && MQType.forType(groupInfo.getMqType()) != MQType.NONE;
log.info("zookeeper disabled was [{}] for groupId [{}]", enable, groupId);
return enable;
} else {
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/ZookeeperEnabledSelector.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/ZookeeperEnabledSelector.java
index 68d96bfa8..712055ef8 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/ZookeeperEnabledSelector.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/ZookeeperEnabledSelector.java
@@ -45,7 +45,7 @@ public class ZookeeperEnabledSelector implements EventSelector {
GroupResourceProcessForm groupResourceForm = (GroupResourceProcessForm) processForm;
InlongGroupInfo groupInfo = groupResourceForm.getGroupInfo();
boolean enable =
- groupInfo.getZookeeperEnabled() == 1 && MQType.forType(groupInfo.getMiddlewareType()) != MQType.NONE;
+ groupInfo.getEnableZookeeper() == 1 && MQType.forType(groupInfo.getMqType()) != MQType.NONE;
log.info("zookeeper enabled was [{}] for groupId [{}]", enable, groupId);
return enable;
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/DataFlowUtils.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/DataFlowUtils.java
index 61ec4617a..3643e4139 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/DataFlowUtils.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/DataFlowUtils.java
@@ -93,7 +93,7 @@ public class DataFlowUtils {
// Get source info
String masterAddress = commonOperateService.getSpecifiedParam(InlongGroupSettings.TUBE_MASTER_URL);
- PulsarClusterInfo pulsarCluster = commonOperateService.getPulsarClusterInfo(groupInfo.getMiddlewareType());
+ PulsarClusterInfo pulsarCluster = commonOperateService.getPulsarClusterInfo(groupInfo.getMqType());
SourceInfo sourceInfo = SourceInfoUtils.createSourceInfo(pulsarCluster, masterAddress, clusterBean,
groupInfo, streamInfo, sourceResponse, sourceFields);
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/NodeRelationShipUtils.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/NodeRelationShipUtils.java
index c3e0c8734..7f93fa3c1 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/NodeRelationShipUtils.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/NodeRelationShipUtils.java
@@ -62,20 +62,18 @@ import java.util.stream.Collectors;
public class NodeRelationShipUtils {
public static List<NodeRelationShip> createNodeRelationShipsForStream(InlongStreamInfo streamInfo) {
- String tempView = streamInfo.getTempView();
+ String tempView = streamInfo.getExtParams();
if (StringUtils.isEmpty(tempView)) {
log.warn("StreamNodeRelationShip is empty for Stream={}", streamInfo);
return Lists.newArrayList();
}
- StreamPipeline pipeline = StreamParseUtils.parseStreamPipeline(streamInfo.getTempView(),
+ StreamPipeline pipeline = StreamParseUtils.parseStreamPipeline(streamInfo.getExtParams(),
streamInfo.getInlongStreamId());
- List<NodeRelationShip> nodeRelationShips = pipeline.getPipeline().stream()
+ return pipeline.getPipeline().stream()
.map(streamNodeRelationShip -> new NodeRelationShip(
Lists.newArrayList(streamNodeRelationShip.getInputNodes()),
Lists.newArrayList(streamNodeRelationShip.getOutputNodes())))
- .collect(
- Collectors.toList());
- return nodeRelationShips;
+ .collect(Collectors.toList());
}
public static void optimizeNodeRelationShips(StreamInfo streamInfo, List<TransformResponse> transformResponses) {
@@ -83,19 +81,18 @@ public class NodeRelationShipUtils {
return;
}
Map<String, TransformDefinition> transformTypeMap = transformResponses.stream().collect(
- Collectors.toMap(transformResponse -> transformResponse.getTransformName(),
- transformResponse -> {
- TransformType transformType = TransformType.forType(transformResponse.getTransformType());
- return StreamParseUtils.parseTransformDefinition(transformResponse.getTransformDefinition(),
- transformType);
- }));
+ Collectors.toMap(TransformResponse::getTransformName, transformResponse -> {
+ TransformType transformType = TransformType.forType(transformResponse.getTransformType());
+ return StreamParseUtils.parseTransformDefinition(transformResponse.getTransformDefinition(),
+ transformType);
+ }));
List<Node> nodes = streamInfo.getNodes();
Map<String, TransformNode> joinNodes = nodes.stream().filter(node -> node instanceof TransformNode)
.map(node -> (TransformNode) node)
.filter(transformNode -> {
TransformDefinition transformDefinition = transformTypeMap.get(transformNode.getName());
return transformDefinition.getTransformType() == TransformType.JOINER;
- }).collect(Collectors.toMap(transformNode -> transformNode.getName(), transformNode -> transformNode));
+ }).collect(Collectors.toMap(TransformNode::getName, transformNode -> transformNode));
List<NodeRelationShip> relationShips = streamInfo.getRelations();
Iterator<NodeRelationShip> shipIterator = relationShips.listIterator();
List<NodeRelationShip> joinRelationShips = Lists.newArrayList();
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/SourceInfoUtils.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/SourceInfoUtils.java
index 8e5f4b6f0..08c8d4d52 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/SourceInfoUtils.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/SourceInfoUtils.java
@@ -64,7 +64,7 @@ public class SourceInfoUtils {
ClusterBean clusterBean, InlongGroupInfo groupInfo, InlongStreamInfo streamInfo,
SourceResponse sourceResponse, List<FieldInfo> sourceFields) {
- MQType mqType = MQType.forType(groupInfo.getMiddlewareType());
+ MQType mqType = MQType.forType(groupInfo.getMqType());
DeserializationInfo deserializationInfo = SerializationUtils.createDeserialInfo(sourceResponse, streamInfo);
SourceInfo sourceInfo;
if (mqType == MQType.PULSAR || mqType == MQType.TDMQ_PULSAR) {
@@ -86,14 +86,14 @@ public class SourceInfoUtils {
private static SourceInfo createPulsarSourceInfo(PulsarClusterInfo pulsarCluster, ClusterBean clusterBean,
InlongGroupInfo groupInfo, InlongStreamInfo streamInfo,
DeserializationInfo deserializationInfo, List<FieldInfo> fieldInfos) {
- String topicName = streamInfo.getMqResourceObj();
+ String topicName = streamInfo.getMqResource();
InlongGroupPulsarInfo pulsarInfo = (InlongGroupPulsarInfo) groupInfo.getMqExtInfo();
String tenant = clusterBean.getDefaultTenant();
if (StringUtils.isNotEmpty(pulsarInfo.getTenant())) {
tenant = pulsarInfo.getTenant();
}
- final String namespace = groupInfo.getMqResourceObj();
+ final String namespace = groupInfo.getMqResource();
// Full name of topic in Pulsar
final String fullTopicName = "persistent://" + tenant + "/" + namespace + "/" + topicName;
final String consumerGroup = clusterBean.getAppName() + "_" + topicName + "_consumer_group";
@@ -115,7 +115,7 @@ public class SourceInfoUtils {
private static TubeSourceInfo createTubeSourceInfo(InlongGroupInfo groupInfo, String masterAddress,
ClusterBean clusterBean, DeserializationInfo deserializationInfo, List<FieldInfo> fieldInfos) {
Preconditions.checkNotNull(masterAddress, "tube cluster address cannot be empty");
- String topic = groupInfo.getMqResourceObj();
+ String topic = groupInfo.getMqResource();
String consumerGroup = clusterBean.getAppName() + "_" + topic + "_consumer_group";
return new TubeSourceInfo(topic, masterAddress, consumerGroup, deserializationInfo,
fieldInfos.toArray(new FieldInfo[0]));
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/listener/AbstractSourceOperateListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/listener/AbstractSourceOperateListener.java
index 43ab41992..f03035584 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/listener/AbstractSourceOperateListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/listener/AbstractSourceOperateListener.java
@@ -79,8 +79,7 @@ public abstract class AbstractSourceOperateListener implements DataSourceOperate
if (CollectionUtils.isNotEmpty(unOperatedSources)) {
GroupOperateType groupOperateType = getOperateType(context.getProcessForm());
StringBuilder builder = new StringBuilder("Unsupported operate ").append(groupOperateType).append(" for (");
- unOperatedSources.stream()
- .forEach(source -> builder.append(" ").append(source.getSourceName()).append(" "));
+ unOperatedSources.forEach(source -> builder.append(" ").append(source.getSourceName()).append(" "));
String errMsg = builder.append(")").toString();
throw new WorkflowListenerException(errMsg);
} else {
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/consumption/listener/ConsumptionCompleteProcessListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/consumption/listener/ConsumptionCompleteProcessListener.java
index a2e05e487..4720404ab 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/consumption/listener/ConsumptionCompleteProcessListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/consumption/listener/ConsumptionCompleteProcessListener.java
@@ -117,14 +117,14 @@ public class ConsumptionCompleteProcessListener implements ProcessEventListener
String groupId = entity.getInlongGroupId();
InlongGroupInfo groupInfo = groupService.get(groupId);
Preconditions.checkNotNull(groupInfo, "inlong group not found for groupId=" + groupId);
- String mqResourceObj = groupInfo.getMqResourceObj();
- Preconditions.checkNotNull(mqResourceObj, "mq resource cannot empty for groupId=" + groupId);
+ String mqResource = groupInfo.getMqResource();
+ Preconditions.checkNotNull(mqResource, "mq resource cannot empty for groupId=" + groupId);
PulsarClusterInfo globalCluster = commonOperateService.getPulsarClusterInfo(entity.getMiddlewareType());
try (PulsarAdmin pulsarAdmin = PulsarUtils.getPulsarAdmin(globalCluster)) {
PulsarTopicBean topicMessage = new PulsarTopicBean();
String tenant = clusterBean.getDefaultTenant();
topicMessage.setTenant(tenant);
- topicMessage.setNamespace(mqResourceObj);
+ topicMessage.setNamespace(mqResource);
// If cross-regional replication is started, each cluster needs to create consumer groups in cycles
String consumerGroup = entity.getConsumerGroupId();
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/stream/CreateStreamWorkflowDefinition.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/stream/CreateStreamWorkflowDefinition.java
index 5cc50f53d..a8d12fd4c 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/stream/CreateStreamWorkflowDefinition.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/stream/CreateStreamWorkflowDefinition.java
@@ -22,10 +22,10 @@ import org.apache.commons.collections.CollectionUtils;
import org.apache.inlong.manager.common.enums.MQType;
import org.apache.inlong.manager.common.enums.SinkType;
import org.apache.inlong.manager.common.pojo.workflow.form.GroupResourceProcessForm;
-import org.apache.inlong.manager.service.sink.StreamSinkService;
-import org.apache.inlong.manager.service.resource.CreateSinkResourceForStreamListener;
import org.apache.inlong.manager.service.mq.CreatePulsarGroupForStreamTaskListener;
import org.apache.inlong.manager.service.mq.CreatePulsarTopicForStreamTaskListener;
+import org.apache.inlong.manager.service.resource.CreateSinkResourceForStreamListener;
+import org.apache.inlong.manager.service.sink.StreamSinkService;
import org.apache.inlong.manager.service.sort.PushSortConfigListener;
import org.apache.inlong.manager.service.workflow.ProcessName;
import org.apache.inlong.manager.service.workflow.WorkflowDefinition;
@@ -90,11 +90,11 @@ public class CreateStreamWorkflowDefinition implements WorkflowDefinition {
ServiceTask createPulsarTopicTask = new ServiceTask();
createPulsarTopicTask.setSkipResolver(c -> {
GroupResourceProcessForm form = (GroupResourceProcessForm) c.getProcessForm();
- MQType mqType = MQType.forType(form.getGroupInfo().getMiddlewareType());
+ MQType mqType = MQType.forType(form.getGroupInfo().getMqType());
if (mqType == MQType.PULSAR || mqType == MQType.TDMQ_PULSAR) {
return false;
}
- log.warn("no need to create pulsar topic for groupId={}, streamId={}, as the middlewareType={}",
+ log.warn("no need to create pulsar topic for groupId={}, streamId={}, as the mqType={}",
form.getInlongGroupId(), form.getInlongStreamId(), mqType);
return true;
});
@@ -106,11 +106,11 @@ public class CreateStreamWorkflowDefinition implements WorkflowDefinition {
ServiceTask createPulsarSubscriptionGroupTask = new ServiceTask();
createPulsarSubscriptionGroupTask.setSkipResolver(c -> {
GroupResourceProcessForm form = (GroupResourceProcessForm) c.getProcessForm();
- MQType mqType = MQType.forType(form.getGroupInfo().getMiddlewareType());
+ MQType mqType = MQType.forType(form.getGroupInfo().getMqType());
if (mqType == MQType.PULSAR || mqType == MQType.TDMQ_PULSAR) {
return false;
}
- log.warn("no need to create pulsar subscription for groupId={}, streamId={}, as the middlewareType={}",
+ log.warn("no need to create pulsar subscription for groupId={}, streamId={}, as the mqType={}",
form.getInlongGroupId(), form.getInlongStreamId(), mqType);
return true;
});
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/InlongGroupProcessOperationTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/InlongGroupProcessOperationTest.java
index 45c947f81..7497d2e0f 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/InlongGroupProcessOperationTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/InlongGroupProcessOperationTest.java
@@ -65,7 +65,7 @@ public class InlongGroupProcessOperationTest extends ServiceBaseTest {
groupInfo.setInlongGroupId(GROUP_ID);
groupInfo.setName(GROUP_NAME);
groupInfo.setInCharges(OPERATOR);
- groupInfo.setMiddlewareType(MQType.PULSAR.getType());
+ groupInfo.setMqType(MQType.PULSAR.getType());
InlongGroupPulsarInfo pulsarInfo = new InlongGroupPulsarInfo();
pulsarInfo.setInlongGroupId(GROUP_ID);
groupInfo.setMqExtInfo(pulsarInfo);
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/InlongGroupServiceTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/InlongGroupServiceTest.java
index cfe2ede2d..bd115f5aa 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/InlongGroupServiceTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/InlongGroupServiceTest.java
@@ -68,13 +68,13 @@ public class InlongGroupServiceTest {
groupInfo = new InlongGroupInfo();
groupInfo.setName(groupName);
- groupInfo.setMiddlewareType(MQType.PULSAR.getType());
+ groupInfo.setMqType(MQType.PULSAR.getType());
groupInfo.setCreator(operator);
groupInfo.setInCharges(operator);
groupInfo.setStatus(GroupStatus.CONFIG_SUCCESSFUL.getCode());
InlongGroupPulsarInfo pulsarInfo = new InlongGroupPulsarInfo();
- pulsarInfo.setMiddlewareType(MQType.PULSAR.getType());
+ pulsarInfo.setMqType(MQType.PULSAR.getType());
pulsarInfo.setEnsemble(3);
pulsarInfo.setWriteQuorum(3);
pulsarInfo.setAckQuorum(2);
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/sort/DisableZkForSortTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/sort/DisableZkForSortTest.java
index 243b02fd0..cd1a27a52 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/sort/DisableZkForSortTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/sort/DisableZkForSortTest.java
@@ -124,7 +124,7 @@ public class DisableZkForSortTest extends WorkflowServiceImplTest {
public void testCreateSortConfigInCreateWorkflow() {
InlongGroupInfo groupInfo = initGroupForm("PULSAR", "test21");
groupInfo.setStatus(GroupStatus.CONFIG_SUCCESSFUL.getCode());
- groupInfo.setZookeeperEnabled(0);
+ groupInfo.setEnableZookeeper(0);
groupService.update(groupInfo.genRequest(), OPERATOR);
InlongStreamInfo streamInfo = createStreamInfo(groupInfo);
createHiveSink(streamInfo);
@@ -149,7 +149,7 @@ public class DisableZkForSortTest extends WorkflowServiceImplTest {
// @Test
public void testCreateSortConfigInUpdateWorkflow() {
InlongGroupInfo groupInfo = initGroupForm("PULSAR", "test20");
- groupInfo.setZookeeperEnabled(0);
+ groupInfo.setEnableZookeeper(0);
groupService.updateStatus(GROUP_ID, GroupStatus.CONFIG_SUCCESSFUL.getCode(), OPERATOR);
groupService.update(groupInfo.genRequest(), OPERATOR);
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/workflow/ServiceTaskListenerFactoryTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/workflow/ServiceTaskListenerFactoryTest.java
index 4239f67f1..800ddf97e 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/workflow/ServiceTaskListenerFactoryTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/workflow/ServiceTaskListenerFactoryTest.java
@@ -47,7 +47,7 @@ public class ServiceTaskListenerFactoryTest extends ServiceBaseTest {
GroupResourceProcessForm processForm = new GroupResourceProcessForm();
InlongGroupInfo groupInfo = new InlongGroupInfo();
//check pulsar listener
- groupInfo.setMiddlewareType(MQType.PULSAR.getType());
+ groupInfo.setMqType(MQType.PULSAR.getType());
groupInfo.setMqExtInfo(new InlongGroupPulsarInfo());
processForm.setGroupInfo(groupInfo);
WorkflowContext context = new WorkflowContext();
@@ -61,7 +61,7 @@ public class ServiceTaskListenerFactoryTest extends ServiceBaseTest {
Assert.assertTrue(queueOperateListeners.get(1) instanceof CreatePulsarGroupTaskListener);
// check tube listener
- groupInfo.setMiddlewareType(MQType.TUBE.getType());
+ groupInfo.setMqType(MQType.TUBE.getType());
queueOperateListeners = serviceTaskListenerFactory.getQueueOperateListener(context);
Assert.assertEquals(2, queueOperateListeners.size());
Assert.assertTrue(queueOperateListeners.get(0) instanceof CreateTubeTopicTaskListener);
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/workflow/WorkflowServiceImplTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/workflow/WorkflowServiceImplTest.java
index 2ecdc217f..cff8f12b6 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/workflow/WorkflowServiceImplTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/workflow/WorkflowServiceImplTest.java
@@ -39,9 +39,7 @@ import org.apache.inlong.manager.dao.mapper.WorkflowTaskEntityMapper;
import org.apache.inlong.manager.service.ServiceBaseTest;
import org.apache.inlong.manager.service.core.InlongGroupService;
import org.apache.inlong.manager.service.core.InlongStreamService;
-import org.apache.inlong.manager.service.mocks.MockDeleteSortListener;
import org.apache.inlong.manager.service.mocks.MockPlugin;
-import org.apache.inlong.manager.service.mocks.MockRestartSortListener;
import org.apache.inlong.manager.service.mocks.MockStopSortListener;
import org.apache.inlong.manager.service.mq.CreatePulsarGroupTaskListener;
import org.apache.inlong.manager.service.mq.CreatePulsarResourceTaskListener;
@@ -116,14 +114,14 @@ public class WorkflowServiceImplTest extends ServiceBaseTest {
/**
* Init inlong group form
*/
- public InlongGroupInfo initGroupForm(String middlewareType) {
- return initGroupForm(middlewareType, "test" + subType);
+ public InlongGroupInfo initGroupForm(String mqType) {
+ return initGroupForm(mqType, "test" + subType);
}
/**
* Init inlong group form
*/
- public InlongGroupInfo initGroupForm(String middlewareType, String inLongGroupName) {
+ public InlongGroupInfo initGroupForm(String mqType, String inLongGroupName) {
String inLongGroupId = "b_" + inLongGroupName;
processName = ProcessName.CREATE_GROUP_RESOURCE;
applicant = OPERATOR;
@@ -139,17 +137,15 @@ public class WorkflowServiceImplTest extends ServiceBaseTest {
groupInfo.setName(inLongGroupName);
groupInfo.setInCharges(OPERATOR);
groupInfo.setInlongGroupId(inLongGroupId);
- groupInfo.setMiddlewareType(middlewareType);
+ groupInfo.setMqType(mqType);
groupInfo.setMqExtInfo(new InlongGroupPulsarInfo());
- groupInfo.setMqResourceObj("test-queue");
+ groupInfo.setMqResource("test-queue");
groupService.save(groupInfo.genRequest(), OPERATOR);
groupService.updateStatus(inLongGroupId, GroupStatus.TO_BE_APPROVAL.getCode(), OPERATOR);
- // groupService.update(groupInfo.genRequest(), OPERATOR);
groupService.updateStatus(inLongGroupId, GroupStatus.APPROVE_PASSED.getCode(), OPERATOR);
groupService.update(groupInfo.genRequest(), OPERATOR);
groupService.updateStatus(inLongGroupId, GroupStatus.CONFIG_ING.getCode(), OPERATOR);
- // groupService.update(groupInfo.genRequest(), OPERATOR);
form = new GroupResourceProcessForm();
form.setGroupInfo(groupInfo);
@@ -177,10 +173,9 @@ public class WorkflowServiceImplTest extends ServiceBaseTest {
InlongStreamRequest request = new InlongStreamRequest();
request.setInlongGroupId(groupInfo.getInlongGroupId());
request.setInlongStreamId(STREAM_ID);
- request.setMqResourceObj(STREAM_ID);
+ request.setMqResource(STREAM_ID);
request.setDataSeparator("124");
request.setDataEncoding(DATA_ENCODING);
- request.setInCharges(OPERATOR);
request.setFieldList(createStreamFields(groupInfo.getInlongGroupId(), STREAM_ID));
streamService.save(request, OPERATOR);
@@ -304,10 +299,13 @@ public class WorkflowServiceImplTest extends ServiceBaseTest {
Assert.assertTrue(stopSortTask instanceof ServiceTask);
List<TaskEventListener> listeners = Lists.newArrayList(stopSortTask.getNameToListenerMap().values());
Assert.assertTrue(listeners.get(0) instanceof MockStopSortListener);
+ Assert.assertEquals(2, listeners.size());
WorkflowTask stopSourceTask = process.getTaskByName("stopSource");
Assert.assertTrue(stopSourceTask instanceof ServiceTask);
listeners = Lists.newArrayList(stopSourceTask.getNameToListenerMap().values());
+ Assert.assertTrue(listeners.get(0) instanceof MockStopSortListener);
+ Assert.assertEquals(2, listeners.size());
}
@Test
@@ -334,12 +332,13 @@ public class WorkflowServiceImplTest extends ServiceBaseTest {
WorkflowProcess process = context.getProcess();
WorkflowTask restartSort = process.getTaskByName("restartSort");
Assert.assertTrue(restartSort instanceof ServiceTask);
+ // MockRestartSortListener + CreateSortConfigListener
List<TaskEventListener> listeners = Lists.newArrayList(restartSort.getNameToListenerMap().values());
- Assert.assertEquals(1, listeners.size());
- Assert.assertTrue(listeners.get(0) instanceof MockRestartSortListener);
+ Assert.assertEquals(2, listeners.size());
WorkflowTask restartSourceTask = process.getTaskByName("restartSource");
Assert.assertTrue(restartSourceTask instanceof ServiceTask);
+ // MockRestartSourceListener + SourceRestartListener
listeners = Lists.newArrayList(restartSourceTask.getNameToListenerMap().values());
Assert.assertEquals(2, listeners.size());
}
@@ -365,16 +364,18 @@ public class WorkflowServiceImplTest extends ServiceBaseTest {
WorkflowResult result = WorkflowBeanUtils.result(context);
ProcessResponse view = result.getProcessInfo();
Assert.assertSame(view.getStatus(), ProcessStatus.COMPLETED);
+
WorkflowProcess process = context.getProcess();
WorkflowTask deleteSort = process.getTaskByName("deleteSort");
Assert.assertTrue(deleteSort instanceof ServiceTask);
+ // CreateSortConfigListener + MockDeleteSortListener
List<TaskEventListener> listeners = Lists.newArrayList(deleteSort.getNameToListenerMap().values());
- Assert.assertEquals(1, listeners.size());
- Assert.assertTrue(listeners.get(0) instanceof MockDeleteSortListener);
+ Assert.assertEquals(2, listeners.size());
WorkflowTask deleteSourceTask = process.getTaskByName("deleteSource");
Assert.assertTrue(deleteSourceTask instanceof ServiceTask);
listeners = Lists.newArrayList(deleteSourceTask.getNameToListenerMap().values());
+ // SourceDeleteListener + MockDeleteSourceListener
Assert.assertEquals(2, listeners.size());
}
diff --git a/inlong-manager/manager-test/src/main/resources/sql/apache_inlong_manager.sql b/inlong-manager/manager-test/src/main/resources/sql/apache_inlong_manager.sql
index 6dce13aec..ed3847efb 100644
--- a/inlong-manager/manager-test/src/main/resources/sql/apache_inlong_manager.sql
+++ b/inlong-manager/manager-test/src/main/resources/sql/apache_inlong_manager.sql
@@ -21,34 +21,32 @@ SET FOREIGN_KEY_CHECKS = 0;
-- ----------------------------
-- Table structure for inlong_group
-- ----------------------------
-DROP TABLE IF EXISTS `inlong_group`;
-CREATE TABLE `inlong_group`
+CREATE TABLE IF NOT EXISTS `inlong_group`
(
- `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
- `inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id, filled in by the user, undeleted ones cannot be repeated',
- `name` varchar(128) DEFAULT '' COMMENT 'Inlong group name, English, numbers and underscore',
- `cn_name` varchar(256) DEFAULT NULL COMMENT 'Chinese display name',
- `description` varchar(256) DEFAULT '' COMMENT 'Inlong group Introduction',
- `middleware_type` varchar(20) DEFAULT 'TUBE' COMMENT 'The middleware type of message queue, high throughput: TUBE, high consistency: PULSAR',
- `queue_module` VARCHAR(20) NULL DEFAULT 'parallel' COMMENT 'Queue model of Pulsar, parallel: multiple partitions, high throughput, out-of-order messages; serial: single partition, low throughput, and orderly messages',
- `topic_partition_num` INT(4) NULL DEFAULT '3' COMMENT 'The number of partitions of Pulsar Topic, 1-20',
- `mq_resource_obj` varchar(128) NOT NULL COMMENT 'MQ resource object, for Tube, its Topic, for Pulsar, its Namespace',
- `daily_records` int(11) DEFAULT '10' COMMENT 'Number of access records per day, unit: 10,000 records per day',
- `daily_storage` int(11) DEFAULT '10' COMMENT 'Access size by day, unit: GB per day',
- `peak_records` int(11) DEFAULT '1000' COMMENT 'Access peak per second, unit: records per second',
- `max_length` int(11) DEFAULT '10240' COMMENT 'The maximum length of a single piece of data, unit: Byte',
- `schema_name` varchar(128) DEFAULT NULL COMMENT 'Data type, associated data_schema table',
- `in_charges` varchar(512) DEFAULT NULL COMMENT 'Name of responsible person, separated by commas',
- `followers` varchar(512) DEFAULT NULL COMMENT 'Name of followers, separated by commas',
- `status` int(4) DEFAULT '21' COMMENT 'Inlong group status',
- `is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete, 0: not deleted, > 0: deleted',
- `creator` varchar(64) NOT NULL COMMENT 'Creator name',
- `modifier` varchar(64) DEFAULT NULL COMMENT 'Modifier name',
- `create_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Create time',
- `modify_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 'Modify time',
- `temp_view` text DEFAULT NULL COMMENT 'Temporary view, used to save intermediate data that has not been submitted or approved after modification',
- `zookeeper_enabled` int(4) DEFAULT '1' COMMENT 'Need zookeeper support, 0 false 1 true',
- `proxy_cluster_id` int(11) DEFAULT NULL COMMENT 'The id of dataproxy cluster',
+ `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
+ `inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id, filled in by the user, undeleted ones cannot be repeated',
+ `name` varchar(128) DEFAULT '' COMMENT 'Inlong group name, English, Chinese, numbers, etc',
+ `description` varchar(256) DEFAULT '' COMMENT 'Inlong group Introduction',
+ `mq_type` varchar(20) DEFAULT 'TUBE' COMMENT 'The message queue type, high throughput: TUBE, high consistency: PULSAR',
+ `mq_resource` varchar(128) NOT NULL COMMENT 'MQ resource, for Tube, its Topic, for Pulsar, its Namespace',
+ `daily_records` int(11) DEFAULT '10' COMMENT 'Number of access records per day, unit: 10,000 records per day',
+ `daily_storage` int(11) DEFAULT '10' COMMENT 'Access size by day, unit: GB per day',
+ `peak_records` int(11) DEFAULT '1000' COMMENT 'Access peak per second, unit: records per second',
+ `max_length` int(11) DEFAULT '10240' COMMENT 'The maximum length of a single piece of data, unit: Byte',
+ `in_charges` varchar(512) NOT NULL COMMENT 'Name of responsible person, separated by commas',
+ `followers` varchar(512) DEFAULT NULL COMMENT 'Name of followers, separated by commas',
+ `enable_zookeeper` tinyint(2) DEFAULT '1' COMMENT 'Need zookeeper support, 0: false, 1: true',
+ `enable_create_resource` tinyint(2) DEFAULT '1' COMMENT 'Whether to enable create mq resource? 0: disable, 1: enable. Default is 1',
+ `lightweight` tinyint(2) DEFAULT '1' COMMENT 'Whether to use lightweight mode, 0: false, 1: true. Default is 1',
+ `inlong_cluster_tag` varchar(128) DEFAULT NULL COMMENT 'The cluster tag, which links to inlong_cluster table',
+ `ext_params` text DEFAULT NULL COMMENT 'Extended params, will be saved as JSON string, such as queue_module, partition_num,',
+ `status` int(4) DEFAULT '100' COMMENT 'Inlong group status',
+ `previous_status` int(4) DEFAULT '100' COMMENT 'Previous group status',
+ `is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete, 0: not deleted, > 0: deleted',
+ `creator` varchar(64) NOT NULL COMMENT 'Creator name',
+ `modifier` varchar(64) DEFAULT NULL COMMENT 'Modifier name',
+ `create_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Create time',
+ `modify_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 'Modify time',
PRIMARY KEY (`id`),
UNIQUE KEY `unique_inlong_group` (`inlong_group_id`, `is_deleted`)
);
@@ -56,8 +54,7 @@ CREATE TABLE `inlong_group`
-- ----------------------------
-- Table structure for inlong_group_pulsar
-- ----------------------------
-DROP TABLE IF EXISTS `inlong_group_pulsar`;
-CREATE TABLE `inlong_group_pulsar`
+CREATE TABLE IF NOT EXISTS `inlong_group_pulsar`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id, filled in by the user, undeleted ones cannot be repeated',
@@ -81,8 +78,7 @@ CREATE TABLE `inlong_group_pulsar`
-- ----------------------------
-- Table structure for inlong_group_ext
-- ----------------------------
-DROP TABLE IF EXISTS `inlong_group_ext`;
-CREATE TABLE `inlong_group_ext`
+CREATE TABLE IF NOT EXISTS `inlong_group_ext`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
@@ -130,7 +126,7 @@ CREATE TABLE IF NOT EXISTS `inlong_cluster_node`
`type` varchar(20) DEFAULT '' COMMENT 'Cluster type, such as: DATA_PROXY, AGENT, etc',
`ip` varchar(512) NULL COMMENT 'Cluster IP, separated by commas, such as: 127.0.0.1:8080,host2:8081',
`port` int(6) NULL COMMENT 'Cluster port',
- `ext_params` text DEFAULT NULL COMMENT 'Another fields will saved as JSON string',
+ `ext_params` text DEFAULT NULL COMMENT 'Another fields will be saved as JSON string',
`status` int(4) DEFAULT '0' COMMENT 'Cluster status',
`is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete, 0: not deleted, > 0: deleted',
`creator` varchar(64) NOT NULL COMMENT 'Creator name',
@@ -167,8 +163,7 @@ CREATE TABLE IF NOT EXISTS `data_node`
-- ----------------------------
-- Table structure for third_party_cluster
-- ----------------------------
-DROP TABLE IF EXISTS `third_party_cluster`;
-CREATE TABLE `third_party_cluster`
+CREATE TABLE IF NOT EXISTS `third_party_cluster`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`name` varchar(128) NOT NULL COMMENT 'Cluster name',
@@ -194,8 +189,7 @@ CREATE TABLE `third_party_cluster`
-- ----------------------------
-- Table structure for common_db_server
-- ----------------------------
-DROP TABLE IF EXISTS `common_db_server`;
-CREATE TABLE `common_db_server`
+CREATE TABLE IF NOT EXISTS `common_db_server`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`access_type` varchar(20) NOT NULL COMMENT 'Collection type, with Agent, DataProxy client, LoadProxy',
@@ -229,8 +223,7 @@ CREATE TABLE `common_db_server`
-- ----------------------------
-- Table structure for common_file_server
-- ----------------------------
-DROP TABLE IF EXISTS `common_file_server`;
-CREATE TABLE `common_file_server`
+CREATE TABLE IF NOT EXISTS `common_file_server`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`access_type` varchar(20) NOT NULL COMMENT 'Collection type, with Agent, DataProxy, LoadProxy',
@@ -254,8 +247,7 @@ CREATE TABLE `common_file_server`
-- ----------------------------
-- Table structure for consumption
-- ----------------------------
-DROP TABLE IF EXISTS `consumption`;
-CREATE TABLE `consumption`
+CREATE TABLE IF NOT EXISTS `consumption`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`consumer_group_name` varchar(256) DEFAULT NULL COMMENT 'consumer group name',
@@ -278,8 +270,7 @@ CREATE TABLE `consumption`
-- ----------------------------
-- Table structure for consumption_pulsar
-- ----------------------------
-DROP TABLE IF EXISTS `consumption_pulsar`;
-CREATE TABLE `consumption_pulsar`
+CREATE TABLE IF NOT EXISTS `consumption_pulsar`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`consumption_id` int(11) DEFAULT NULL COMMENT 'ID of the consumption information to which it belongs, guaranteed to be uniquely associated with consumption information',
@@ -297,8 +288,7 @@ CREATE TABLE `consumption_pulsar`
-- ----------------------------
-- Table structure for data_proxy_cluster
-- ----------------------------
-DROP TABLE IF EXISTS `data_proxy_cluster`;
-CREATE TABLE `data_proxy_cluster`
+CREATE TABLE IF NOT EXISTS `data_proxy_cluster`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`name` varchar(128) NOT NULL COMMENT 'Cluster name',
@@ -322,8 +312,7 @@ CREATE TABLE `data_proxy_cluster`
-- ----------------------------
-- Table structure for data_schema
-- ----------------------------
-DROP TABLE IF EXISTS `data_schema`;
-CREATE TABLE `data_schema`
+CREATE TABLE IF NOT EXISTS `data_schema`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`name` varchar(128) NOT NULL COMMENT 'Data format name, globally unique',
@@ -335,15 +324,10 @@ CREATE TABLE `data_schema`
UNIQUE KEY `unique_schema_name` (`name`)
);
--- create default data schema
-INSERT INTO `data_schema` (name, agent_type, data_generate_rule, sort_type, time_offset)
-values ('m0_day', 'file_agent', 'day', 0, '-0d');
-
-- ----------------------------
-- Table structure for stream_source_cmd_config
-- ----------------------------
-DROP TABLE IF EXISTS `stream_source_cmd_config`;
-CREATE TABLE `stream_source_cmd_config`
+CREATE TABLE IF NOT EXISTS `stream_source_cmd_config`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'cmd id',
`cmd_type` int(11) NOT NULL,
@@ -359,36 +343,32 @@ CREATE TABLE `stream_source_cmd_config`
-- ----------------------------
-- Table structure for inlong_stream
-- ----------------------------
-DROP TABLE IF EXISTS `inlong_stream`;
-CREATE TABLE `inlong_stream`
+CREATE TABLE IF NOT EXISTS `inlong_stream`
(
- `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
- `inlong_stream_id` varchar(256) NOT NULL COMMENT 'Inlong stream id, non-deleted globally unique',
- `inlong_group_id` varchar(256) NOT NULL COMMENT 'Owning inlong group id',
- `name` varchar(64) DEFAULT NULL COMMENT 'The name of the inlong stream page display, can be Chinese',
- `description` varchar(256) DEFAULT '' COMMENT 'Introduction to inlong stream',
- `mq_resource_obj` varchar(128) DEFAULT NULL COMMENT 'MQ resource object, in the inlong stream, Tube is inlong_stream_id, Pulsar is Topic',
- `data_source_type` varchar(32) DEFAULT 'FILE' COMMENT 'Data source type, including: FILE, BINLOG, AUTO_PUSH, etc.',
- `storage_period` int(11) DEFAULT '1' COMMENT 'The storage period of data in MQ, unit: day',
- `data_type` varchar(20) DEFAULT NULL COMMENT 'Data type, including: CSV, KEY-VALUE, JSON, AVRO, etc.',
- `data_encoding` varchar(8) DEFAULT 'UTF-8' COMMENT 'Data encoding format, including: UTF-8, GBK',
- `data_separator` varchar(8) DEFAULT NULL COMMENT 'The source data field separator, stored as ASCII code',
- `data_escape_char` varchar(8) DEFAULT NULL COMMENT 'Source data field escape character, the default is NULL (NULL), stored as 1 character',
- `have_predefined_fields` tinyint(1) DEFAULT '0' COMMENT '(File, DB access) whether there are predefined fields, 0: none, 1: yes (save to inlong_stream_field)',
- `sync_send` tinyint(1) DEFAULT '0' COMMENT 'order_preserving 0: none, 1: yes',
- `daily_records` int(11) DEFAULT '10' COMMENT 'Number of access records per day, unit: 10,000 records per day',
- `daily_storage` int(11) DEFAULT '10' COMMENT 'Access size by day, unit: GB per day',
- `peak_records` int(11) DEFAULT '1000' COMMENT 'Access peak per second, unit: records per second',
- `max_length` int(11) DEFAULT '10240' COMMENT 'The maximum length of a single piece of data, unit: Byte',
- `in_charges` varchar(512) DEFAULT NULL COMMENT 'Name of responsible person, separated by commas',
- `status` int(4) DEFAULT '0' COMMENT 'Inlong stream status',
- `previous_status` int(4) DEFAULT '0' COMMENT 'Previous status',
- `is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete, 0: not deleted, > 0: deleted',
- `creator` varchar(64) DEFAULT NULL COMMENT 'Creator name',
- `modifier` varchar(64) DEFAULT NULL COMMENT 'Modifier name',
- `create_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Create time',
- `modify_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 'Modify time',
- `temp_view` text DEFAULT NULL COMMENT 'Temporary view, used to save intermediate data that has not been submitted or approved after modification',
+ `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
+ `inlong_group_id` varchar(256) NOT NULL COMMENT 'Owning inlong group id',
+ `inlong_stream_id` varchar(256) NOT NULL COMMENT 'Inlong stream id, non-deleted globally unique',
+ `name` varchar(64) DEFAULT NULL COMMENT 'The name of the inlong stream page display, can be Chinese',
+ `description` varchar(256) DEFAULT '' COMMENT 'Introduction to inlong stream',
+ `mq_resource` varchar(128) DEFAULT NULL COMMENT 'MQ resource, in one stream, corresponding to the filter ID of Tube, corresponding to the topic of Pulsar',
+ `data_type` varchar(20) DEFAULT NULL COMMENT 'Data type, including: CSV, KEY-VALUE, JSON, AVRO, etc.',
+ `data_encoding` varchar(8) DEFAULT 'UTF-8' COMMENT 'Data encoding format, including: UTF-8, GBK, etc.',
+ `data_separator` varchar(8) DEFAULT NULL COMMENT 'The source data field separator, stored as ASCII code',
+ `data_escape_char` varchar(8) DEFAULT NULL COMMENT 'Source data field escape character, the default is NULL (NULL), stored as 1 character',
+ `sync_send` tinyint(1) DEFAULT '0' COMMENT 'order_preserving 0: none, 1: yes',
+ `daily_records` int(11) DEFAULT '10' COMMENT 'Number of access records per day, unit: 10,000 records per day',
+ `daily_storage` int(11) DEFAULT '10' COMMENT 'Access size by day, unit: GB per day',
+ `peak_records` int(11) DEFAULT '1000' COMMENT 'Access peak per second, unit: records per second',
+ `max_length` int(11) DEFAULT '10240' COMMENT 'The maximum length of a single piece of data, unit: Byte',
+ `storage_period` int(11) DEFAULT '1' COMMENT 'The storage period of data in MQ, unit: day',
+ `ext_params` text DEFAULT NULL COMMENT 'Extended params, will be saved as JSON string',
+ `status` int(4) DEFAULT '100' COMMENT 'Inlong stream status',
+ `previous_status` int(4) DEFAULT '100' COMMENT 'Previous status',
+ `is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete, 0: not deleted, > 0: deleted',
+ `creator` varchar(64) DEFAULT NULL COMMENT 'Creator name',
+ `modifier` varchar(64) DEFAULT NULL COMMENT 'Modifier name',
+ `create_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Create time',
+ `modify_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 'Modify time',
PRIMARY KEY (`id`),
UNIQUE KEY `unique_inlong_stream` (`inlong_stream_id`, `inlong_group_id`, `is_deleted`)
);
@@ -396,8 +376,7 @@ CREATE TABLE `inlong_stream`
-- ----------------------------
-- Table structure for inlong_stream_field
-- ----------------------------
-DROP TABLE IF EXISTS `inlong_stream_field`;
-CREATE TABLE `inlong_stream_field`
+CREATE TABLE IF NOT EXISTS `inlong_stream_field`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Owning inlong group id',
@@ -419,8 +398,7 @@ CREATE TABLE `inlong_stream_field`
-- ----------------------------
-- Table structure for operation_log
-- ----------------------------
-DROP TABLE IF EXISTS `operation_log`;
-CREATE TABLE `operation_log`
+CREATE TABLE IF NOT EXISTS `operation_log`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`authentication_type` varchar(64) DEFAULT NULL COMMENT 'Authentication type',
@@ -443,8 +421,7 @@ CREATE TABLE `operation_log`
-- ----------------------------
-- Table structure for role
-- ----------------------------
-DROP TABLE IF EXISTS `role`;
-CREATE TABLE `role`
+CREATE TABLE IF NOT EXISTS `role`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`role_code` varchar(100) NOT NULL COMMENT 'Role code',
@@ -462,8 +439,7 @@ CREATE TABLE `role`
-- ----------------------------
-- Table structure for source_file_basic
-- ----------------------------
-DROP TABLE IF EXISTS `source_file_basic`;
-CREATE TABLE `source_file_basic`
+CREATE TABLE IF NOT EXISTS `source_file_basic`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'ID',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
@@ -487,8 +463,7 @@ CREATE TABLE `source_file_basic`
-- ----------------------------
-- Table structure for source_file_detail
-- ----------------------------
-DROP TABLE IF EXISTS `source_file_detail`;
-CREATE TABLE `source_file_detail`
+CREATE TABLE IF NOT EXISTS `source_file_detail`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Owning inlong group id',
@@ -516,8 +491,7 @@ CREATE TABLE `source_file_detail`
-- ----------------------------
-- Table structure for stream_source
-- ----------------------------
-DROP TABLE IF EXISTS `stream_source`;
-CREATE TABLE `stream_source`
+CREATE TABLE IF NOT EXISTS `stream_source`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'ID',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
@@ -531,7 +505,7 @@ CREATE TABLE `stream_source`
`serialization_type` varchar(20) DEFAULT NULL COMMENT 'Serialization type, support: csv, json, canal, avro, etc',
`snapshot` text DEFAULT NULL COMMENT 'Snapshot of this source task',
`report_time` timestamp NULL COMMENT 'Snapshot time',
- `ext_params` text DEFAULT NULL COMMENT 'Another fields will saved as JSON string, such as filePath, dbName, tableName, etc',
+ `ext_params` text DEFAULT NULL COMMENT 'Another fields will be saved as JSON string, such as filePath, dbName, tableName, etc',
`version` int(11) DEFAULT '1' COMMENT 'Stream source version',
`status` int(4) DEFAULT '0' COMMENT 'Data source status',
`previous_status` int(4) DEFAULT '0' COMMENT 'Previous status',
@@ -549,8 +523,7 @@ CREATE TABLE `stream_source`
-- ----------------------------
-- Table structure for stream_transform
-- ----------------------------
-DROP TABLE IF EXISTS `stream_transform`;
-CREATE TABLE `stream_transform`
+CREATE TABLE IF NOT EXISTS `stream_transform`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
@@ -567,14 +540,13 @@ CREATE TABLE `stream_transform`
`create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Create time',
`modify_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 'Modify time',
PRIMARY KEY (`id`),
- UNIQUE KEY `unique_transform_name` (`inlong_group_id`, `inlong_stream_id`, `transform_name`, `is_deleted`) USING BTREE
+ UNIQUE KEY `unique_transform_name` (`inlong_group_id`, `inlong_stream_id`, `transform_name`, `is_deleted`)
);
-- ----------------------------
-- Table structure for stream_sink
-- ----------------------------
-DROP TABLE IF EXISTS `stream_sink`;
-CREATE TABLE `stream_sink`
+CREATE TABLE IF NOT EXISTS `stream_sink`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Owning inlong group id',
@@ -587,7 +559,7 @@ CREATE TABLE `stream_sink`
`data_node_name` varchar(128) DEFAULT NULL COMMENT 'Node name, which links to data_node table',
`sort_task_name` varchar(512) DEFAULT NULL COMMENT 'Sort task name or task ID',
`sort_consumer_group` varchar(512) DEFAULT NULL COMMENT 'Consumer group name for Sort task',
- `ext_params` text COMMENT 'Another fields, will saved as JSON type',
+ `ext_params` text COMMENT 'Another fields, will be saved as JSON type',
`operate_log` varchar(5000) DEFAULT NULL COMMENT 'Background operate log',
`status` int(11) DEFAULT '0' COMMENT 'Status',
`previous_status` int(11) DEFAULT '0' COMMENT 'Previous status',
@@ -603,8 +575,7 @@ CREATE TABLE `stream_sink`
-- ----------------------------
-- Table structure for stream_sink_ext
-- ----------------------------
-DROP TABLE IF EXISTS `stream_sink_ext`;
-CREATE TABLE `stream_sink_ext`
+CREATE TABLE IF NOT EXISTS `stream_sink_ext`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`sink_type` varchar(20) NOT NULL COMMENT 'Sink type, including: HDFS, HIVE, etc.',
@@ -620,8 +591,7 @@ CREATE TABLE `stream_sink_ext`
-- ----------------------------
-- Table structure for stream_source_field
-- ----------------------------
-DROP TABLE IF EXISTS `stream_source_field`;
-CREATE TABLE `stream_source_field`
+CREATE TABLE IF NOT EXISTS `stream_source_field`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
@@ -645,8 +615,7 @@ CREATE TABLE `stream_source_field`
-- ----------------------------
-- Table structure for stream_transform_field
-- ----------------------------
-DROP TABLE IF EXISTS `stream_transform_field`;
-CREATE TABLE `stream_transform_field`
+CREATE TABLE IF NOT EXISTS `stream_transform_field`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
@@ -672,8 +641,7 @@ CREATE TABLE `stream_transform_field`
-- ----------------------------
-- Table structure for stream_sink_field
-- ----------------------------
-DROP TABLE IF EXISTS `stream_sink_field`;
-CREATE TABLE `stream_sink_field`
+CREATE TABLE IF NOT EXISTS `stream_sink_field`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
@@ -699,8 +667,7 @@ CREATE TABLE `stream_sink_field`
-- ----------------------------
-- Table structure for user
-- ----------------------------
-DROP TABLE IF EXISTS `user`;
-CREATE TABLE `user`
+CREATE TABLE IF NOT EXISTS `user`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`name` varchar(256) NOT NULL COMMENT 'account name',
@@ -715,20 +682,14 @@ CREATE TABLE `user`
UNIQUE KEY `unique_user_name_idx` (`name`)
);
--- create default admin user, username is 'admin', password is 'inlong'
-INSERT INTO `user` (name, password, account_type, due_date, create_time, update_time, create_by, update_by)
-VALUES ('admin', '628ed559bff5ae36bd2184d4216973cf', 0, '2099-12-31 23:59:59',
- CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, 'inlong_init', 'inlong_init');
-
-- ----------------------------
-- Table structure for user_role
-- ----------------------------
-DROP TABLE IF EXISTS `user_role`;
-CREATE TABLE `user_role`
+CREATE TABLE IF NOT EXISTS `user_role`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
- `user_name` varchar(256) NOT NULL COMMENT 'username rtx',
- `role_code` varchar(256) NOT NULL COMMENT 'role',
+ `user_name` varchar(256) NOT NULL COMMENT 'username',
+ `role_code` varchar(256) NOT NULL COMMENT 'role code',
`create_time` datetime NOT NULL,
`update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
`create_by` varchar(256) NOT NULL,
@@ -740,8 +701,7 @@ CREATE TABLE `user_role`
-- ----------------------------
-- Table structure for workflow_approver
-- ----------------------------
-DROP TABLE IF EXISTS `workflow_approver`;
-CREATE TABLE `workflow_approver`
+CREATE TABLE IF NOT EXISTS `workflow_approver`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`process_name` varchar(256) NOT NULL COMMENT 'Process name',
@@ -770,8 +730,7 @@ VALUES ('NEW_CONSUMPTION_PROCESS', 'ut_admin', 'DEFAULT', NULL, 'admin',
-- ----------------------------
-- Table structure for workflow_event_log
-- ----------------------------
-DROP TABLE IF EXISTS `workflow_event_log`;
-CREATE TABLE `workflow_event_log`
+CREATE TABLE IF NOT EXISTS `workflow_event_log`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`process_id` int(11) NOT NULL,
@@ -797,8 +756,7 @@ CREATE TABLE `workflow_event_log`
-- ----------------------------
-- Table structure for workflow_process
-- ----------------------------
-DROP TABLE IF EXISTS `workflow_process`;
-CREATE TABLE `workflow_process`
+CREATE TABLE IF NOT EXISTS `workflow_process`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`name` varchar(256) NOT NULL COMMENT 'process name',
@@ -819,8 +777,7 @@ CREATE TABLE `workflow_process`
-- ----------------------------
-- Table structure for workflow_task
-- ----------------------------
-DROP TABLE IF EXISTS `workflow_task`;
-CREATE TABLE `workflow_task`
+CREATE TABLE IF NOT EXISTS `workflow_task`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`type` varchar(64) NOT NULL COMMENT 'Task type: UserTask / ServiceTask',
@@ -844,8 +801,7 @@ CREATE TABLE `workflow_task`
-- ----------------------------
-- Table structure for cluster_set
-- ----------------------------
-DROP TABLE IF EXISTS `cluster_set`;
-CREATE TABLE `cluster_set`
+CREATE TABLE IF NOT EXISTS `cluster_set`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`set_name` varchar(128) NOT NULL COMMENT 'ClusterSet name, English, numbers and underscore',
@@ -867,8 +823,7 @@ CREATE TABLE `cluster_set`
-- ----------------------------
-- Table structure for cluster_set_inlongid
-- ----------------------------
-DROP TABLE IF EXISTS `cluster_set_inlongid`;
-CREATE TABLE `cluster_set_inlongid`
+CREATE TABLE IF NOT EXISTS `cluster_set_inlongid`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`set_name` varchar(256) NOT NULL COMMENT 'ClusterSet name, English, numbers and underscore',
@@ -880,8 +835,7 @@ CREATE TABLE `cluster_set_inlongid`
-- ----------------------------
-- Table structure for cache_cluster
-- ----------------------------
-DROP TABLE IF EXISTS `cache_cluster`;
-CREATE TABLE `cache_cluster`
+CREATE TABLE IF NOT EXISTS `cache_cluster`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`cluster_name` varchar(128) NOT NULL COMMENT 'CacheCluster name, English, numbers and underscore',
@@ -894,8 +848,7 @@ CREATE TABLE `cache_cluster`
-- ----------------------------
-- Table structure for cache_cluster_ext
-- ----------------------------
-DROP TABLE IF EXISTS `cache_cluster_ext`;
-CREATE TABLE `cache_cluster_ext`
+CREATE TABLE IF NOT EXISTS `cache_cluster_ext`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`cluster_name` varchar(128) NOT NULL COMMENT 'CacheCluster name, English, numbers and underscore',
@@ -910,8 +863,7 @@ CREATE TABLE `cache_cluster_ext`
-- ----------------------------
-- Table structure for cache_topic
-- ----------------------------
-DROP TABLE IF EXISTS `cache_topic`;
-CREATE TABLE `cache_topic`
+CREATE TABLE IF NOT EXISTS `cache_topic`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`topic_name` varchar(128) NOT NULL COMMENT 'Topic name, English, numbers and underscore',
@@ -924,8 +876,7 @@ CREATE TABLE `cache_topic`
-- ----------------------------
-- Table structure for proxy_cluster
-- ----------------------------
-DROP TABLE IF EXISTS `proxy_cluster`;
-CREATE TABLE `proxy_cluster`
+CREATE TABLE IF NOT EXISTS `proxy_cluster`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`cluster_name` varchar(128) NOT NULL COMMENT 'ProxyCluster name, English, numbers and underscore',
@@ -938,8 +889,7 @@ CREATE TABLE `proxy_cluster`
-- ----------------------------
-- Table structure for proxy_cluster_to_cache_cluster
-- ----------------------------
-DROP TABLE IF EXISTS `proxy_cluster_to_cache_cluster`;
-CREATE TABLE `proxy_cluster_to_cache_cluster`
+CREATE TABLE IF NOT EXISTS `proxy_cluster_to_cache_cluster`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`proxy_cluster_name` varchar(128) NOT NULL COMMENT 'ProxyCluster name, English, numbers and underscore',
@@ -951,8 +901,7 @@ CREATE TABLE `proxy_cluster_to_cache_cluster`
-- ----------------------------
-- Table structure for flume_source
-- ----------------------------
-DROP TABLE IF EXISTS `flume_source`;
-CREATE TABLE `flume_source`
+CREATE TABLE IF NOT EXISTS `flume_source`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`source_name` varchar(128) NOT NULL COMMENT 'FlumeSource name, English, numbers and underscore',
@@ -967,8 +916,7 @@ CREATE TABLE `flume_source`
-- ----------------------------
-- Table structure for flume_source_ext
-- ----------------------------
-DROP TABLE IF EXISTS `flume_source_ext`;
-CREATE TABLE `flume_source_ext`
+CREATE TABLE IF NOT EXISTS `flume_source_ext`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`parent_name` varchar(128) NOT NULL COMMENT 'FlumeSource name, English, numbers and underscore',
@@ -984,8 +932,7 @@ CREATE TABLE `flume_source_ext`
-- ----------------------------
-- Table structure for flume_channel
-- ----------------------------
-DROP TABLE IF EXISTS `flume_channel`;
-CREATE TABLE `flume_channel`
+CREATE TABLE IF NOT EXISTS `flume_channel`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`channel_name` varchar(128) NOT NULL COMMENT 'FlumeChannel name, English, numbers and underscore',
@@ -998,8 +945,7 @@ CREATE TABLE `flume_channel`
-- ----------------------------
-- Table structure for flume_channel_ext
-- ----------------------------
-DROP TABLE IF EXISTS `flume_channel_ext`;
-CREATE TABLE `flume_channel_ext`
+CREATE TABLE IF NOT EXISTS `flume_channel_ext`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`parent_name` varchar(128) NOT NULL COMMENT 'FlumeChannel name, English, numbers and underscore',
@@ -1015,8 +961,7 @@ CREATE TABLE `flume_channel_ext`
-- ----------------------------
-- Table structure for flume_sink
-- ----------------------------
-DROP TABLE IF EXISTS `flume_sink`;
-CREATE TABLE `flume_sink`
+CREATE TABLE IF NOT EXISTS `flume_sink`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`sink_name` varchar(128) NOT NULL COMMENT 'FlumeSink name, English, numbers and underscore',
@@ -1030,8 +975,7 @@ CREATE TABLE `flume_sink`
-- ----------------------------
-- Table structure for flume_sink_ext
-- ----------------------------
-DROP TABLE IF EXISTS `flume_sink_ext`;
-CREATE TABLE `flume_sink_ext`
+CREATE TABLE IF NOT EXISTS `flume_sink_ext`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`parent_name` varchar(128) NOT NULL COMMENT 'FlumeSink name, English, numbers and underscore',
@@ -1048,8 +992,7 @@ CREATE TABLE `flume_sink_ext`
-- ----------------------------
-- Table structure for db_collector_detail_task
-- ----------------------------
-DROP TABLE IF EXISTS `db_collector_detail_task`;
-CREATE TABLE `db_collector_detail_task`
+CREATE TABLE IF NOT EXISTS `db_collector_detail_task`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`main_id` varchar(128) NOT NULL COMMENT 'main task id',
@@ -1078,8 +1021,7 @@ CREATE TABLE `db_collector_detail_task`
-- ----------------------------
-- Table structure for sort_cluster_config
-- ----------------------------
-DROP TABLE IF EXISTS `sort_cluster_config`;
-CREATE TABLE `sort_cluster_config`
+CREATE TABLE IF NOT EXISTS `sort_cluster_config`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`cluster_name` varchar(128) NOT NULL COMMENT 'Cluster name',
@@ -1092,8 +1034,7 @@ CREATE TABLE `sort_cluster_config`
-- ----------------------------
-- Table structure for sort_task_id_param
-- ----------------------------
-DROP TABLE IF EXISTS `sort_task_id_param`;
-CREATE TABLE `sort_task_id_param`
+CREATE TABLE IF NOT EXISTS `sort_task_id_param`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`task_name` varchar(128) NOT NULL COMMENT 'Task name',
@@ -1108,8 +1049,7 @@ CREATE TABLE `sort_task_id_param`
-- ----------------------------
-- Table structure for sort_task_sink_param
-- ----------------------------
-DROP TABLE IF EXISTS `sort_task_sink_param`;
-CREATE TABLE `sort_task_sink_param`
+CREATE TABLE IF NOT EXISTS `sort_task_sink_param`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`task_name` varchar(128) NOT NULL COMMENT 'Task name',
@@ -1123,15 +1063,14 @@ CREATE TABLE `sort_task_sink_param`
-- ----------------------------
-- Table structure for sort_source_config
-- ----------------------------
-DROP TABLE IF EXISTS `sort_source_config`;
-CREATE TABLE `sort_source_config`
+CREATE TABLE IF NOT EXISTS `sort_source_config`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`cluster_name` varchar(128) NOT NULL COMMENT 'Cluster name',
`task_name` varchar(128) NOT NULL COMMENT 'Task name',
`zone_name` varchar(128) NOT NULL COMMENT 'Cache zone name',
`topic` varchar(128) DEFAULT '' COMMENT 'Topic',
- `ext_params` text DEFAULT NULL COMMENT 'Another fields, will saved as JSON type',
+ `ext_params` text DEFAULT NULL COMMENT 'Another fields, will be saved as JSON type',
PRIMARY KEY (`id`),
KEY `index_sort_source_config` (`cluster_name`, `task_name`)
);
@@ -1140,8 +1079,7 @@ CREATE TABLE `sort_source_config`
-- ----------------------------
-- Table structure for config log report
-- ----------------------------
-DROP TABLE IF EXISTS `stream_config_log`;
-CREATE TABLE `stream_config_log`
+CREATE TABLE IF NOT EXISTS `stream_config_log`
(
`ip` varchar(24) NOT NULL COMMENT 'client host ip',
`version` varchar(64) DEFAULT NULL COMMENT 'client version',
@@ -1159,8 +1097,7 @@ CREATE TABLE `stream_config_log`
-- ----------------------------
-- Table structure for inlong component heartbeat
-- ----------------------------
-DROP TABLE IF EXISTS `component_heartbeat`;
-CREATE TABLE `component_heartbeat`
+CREATE TABLE IF NOT EXISTS `component_heartbeat`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`component` varchar(64) NOT NULL DEFAULT '' COMMENT 'Component name, such as: Agent, Sort...',
@@ -1177,8 +1114,7 @@ CREATE TABLE `component_heartbeat`
-- ----------------------------
-- Table structure for inlong group heartbeat
-- ----------------------------
-DROP TABLE IF EXISTS `group_heartbeat`;
-CREATE TABLE `group_heartbeat`
+CREATE TABLE IF NOT EXISTS `group_heartbeat`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`component` varchar(64) NOT NULL DEFAULT '' COMMENT 'Component name, such as: Agent, Sort...',
@@ -1196,8 +1132,7 @@ CREATE TABLE `group_heartbeat`
-- ----------------------------
-- Table structure for inlong stream heartbeat
-- ----------------------------
-DROP TABLE IF EXISTS `stream_heartbeat`;
-CREATE TABLE `stream_heartbeat`
+CREATE TABLE IF NOT EXISTS `stream_heartbeat`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`component` varchar(64) NOT NULL DEFAULT '' COMMENT 'Component name, such as: Agent, Sort...',
diff --git a/inlong-manager/manager-web/sql/apache_inlong_manager.sql b/inlong-manager/manager-web/sql/apache_inlong_manager.sql
index ab0b25b5b..dc1ccaa47 100644
--- a/inlong-manager/manager-web/sql/apache_inlong_manager.sql
+++ b/inlong-manager/manager-web/sql/apache_inlong_manager.sql
@@ -27,34 +27,32 @@ USE apache_inlong_manager;
-- ----------------------------
-- Table structure for inlong_group
-- ----------------------------
-DROP TABLE IF EXISTS `inlong_group`;
-CREATE TABLE `inlong_group`
+CREATE TABLE IF NOT EXISTS `inlong_group`
(
- `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
- `inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id, filled in by the user, undeleted ones cannot be repeated',
- `name` varchar(128) DEFAULT '' COMMENT 'Inlong group name, English, numbers and underscore',
- `cn_name` varchar(256) DEFAULT NULL COMMENT 'Chinese display name',
- `description` varchar(256) DEFAULT '' COMMENT 'Inlong group Introduction',
- `middleware_type` varchar(20) DEFAULT 'TUBE' COMMENT 'The middleware type of message queue, high throughput: TUBE, high consistency: PULSAR',
- `queue_module` VARCHAR(20) NULL DEFAULT 'parallel' COMMENT 'Queue model of Pulsar, parallel: multiple partitions, high throughput, out-of-order messages; serial: single partition, low throughput, and orderly messages',
- `topic_partition_num` INT(4) NULL DEFAULT '3' COMMENT 'The number of partitions of Pulsar Topic, 1-20',
- `mq_resource_obj` varchar(128) NOT NULL COMMENT 'MQ resource object, for Tube, its Topic, for Pulsar, its Namespace',
- `daily_records` int(11) DEFAULT '10' COMMENT 'Number of access records per day, unit: 10,000 records per day',
- `daily_storage` int(11) DEFAULT '10' COMMENT 'Access size by day, unit: GB per day',
- `peak_records` int(11) DEFAULT '1000' COMMENT 'Access peak per second, unit: records per second',
- `max_length` int(11) DEFAULT '10240' COMMENT 'The maximum length of a single piece of data, unit: Byte',
- `schema_name` varchar(128) DEFAULT NULL COMMENT 'Data type, associated data_schema table',
- `in_charges` varchar(512) NOT NULL COMMENT 'Name of responsible person, separated by commas',
- `followers` varchar(512) DEFAULT NULL COMMENT 'Name of followers, separated by commas',
- `status` int(4) DEFAULT '21' COMMENT 'Inlong group status',
- `is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete, 0: not deleted, > 0: deleted',
- `creator` varchar(64) NOT NULL COMMENT 'Creator name',
- `modifier` varchar(64) DEFAULT NULL COMMENT 'Modifier name',
- `create_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Create time',
- `modify_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 'Modify time',
- `temp_view` text DEFAULT NULL COMMENT 'Temporary view, used to save intermediate data that has not been submitted or approved after modification',
- `zookeeper_enabled` int(4) DEFAULT '1' COMMENT 'Need zookeeper support, 0: false, 1: true',
- `proxy_cluster_id` int(11) DEFAULT NULL COMMENT 'The id of dataproxy cluster',
+ `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
+ `inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id, filled in by the user, undeleted ones cannot be repeated',
+ `name` varchar(128) DEFAULT '' COMMENT 'Inlong group name, English, Chinese, numbers, etc',
+ `description` varchar(256) DEFAULT '' COMMENT 'Inlong group Introduction',
+ `mq_type` varchar(20) DEFAULT 'TUBE' COMMENT 'The message queue type, high throughput: TUBE, high consistency: PULSAR',
+ `mq_resource` varchar(128) NOT NULL COMMENT 'MQ resource, for Tube, its Topic, for Pulsar, its Namespace',
+ `daily_records` int(11) DEFAULT '10' COMMENT 'Number of access records per day, unit: 10,000 records per day',
+ `daily_storage` int(11) DEFAULT '10' COMMENT 'Access size by day, unit: GB per day',
+ `peak_records` int(11) DEFAULT '1000' COMMENT 'Access peak per second, unit: records per second',
+ `max_length` int(11) DEFAULT '10240' COMMENT 'The maximum length of a single piece of data, unit: Byte',
+ `enable_zookeeper` tinyint(2) DEFAULT '0' COMMENT 'Whether to enable zookeeper? 0: disable, 1: enable. Default is 0',
+ `enable_create_resource` tinyint(2) DEFAULT '1' COMMENT 'Whether to enable create mq resource? 0: disable, 1: enable. Default is 1',
+ `lightweight` tinyint(2) DEFAULT '1' COMMENT 'Whether to use lightweight mode, 0: false, 1: true. Default is 1',
+ `inlong_cluster_tag` varchar(128) DEFAULT NULL COMMENT 'The cluster tag, which links to inlong_cluster table',
+ `ext_params` text DEFAULT NULL COMMENT 'Extended params, will be saved as JSON string, such as queue_module, partition_num',
+ `in_charges` varchar(512) NOT NULL COMMENT 'Name of responsible person, separated by commas',
+ `followers` varchar(512) DEFAULT NULL COMMENT 'Name of followers, separated by commas',
+ `status` int(4) DEFAULT '100' COMMENT 'Inlong group status',
+ `previous_status` int(4) DEFAULT '100' COMMENT 'Previous group status',
+ `is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete, 0: not deleted, > 0: deleted',
+ `creator` varchar(64) NOT NULL COMMENT 'Creator name',
+ `modifier` varchar(64) DEFAULT NULL COMMENT 'Modifier name',
+ `create_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Create time',
+ `modify_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 'Modify time',
PRIMARY KEY (`id`),
UNIQUE KEY `unique_inlong_group` (`inlong_group_id`, `is_deleted`)
) ENGINE = InnoDB
@@ -63,8 +61,7 @@ CREATE TABLE `inlong_group`
-- ----------------------------
-- Table structure for inlong_group_pulsar
-- ----------------------------
-DROP TABLE IF EXISTS `inlong_group_pulsar`;
-CREATE TABLE `inlong_group_pulsar`
+CREATE TABLE IF NOT EXISTS `inlong_group_pulsar`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id, filled in by the user, undeleted ones cannot be repeated',
@@ -89,8 +86,7 @@ CREATE TABLE `inlong_group_pulsar`
-- ----------------------------
-- Table structure for inlong_group_ext
-- ----------------------------
-DROP TABLE IF EXISTS `inlong_group_ext`;
-CREATE TABLE `inlong_group_ext`
+CREATE TABLE IF NOT EXISTS `inlong_group_ext`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
@@ -140,7 +136,7 @@ CREATE TABLE IF NOT EXISTS `inlong_cluster_node`
`type` varchar(20) DEFAULT '' COMMENT 'Cluster type, such as: DATA_PROXY, AGENT, etc',
`ip` varchar(512) NULL COMMENT 'Cluster IP, separated by commas, such as: 127.0.0.1:8080,host2:8081',
`port` int(6) NULL COMMENT 'Cluster port',
- `ext_params` text DEFAULT NULL COMMENT 'Another fields will saved as JSON string',
+ `ext_params` text DEFAULT NULL COMMENT 'Another fields will be saved as JSON string',
`status` int(4) DEFAULT '0' COMMENT 'Cluster status',
`is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete, 0: not deleted, > 0: deleted',
`creator` varchar(64) NOT NULL COMMENT 'Creator name',
@@ -163,7 +159,7 @@ CREATE TABLE IF NOT EXISTS `data_node`
`url` varchar(512) DEFAULT NULL COMMENT 'Node URL',
`username` varchar(128) DEFAULT NULL COMMENT 'Username for node',
`token` varchar(512) DEFAULT NULL COMMENT 'Node token',
- `ext_params` text DEFAULT NULL COMMENT 'Extended params, will saved as JSON string',
+ `ext_params` text DEFAULT NULL COMMENT 'Extended params, will be saved as JSON string',
`in_charges` varchar(512) NOT NULL COMMENT 'Name of responsible person, separated by commas',
`status` int(4) DEFAULT '0' COMMENT 'Node status',
`is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete, 0: not deleted, > 0: deleted',
@@ -179,8 +175,7 @@ CREATE TABLE IF NOT EXISTS `data_node`
-- ----------------------------
-- Table structure for third_party_cluster
-- ----------------------------
-DROP TABLE IF EXISTS `third_party_cluster`;
-CREATE TABLE `third_party_cluster`
+CREATE TABLE IF NOT EXISTS `third_party_cluster`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`name` varchar(128) NOT NULL COMMENT 'Cluster name',
@@ -207,8 +202,7 @@ CREATE TABLE `third_party_cluster`
-- ----------------------------
-- Table structure for common_db_server
-- ----------------------------
-DROP TABLE IF EXISTS `common_db_server`;
-CREATE TABLE `common_db_server`
+CREATE TABLE IF NOT EXISTS `common_db_server`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`access_type` varchar(20) NOT NULL COMMENT 'Collection type, with Agent, DataProxy client, LoadProxy',
@@ -243,8 +237,7 @@ CREATE TABLE `common_db_server`
-- ----------------------------
-- Table structure for common_file_server
-- ----------------------------
-DROP TABLE IF EXISTS `common_file_server`;
-CREATE TABLE `common_file_server`
+CREATE TABLE IF NOT EXISTS `common_file_server`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`access_type` varchar(20) NOT NULL COMMENT 'Collection type, with Agent, DataProxy, LoadProxy',
@@ -269,8 +262,7 @@ CREATE TABLE `common_file_server`
-- ----------------------------
-- Table structure for consumption
-- ----------------------------
-DROP TABLE IF EXISTS `consumption`;
-CREATE TABLE `consumption`
+CREATE TABLE IF NOT EXISTS `consumption`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`consumer_group_name` varchar(256) DEFAULT NULL COMMENT 'consumer group name',
@@ -294,8 +286,7 @@ CREATE TABLE `consumption`
-- ----------------------------
-- Table structure for consumption_pulsar
-- ----------------------------
-DROP TABLE IF EXISTS `consumption_pulsar`;
-CREATE TABLE `consumption_pulsar`
+CREATE TABLE IF NOT EXISTS `consumption_pulsar`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`consumption_id` int(11) DEFAULT NULL COMMENT 'ID of the consumption information to which it belongs, guaranteed to be uniquely associated with consumption information',
@@ -314,8 +305,7 @@ CREATE TABLE `consumption_pulsar`
-- ----------------------------
-- Table structure for data_proxy_cluster
-- ----------------------------
-DROP TABLE IF EXISTS `data_proxy_cluster`;
-CREATE TABLE `data_proxy_cluster`
+CREATE TABLE IF NOT EXISTS `data_proxy_cluster`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`name` varchar(128) NOT NULL COMMENT 'Cluster name',
@@ -340,8 +330,7 @@ CREATE TABLE `data_proxy_cluster`
-- ----------------------------
-- Table structure for data_schema
-- ----------------------------
-DROP TABLE IF EXISTS `data_schema`;
-CREATE TABLE `data_schema`
+CREATE TABLE IF NOT EXISTS `data_schema`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`name` varchar(128) NOT NULL COMMENT 'Data format name, globally unique',
@@ -361,8 +350,7 @@ values ('m0_day', 'file_agent', 'day', 0, '-0d');
-- ----------------------------
-- Table structure for stream_source_cmd_config
-- ----------------------------
-DROP TABLE IF EXISTS `stream_source_cmd_config`;
-CREATE TABLE `stream_source_cmd_config`
+CREATE TABLE IF NOT EXISTS `stream_source_cmd_config`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'cmd id',
`cmd_type` int(11) NOT NULL,
@@ -380,36 +368,32 @@ CREATE TABLE `stream_source_cmd_config`
-- ----------------------------
-- Table structure for inlong_stream
-- ----------------------------
-DROP TABLE IF EXISTS `inlong_stream`;
-CREATE TABLE `inlong_stream`
+CREATE TABLE IF NOT EXISTS `inlong_stream`
(
- `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
- `inlong_stream_id` varchar(256) NOT NULL COMMENT 'Inlong stream id, non-deleted globally unique',
- `inlong_group_id` varchar(256) NOT NULL COMMENT 'Owning inlong group id',
- `name` varchar(64) DEFAULT NULL COMMENT 'The name of the inlong stream page display, can be Chinese',
- `description` varchar(256) DEFAULT '' COMMENT 'Introduction to inlong stream',
- `mq_resource_obj` varchar(128) DEFAULT NULL COMMENT 'MQ resource object, in the inlong stream, Tube is inlong_stream_id, Pulsar is Topic',
- `data_source_type` varchar(32) DEFAULT 'FILE' COMMENT 'Data source type, including: FILE, BINLOG, AUTO_PUSH, etc.',
- `storage_period` int(11) DEFAULT '1' COMMENT 'The storage period of data in MQ, unit: day',
- `data_type` varchar(20) DEFAULT NULL COMMENT 'Data type, including: CSV, KEY-VALUE, JSON, AVRO, etc.',
- `data_encoding` varchar(8) DEFAULT 'UTF-8' COMMENT 'Data encoding format, including: UTF-8, GBK',
- `data_separator` varchar(8) DEFAULT NULL COMMENT 'The source data field separator, stored as ASCII code',
- `data_escape_char` varchar(8) DEFAULT NULL COMMENT 'Source data field escape character, the default is NULL (NULL), stored as 1 character',
- `have_predefined_fields` tinyint(1) DEFAULT '0' COMMENT '(File, DB access) whether there are predefined fields, 0: none, 1: yes (save to inlong_stream_field)',
- `sync_send` tinyint(1) DEFAULT '0' COMMENT 'order_preserving 0: none, 1: yes',
- `daily_records` int(11) DEFAULT '10' COMMENT 'Number of access records per day, unit: 10,000 records per day',
- `daily_storage` int(11) DEFAULT '10' COMMENT 'Access size by day, unit: GB per day',
- `peak_records` int(11) DEFAULT '1000' COMMENT 'Access peak per second, unit: records per second',
- `max_length` int(11) DEFAULT '10240' COMMENT 'The maximum length of a single piece of data, unit: Byte',
- `in_charges` varchar(512) DEFAULT NULL COMMENT 'Name of responsible person, separated by commas',
- `status` int(4) DEFAULT '0' COMMENT 'Inlong stream status',
- `previous_status` int(4) DEFAULT '0' COMMENT 'Previous status',
- `is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete, 0: not deleted, > 0: deleted',
- `creator` varchar(64) DEFAULT NULL COMMENT 'Creator name',
- `modifier` varchar(64) DEFAULT NULL COMMENT 'Modifier name',
- `create_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Create time',
- `modify_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 'Modify time',
- `temp_view` text DEFAULT NULL COMMENT 'Temporary view, used to save intermediate data that has not been submitted or approved after modification',
+ `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
+ `inlong_group_id` varchar(256) NOT NULL COMMENT 'Owning inlong group id',
+ `inlong_stream_id` varchar(256) NOT NULL COMMENT 'Inlong stream id, non-deleted globally unique',
+ `name` varchar(64) DEFAULT NULL COMMENT 'The name of the inlong stream page display, can be Chinese',
+ `description` varchar(256) DEFAULT '' COMMENT 'Introduction to inlong stream',
+ `mq_resource` varchar(128) DEFAULT NULL COMMENT 'MQ resource, in one stream, corresponding to the filter ID of Tube, corresponding to the topic of Pulsar',
+ `data_type` varchar(20) DEFAULT NULL COMMENT 'Data type, including: CSV, KEY-VALUE, JSON, AVRO, etc.',
+ `data_encoding` varchar(8) DEFAULT 'UTF-8' COMMENT 'Data encoding format, including: UTF-8, GBK, etc.',
+ `data_separator` varchar(8) DEFAULT NULL COMMENT 'The source data field separator, stored as ASCII code',
+ `data_escape_char` varchar(8) DEFAULT NULL COMMENT 'Source data field escape character, the default is NULL (NULL), stored as 1 character',
+ `sync_send` tinyint(1) DEFAULT '0' COMMENT 'order_preserving 0: none, 1: yes',
+ `daily_records` int(11) DEFAULT '10' COMMENT 'Number of access records per day, unit: 10,000 records per day',
+ `daily_storage` int(11) DEFAULT '10' COMMENT 'Access size by day, unit: GB per day',
+ `peak_records` int(11) DEFAULT '1000' COMMENT 'Access peak per second, unit: records per second',
+ `max_length` int(11) DEFAULT '10240' COMMENT 'The maximum length of a single piece of data, unit: Byte',
+ `storage_period` int(11) DEFAULT '1' COMMENT 'The storage period of data in MQ, unit: day',
+ `ext_params` text DEFAULT NULL COMMENT 'Extended params, will be saved as JSON string',
+ `status` int(4) DEFAULT '100' COMMENT 'Inlong stream status',
+ `previous_status` int(4) DEFAULT '100' COMMENT 'Previous status',
+ `is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete, 0: not deleted, > 0: deleted',
+ `creator` varchar(64) DEFAULT NULL COMMENT 'Creator name',
+ `modifier` varchar(64) DEFAULT NULL COMMENT 'Modifier name',
+ `create_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Create time',
+ `modify_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 'Modify time',
PRIMARY KEY (`id`),
UNIQUE KEY `unique_inlong_stream` (`inlong_stream_id`, `inlong_group_id`, `is_deleted`)
) ENGINE = InnoDB
@@ -418,8 +402,7 @@ CREATE TABLE `inlong_stream`
-- ----------------------------
-- Table structure for inlong_stream_field
-- ----------------------------
-DROP TABLE IF EXISTS `inlong_stream_field`;
-CREATE TABLE `inlong_stream_field`
+CREATE TABLE IF NOT EXISTS `inlong_stream_field`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Owning inlong group id',
@@ -442,8 +425,7 @@ CREATE TABLE `inlong_stream_field`
-- ----------------------------
-- Table structure for operation_log
-- ----------------------------
-DROP TABLE IF EXISTS `operation_log`;
-CREATE TABLE `operation_log`
+CREATE TABLE IF NOT EXISTS `operation_log`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`authentication_type` varchar(64) DEFAULT NULL COMMENT 'Authentication type',
@@ -467,8 +449,7 @@ CREATE TABLE `operation_log`
-- ----------------------------
-- Table structure for role
-- ----------------------------
-DROP TABLE IF EXISTS `role`;
-CREATE TABLE `role`
+CREATE TABLE IF NOT EXISTS `role`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`role_code` varchar(100) NOT NULL COMMENT 'Role code',
@@ -487,8 +468,7 @@ CREATE TABLE `role`
-- ----------------------------
-- Table structure for source_file_basic
-- ----------------------------
-DROP TABLE IF EXISTS `source_file_basic`;
-CREATE TABLE `source_file_basic`
+CREATE TABLE IF NOT EXISTS `source_file_basic`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'ID',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
@@ -513,8 +493,7 @@ CREATE TABLE `source_file_basic`
-- ----------------------------
-- Table structure for source_file_detail
-- ----------------------------
-DROP TABLE IF EXISTS `source_file_detail`;
-CREATE TABLE `source_file_detail`
+CREATE TABLE IF NOT EXISTS `source_file_detail`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Owning inlong group id',
@@ -543,8 +522,7 @@ CREATE TABLE `source_file_detail`
-- ----------------------------
-- Table structure for stream_source
-- ----------------------------
-DROP TABLE IF EXISTS `stream_source`;
-CREATE TABLE `stream_source`
+CREATE TABLE IF NOT EXISTS `stream_source`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'ID',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
@@ -558,7 +536,7 @@ CREATE TABLE `stream_source`
`serialization_type` varchar(20) DEFAULT NULL COMMENT 'Serialization type, support: csv, json, canal, avro, etc',
`snapshot` text DEFAULT NULL COMMENT 'Snapshot of this source task',
`report_time` timestamp NULL COMMENT 'Snapshot time',
- `ext_params` text DEFAULT NULL COMMENT 'Another fields will saved as JSON string, such as filePath, dbName, tableName, etc',
+ `ext_params` text DEFAULT NULL COMMENT 'Another fields will be saved as JSON string, such as filePath, dbName, tableName, etc',
`version` int(11) DEFAULT '1' COMMENT 'Stream source version',
`status` int(4) DEFAULT '0' COMMENT 'Data source status',
`previous_status` int(4) DEFAULT '0' COMMENT 'Previous status',
@@ -577,8 +555,7 @@ CREATE TABLE `stream_source`
-- ----------------------------
-- Table structure for stream_transform
-- ----------------------------
-DROP TABLE IF EXISTS `stream_transform`;
-CREATE TABLE `stream_transform`
+CREATE TABLE IF NOT EXISTS `stream_transform`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
@@ -602,8 +579,7 @@ CREATE TABLE `stream_transform`
-- ----------------------------
-- Table structure for stream_sink
-- ----------------------------
-DROP TABLE IF EXISTS `stream_sink`;
-CREATE TABLE `stream_sink`
+CREATE TABLE IF NOT EXISTS `stream_sink`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Owning inlong group id',
@@ -611,12 +587,12 @@ CREATE TABLE `stream_sink`
`sink_type` varchar(15) DEFAULT 'HIVE' COMMENT 'Sink type, including: HIVE, ES, etc',
`sink_name` varchar(128) NOT NULL DEFAULT '' COMMENT 'Sink name',
`description` varchar(500) NULL COMMENT 'Sink description',
- `enable_create_resource` tinyint(1) DEFAULT '1' COMMENT 'Whether to enable create sink resource? 0: disable, 1: enable. default is 1',
+ `enable_create_resource` tinyint(2) DEFAULT '1' COMMENT 'Whether to enable create sink resource? 0: disable, 1: enable. default is 1',
`inlong_cluster_name` varchar(128) DEFAULT NULL COMMENT 'Cluster name, which links to inlong_cluster table',
`data_node_name` varchar(128) DEFAULT NULL COMMENT 'Node name, which links to data_node table',
`sort_task_name` varchar(512) DEFAULT NULL COMMENT 'Sort task name or task ID',
`sort_consumer_group` varchar(512) DEFAULT NULL COMMENT 'Consumer group name for Sort task',
- `ext_params` text NULL COMMENT 'Another fields, will saved as JSON type',
+ `ext_params` text NULL COMMENT 'Another fields, will be saved as JSON type',
`operate_log` text DEFAULT NULL COMMENT 'Background operate log',
`status` int(11) DEFAULT '0' COMMENT 'Status',
`previous_status` int(11) DEFAULT '0' COMMENT 'Previous status',
@@ -633,8 +609,7 @@ CREATE TABLE `stream_sink`
-- ----------------------------
-- Table structure for stream_sink_ext
-- ----------------------------
-DROP TABLE IF EXISTS `stream_sink_ext`;
-CREATE TABLE `stream_sink_ext`
+CREATE TABLE IF NOT EXISTS `stream_sink_ext`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`sink_type` varchar(20) NOT NULL COMMENT 'Sink type, including: HDFS, HIVE, etc.',
@@ -651,8 +626,7 @@ CREATE TABLE `stream_sink_ext`
-- ----------------------------
-- Table structure for stream_source_field
-- ----------------------------
-DROP TABLE IF EXISTS `stream_source_field`;
-CREATE TABLE `stream_source_field`
+CREATE TABLE IF NOT EXISTS `stream_source_field`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
@@ -676,8 +650,7 @@ CREATE TABLE `stream_source_field`
-- ----------------------------
-- Table structure for stream_transform_field
-- ----------------------------
-DROP TABLE IF EXISTS `stream_transform_field`;
-CREATE TABLE `stream_transform_field`
+CREATE TABLE IF NOT EXISTS `stream_transform_field`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
@@ -693,7 +666,8 @@ CREATE TABLE `stream_transform_field`
`field_format` varchar(50) DEFAULT NULL COMMENT 'Field format, including: MICROSECONDS, MILLISECONDS, SECONDS, SQL, ISO_8601 and custom such as yyyy-MM-dd HH:mm:ss',
`rank_num` smallint(6) DEFAULT '0' COMMENT 'Field order (front-end display field order)',
`is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete, 0: not deleted, > 0: deleted',
- `origin_node_name` varchar(256) DEFAULT '' COMMENT 'Origin Node name which stream field belongs',
+ `origin_node_name` varchar(256) DEFAULT '' COMMENT 'Origin node name which stream field belongs',
+ -- The source node name of the transport field
`origin_field_name` varchar(50) DEFAULT '' COMMENT 'Origin field name before transform operation',
PRIMARY KEY (`id`),
KEY `index_transform_id` (`transform_id`)
@@ -703,8 +677,7 @@ CREATE TABLE `stream_transform_field`
-- ----------------------------
-- Table structure for stream_sink_field
-- ----------------------------
-DROP TABLE IF EXISTS `stream_sink_field`;
-CREATE TABLE `stream_sink_field`
+CREATE TABLE IF NOT EXISTS `stream_sink_field`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id',
@@ -731,8 +704,7 @@ CREATE TABLE `stream_sink_field`
-- ----------------------------
-- Table structure for user
-- ----------------------------
-DROP TABLE IF EXISTS `user`;
-CREATE TABLE `user`
+CREATE TABLE IF NOT EXISTS `user`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`name` varchar(256) NOT NULL COMMENT 'account name',
@@ -756,8 +728,7 @@ VALUES ('admin', '628ed559bff5ae36bd2184d4216973cf', 0, '2099-12-31 23:59:59',
-- ----------------------------
-- Table structure for user_role
-- ----------------------------
-DROP TABLE IF EXISTS `user_role`;
-CREATE TABLE `user_role`
+CREATE TABLE IF NOT EXISTS `user_role`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`user_name` varchar(256) NOT NULL COMMENT 'username rtx',
@@ -774,8 +745,7 @@ CREATE TABLE `user_role`
-- ----------------------------
-- Table structure for workflow_approver
-- ----------------------------
-DROP TABLE IF EXISTS `workflow_approver`;
-CREATE TABLE `workflow_approver`
+CREATE TABLE IF NOT EXISTS `workflow_approver`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`process_name` varchar(256) NOT NULL COMMENT 'Process name',
@@ -805,8 +775,7 @@ VALUES ('NEW_CONSUMPTION_PROCESS', 'ut_admin', 'DEFAULT', NULL, 'admin',
-- ----------------------------
-- Table structure for workflow_event_log
-- ----------------------------
-DROP TABLE IF EXISTS `workflow_event_log`;
-CREATE TABLE `workflow_event_log`
+CREATE TABLE IF NOT EXISTS `workflow_event_log`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`process_id` int(11) NOT NULL,
@@ -833,8 +802,7 @@ CREATE TABLE `workflow_event_log`
-- ----------------------------
-- Table structure for workflow_process
-- ----------------------------
-DROP TABLE IF EXISTS `workflow_process`;
-CREATE TABLE `workflow_process`
+CREATE TABLE IF NOT EXISTS `workflow_process`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`name` varchar(256) NOT NULL COMMENT 'process name',
@@ -856,8 +824,7 @@ CREATE TABLE `workflow_process`
-- ----------------------------
-- Table structure for workflow_task
-- ----------------------------
-DROP TABLE IF EXISTS `workflow_task`;
-CREATE TABLE `workflow_task`
+CREATE TABLE IF NOT EXISTS `workflow_task`
(
`id` int(11) NOT NULL AUTO_INCREMENT,
`type` varchar(64) NOT NULL COMMENT 'Task type: UserTask / ServiceTask',
@@ -882,8 +849,7 @@ CREATE TABLE `workflow_task`
-- ----------------------------
-- Table structure for cluster_set
-- ----------------------------
-DROP TABLE IF EXISTS `cluster_set`;
-CREATE TABLE `cluster_set`
+CREATE TABLE IF NOT EXISTS `cluster_set`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`set_name` varchar(128) NOT NULL COMMENT 'ClusterSet name, English, numbers and underscore',
@@ -906,8 +872,7 @@ CREATE TABLE `cluster_set`
-- ----------------------------
-- Table structure for cluster_set_inlongid
-- ----------------------------
-DROP TABLE IF EXISTS `cluster_set_inlongid`;
-CREATE TABLE `cluster_set_inlongid`
+CREATE TABLE IF NOT EXISTS `cluster_set_inlongid`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`set_name` varchar(256) NOT NULL COMMENT 'ClusterSet name, English, numbers and underscore',
@@ -920,8 +885,7 @@ CREATE TABLE `cluster_set_inlongid`
-- ----------------------------
-- Table structure for cache_cluster
-- ----------------------------
-DROP TABLE IF EXISTS `cache_cluster`;
-CREATE TABLE `cache_cluster`
+CREATE TABLE IF NOT EXISTS `cache_cluster`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`cluster_name` varchar(128) NOT NULL COMMENT 'CacheCluster name, English, numbers and underscore',
@@ -935,8 +899,7 @@ CREATE TABLE `cache_cluster`
-- ----------------------------
-- Table structure for cache_cluster_ext
-- ----------------------------
-DROP TABLE IF EXISTS `cache_cluster_ext`;
-CREATE TABLE `cache_cluster_ext`
+CREATE TABLE IF NOT EXISTS `cache_cluster_ext`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`cluster_name` varchar(128) NOT NULL COMMENT 'CacheCluster name, English, numbers and underscore',
@@ -952,8 +915,7 @@ CREATE TABLE `cache_cluster_ext`
-- ----------------------------
-- Table structure for cache_topic
-- ----------------------------
-DROP TABLE IF EXISTS `cache_topic`;
-CREATE TABLE `cache_topic`
+CREATE TABLE IF NOT EXISTS `cache_topic`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`topic_name` varchar(128) NOT NULL COMMENT 'Topic name, English, numbers and underscore',
@@ -967,8 +929,7 @@ CREATE TABLE `cache_topic`
-- ----------------------------
-- Table structure for proxy_cluster
-- ----------------------------
-DROP TABLE IF EXISTS `proxy_cluster`;
-CREATE TABLE `proxy_cluster`
+CREATE TABLE IF NOT EXISTS `proxy_cluster`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`cluster_name` varchar(128) NOT NULL COMMENT 'ProxyCluster name, English, numbers and underscore',
@@ -982,8 +943,7 @@ CREATE TABLE `proxy_cluster`
-- ----------------------------
-- Table structure for proxy_cluster_to_cache_cluster
-- ----------------------------
-DROP TABLE IF EXISTS `proxy_cluster_to_cache_cluster`;
-CREATE TABLE `proxy_cluster_to_cache_cluster`
+CREATE TABLE IF NOT EXISTS `proxy_cluster_to_cache_cluster`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`proxy_cluster_name` varchar(128) NOT NULL COMMENT 'ProxyCluster name, English, numbers and underscore',
@@ -996,8 +956,7 @@ CREATE TABLE `proxy_cluster_to_cache_cluster`
-- ----------------------------
-- Table structure for flume_source
-- ----------------------------
-DROP TABLE IF EXISTS `flume_source`;
-CREATE TABLE `flume_source`
+CREATE TABLE IF NOT EXISTS `flume_source`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`source_name` varchar(128) NOT NULL COMMENT 'FlumeSource name, English, numbers and underscore',
@@ -1013,8 +972,7 @@ CREATE TABLE `flume_source`
-- ----------------------------
-- Table structure for flume_source_ext
-- ----------------------------
-DROP TABLE IF EXISTS `flume_source_ext`;
-CREATE TABLE `flume_source_ext`
+CREATE TABLE IF NOT EXISTS `flume_source_ext`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`parent_name` varchar(128) NOT NULL COMMENT 'FlumeSource name, English, numbers and underscore',
@@ -1031,8 +989,7 @@ CREATE TABLE `flume_source_ext`
-- ----------------------------
-- Table structure for flume_channel
-- ----------------------------
-DROP TABLE IF EXISTS `flume_channel`;
-CREATE TABLE `flume_channel`
+CREATE TABLE IF NOT EXISTS `flume_channel`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`channel_name` varchar(128) NOT NULL COMMENT 'FlumeChannel name, English, numbers and underscore',
@@ -1046,8 +1003,7 @@ CREATE TABLE `flume_channel`
-- ----------------------------
-- Table structure for flume_channel_ext
-- ----------------------------
-DROP TABLE IF EXISTS `flume_channel_ext`;
-CREATE TABLE `flume_channel_ext`
+CREATE TABLE IF NOT EXISTS `flume_channel_ext`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`parent_name` varchar(128) NOT NULL COMMENT 'FlumeChannel name, English, numbers and underscore',
@@ -1064,8 +1020,7 @@ CREATE TABLE `flume_channel_ext`
-- ----------------------------
-- Table structure for flume_sink
-- ----------------------------
-DROP TABLE IF EXISTS `flume_sink`;
-CREATE TABLE `flume_sink`
+CREATE TABLE IF NOT EXISTS `flume_sink`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`sink_name` varchar(128) NOT NULL COMMENT 'FlumeSink name, English, numbers and underscore',
@@ -1080,8 +1035,7 @@ CREATE TABLE `flume_sink`
-- ----------------------------
-- Table structure for flume_sink_ext
-- ----------------------------
-DROP TABLE IF EXISTS `flume_sink_ext`;
-CREATE TABLE `flume_sink_ext`
+CREATE TABLE IF NOT EXISTS `flume_sink_ext`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`parent_name` varchar(128) NOT NULL COMMENT 'FlumeSink name, English, numbers and underscore',
@@ -1098,8 +1052,7 @@ CREATE TABLE `flume_sink_ext`
-- ----------------------------
-- Table structure for db_collector_detail_task
-- ----------------------------
-DROP TABLE IF EXISTS `db_collector_detail_task`;
-CREATE TABLE `db_collector_detail_task`
+CREATE TABLE IF NOT EXISTS `db_collector_detail_task`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`main_id` varchar(128) NOT NULL COMMENT 'main task id',
@@ -1129,8 +1082,7 @@ CREATE TABLE `db_collector_detail_task`
-- ----------------------------
-- Table structure for sort_cluster_config
-- ----------------------------
-DROP TABLE IF EXISTS `sort_cluster_config`;
-CREATE TABLE `sort_cluster_config`
+CREATE TABLE IF NOT EXISTS `sort_cluster_config`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`cluster_name` varchar(128) NOT NULL COMMENT 'Cluster name',
@@ -1144,8 +1096,7 @@ CREATE TABLE `sort_cluster_config`
-- ----------------------------
-- Table structure for sort_task_id_param
-- ----------------------------
-DROP TABLE IF EXISTS `sort_task_id_param`;
-CREATE TABLE `sort_task_id_param`
+CREATE TABLE IF NOT EXISTS `sort_task_id_param`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`task_name` varchar(128) NOT NULL COMMENT 'Task name',
@@ -1161,8 +1112,7 @@ CREATE TABLE `sort_task_id_param`
-- ----------------------------
-- Table structure for sort_task_sink_param
-- ----------------------------
-DROP TABLE IF EXISTS `sort_task_sink_param`;
-CREATE TABLE `sort_task_sink_param`
+CREATE TABLE IF NOT EXISTS `sort_task_sink_param`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`task_name` varchar(128) NOT NULL COMMENT 'Task name',
@@ -1177,15 +1127,14 @@ CREATE TABLE `sort_task_sink_param`
-- ----------------------------
-- Table structure for sort_source_config
-- ----------------------------
-DROP TABLE IF EXISTS `sort_source_config`;
-CREATE TABLE `sort_source_config`
+CREATE TABLE IF NOT EXISTS `sort_source_config`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`cluster_name` varchar(128) NOT NULL COMMENT 'Cluster name',
`task_name` varchar(128) NOT NULL COMMENT 'Task name',
`zone_name` varchar(128) NOT NULL COMMENT 'Cache zone name',
`topic` varchar(128) DEFAULT '' COMMENT 'Topic',
- `ext_params` text DEFAULT NULL COMMENT 'Another fields, will saved as JSON type',
+ `ext_params` text DEFAULT NULL COMMENT 'Another fields, will be saved as JSON type',
PRIMARY KEY (`id`),
KEY `index_sort_source_config` (`cluster_name`, `task_name`)
) ENGINE = InnoDB
@@ -1194,8 +1143,7 @@ CREATE TABLE `sort_source_config`
-- ----------------------------
-- Table structure for config log report
-- ----------------------------
-DROP TABLE IF EXISTS `stream_config_log`;
-CREATE TABLE `stream_config_log`
+CREATE TABLE IF NOT EXISTS `stream_config_log`
(
`ip` varchar(24) NOT NULL COMMENT 'client host ip',
`version` varchar(64) DEFAULT NULL COMMENT 'client version',
@@ -1214,8 +1162,7 @@ CREATE TABLE `stream_config_log`
-- ----------------------------
-- Table structure for inlong component heartbeat
-- ----------------------------
-DROP TABLE IF EXISTS `component_heartbeat`;
-CREATE TABLE `component_heartbeat`
+CREATE TABLE IF NOT EXISTS `component_heartbeat`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`component` varchar(64) NOT NULL DEFAULT '' COMMENT 'Component name, such as: Agent, Sort...',
@@ -1233,8 +1180,7 @@ CREATE TABLE `component_heartbeat`
-- ----------------------------
-- Table structure for inlong group heartbeat
-- ----------------------------
-DROP TABLE IF EXISTS `group_heartbeat`;
-CREATE TABLE `group_heartbeat`
+CREATE TABLE IF NOT EXISTS `group_heartbeat`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`component` varchar(64) NOT NULL DEFAULT '' COMMENT 'Component name, such as: Agent, Sort...',
@@ -1253,8 +1199,7 @@ CREATE TABLE `group_heartbeat`
-- ----------------------------
-- Table structure for inlong stream heartbeat
-- ----------------------------
-DROP TABLE IF EXISTS `stream_heartbeat`;
-CREATE TABLE `stream_heartbeat`
+CREATE TABLE IF NOT EXISTS `stream_heartbeat`
(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`component` varchar(64) NOT NULL DEFAULT '' COMMENT 'Component name, such as: Agent, Sort...',