[INLONG-9559][Manager] Add Dameng Extract Node to Inlong Manager

inlong-common/src/main/java/org/apache/inlong/common/enums/TaskTypeEnum.java

@@ -36,6 +36,7 @@ public enum TaskTypeEnum {
     REDIS(11),
     MQTT(12),
     HUDI(13),
+    DAMENG(14),
 
     // only used for unit test
     MOCK(201)

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/DataNodeType.java

@@ -42,4 +42,6 @@ public class DataNodeType {
      */
     public static final String CLS = "CLS";
     public static final String PULSAR = "PULSAR";
+
+    public static final String DAMENG = "DAMENG";
 }

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/SourceType.java

@@ -37,6 +37,8 @@ public class SourceType extends StreamType {
     public static final String REDIS = "REDIS";
     public static final String MQTT = "MQTT";
 
+    public static final String DAMENG = "DAMENG";
+
     public static final Map<String, TaskTypeEnum> SOURCE_TASK_MAP = new HashMap<String, TaskTypeEnum>() {
 
         {
@@ -55,6 +57,7 @@ public class SourceType extends StreamType {
             put(REDIS, TaskTypeEnum.REDIS);
             put(MQTT, TaskTypeEnum.MQTT);
             put(HUDI, TaskTypeEnum.HUDI);
+            put(DAMENG, TaskTypeEnum.DAMENG);
 
         }
     };

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/fieldtype/strategy/DamengFieldTypeStrategy.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.common.fieldtype.strategy;
+
+import org.apache.inlong.manager.common.consts.DataNodeType;
+import org.apache.inlong.manager.common.fieldtype.FieldTypeMappingReader;
+
+import org.apache.commons.lang3.StringUtils;
+
+import static org.apache.inlong.manager.common.consts.InlongConstants.LEFT_BRACKET;
+
+public class DamengFieldTypeStrategy implements FieldTypeMappingStrategy {
+
+    private final FieldTypeMappingReader reader;
+
+    public DamengFieldTypeStrategy() {
+        this.reader = new FieldTypeMappingReader(DataNodeType.DAMENG);
+    }
+
+    @Override
+    public String getFieldTypeMapping(String sourceType) {
+        String dataType = StringUtils.substringBefore(sourceType, LEFT_BRACKET).toUpperCase();
+        return reader.getFIELD_TYPE_MAPPING_MAP().getOrDefault(dataType, sourceType.toUpperCase());
+    }
+}

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/node/ExtractNodeProviderFactory.java

@@ -20,6 +20,7 @@
 import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
 import org.apache.inlong.manager.common.exceptions.BusinessException;
 import org.apache.inlong.manager.pojo.sort.node.base.ExtractNodeProvider;
+import org.apache.inlong.manager.pojo.sort.node.provider.DMProvider;
 import org.apache.inlong.manager.pojo.sort.node.provider.HudiProvider;
 import org.apache.inlong.manager.pojo.sort.node.provider.IcebergProvider;
 import org.apache.inlong.manager.pojo.sort.node.provider.KafkaProvider;
@@ -58,7 +59,7 @@ public class ExtractNodeProviderFactory {
         EXTRACT_NODE_PROVIDER_LIST.add(new PostgreSQLProvider());
         EXTRACT_NODE_PROVIDER_LIST.add(new MySQLBinlogProvider());
         EXTRACT_NODE_PROVIDER_LIST.add(new IcebergProvider());
-
+        EXTRACT_NODE_PROVIDER_LIST.add(new DMProvider());
     }
 
     /**

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/node/provider/DMProvider.java

@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.pojo.sort.node.provider;
+
+import org.apache.inlong.manager.common.consts.SourceType;
+import org.apache.inlong.manager.common.fieldtype.strategy.DamengFieldTypeStrategy;
+import org.apache.inlong.manager.common.fieldtype.strategy.FieldTypeMappingStrategy;
+import org.apache.inlong.manager.pojo.sort.node.base.ExtractNodeProvider;
+import org.apache.inlong.manager.pojo.source.dameng.DamengSource;
+import org.apache.inlong.manager.pojo.stream.StreamNode;
+import org.apache.inlong.sort.protocol.FieldInfo;
+import org.apache.inlong.sort.protocol.constant.DMConstant;
+import org.apache.inlong.sort.protocol.node.ExtractNode;
+import org.apache.inlong.sort.protocol.node.extract.DamengExtractNode;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * The Provider for creating MySQLBinlog extract nodes.
+ */
+public class DMProvider implements ExtractNodeProvider {
+
+    private static final FieldTypeMappingStrategy FIELD_TYPE_MAPPING_STRATEGY = new DamengFieldTypeStrategy();
+
+    @Override
+    public Boolean accept(String sourceType) {
+        return SourceType.DAMENG.equals(sourceType);
+    }
+
+    @Override
+    public ExtractNode createExtractNode(StreamNode streamNodeInfo) {
+        DamengSource source = (DamengSource) streamNodeInfo;
+        List<FieldInfo> fieldInfos = parseStreamFieldInfos(source.getFieldList(), source.getSourceName(),
+                FIELD_TYPE_MAPPING_STRATEGY);
+        DMConstant.ScanStartUpMode scanStartupMode = StringUtils.isBlank(source.getScanStartupMode())
+                ? null
+                : DMConstant.ScanStartUpMode.forName(source.getScanStartupMode());
+        Map<String, String> properties = parseProperties(source.getProperties());
+
+        return new DamengExtractNode(
+                source.getSourceName(),
+                source.getSourceName(),
+                fieldInfos,
+                null,
+                properties,
+                source.getPrimaryKey(),
+                source.getHost(),
+                source.getUsername(),
+                source.getPassword(),
+                source.getDbName(),
+                source.getSchemaName(),
+                source.getTableName(),
+                source.getPort(),
+                scanStartupMode);
+    }
+
+}

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/source/dameng/DamengSource.java

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.pojo.source.dameng;
+
+import org.apache.inlong.manager.common.consts.SourceType;
+import org.apache.inlong.manager.common.util.CommonBeanUtils;
+import org.apache.inlong.manager.common.util.JsonTypeDefine;
+import org.apache.inlong.manager.pojo.source.SourceRequest;
+import org.apache.inlong.manager.pojo.source.StreamSource;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+
+@Data
+@SuperBuilder
+@AllArgsConstructor
+@ToString(callSuper = true)
+@EqualsAndHashCode(callSuper = true)
+@ApiModel(value = "Dameng source info")
+@JsonTypeDefine(value = SourceType.DAMENG)
+public class DamengSource extends StreamSource {
+
+    @ApiModelProperty("The database name")
+    private String dbName;
+
+    @ApiModelProperty("Schema name")
+    private String schemaName;
+
+    @ApiModelProperty("The table name")
+    private String tableName;
+
+    @ApiModelProperty("host")
+    private String host;
+
+    @ApiModelProperty("port")
+    private Integer port;
+
+    @ApiModelProperty("username")
+    private String username;
+
+    @ApiModelProperty("password")
+    private String password;
+
+    @ApiModelProperty("Scan startup mode")
+    private String scanStartupMode;
+
+    @ApiModelProperty("Primary key must be shared by all tables")
+    private String primaryKey;
+
+    @ApiModelProperty("Need transfer total database")
+    @Builder.Default
+    private boolean allMigration = false;
+
+    public DamengSource() {
+        this.setSourceType(SourceType.DAMENG);
+    }
+
+    @Override
+    public SourceRequest genSourceRequest() {
+        return CommonBeanUtils.copyProperties(this, DamengSourceRequest::new);
+    }
+
+}

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/source/dameng/DamengSourceDTO.java

@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.pojo.source.dameng;
+
+import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
+import org.apache.inlong.manager.common.exceptions.BusinessException;
+import org.apache.inlong.manager.common.util.CommonBeanUtils;
+import org.apache.inlong.manager.common.util.JsonUtils;
+
+import io.swagger.annotations.ApiModelProperty;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import javax.validation.constraints.NotNull;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class DamengSourceDTO {
+
+    @ApiModelProperty("The database name")
+    private String dbName;
+
+    @ApiModelProperty("Schema name")
+    private String schemaName;
+
+    @ApiModelProperty("The table name")
+    private String tableName;
+
+    @ApiModelProperty("host")
+    private String host;
+
+    @ApiModelProperty("port")
+    private String port;
+
+    @ApiModelProperty("username")
+    private String username;
+
+    @ApiModelProperty("password")
+    private String password;
+
+    @ApiModelProperty("Scan startup mode")
+    private String scanStartupMode;
+
+    @ApiModelProperty("Primary key must be shared by all tables")
+    private String primaryKey;
+
+    @ApiModelProperty("Need transfer total database")
+    @Builder.Default
+    private boolean allMigration = false;
+
+    /**
+     * Get the dto instance from the request
+     */
+    public static DamengSourceDTO getFromRequest(DamengSourceRequest request) {
+        DamengSourceDTO damengSourceDTO = new DamengSourceDTO();
+        CommonBeanUtils.copyProperties(request, damengSourceDTO);
+        return damengSourceDTO;
+    }
+
+    /**
+     * Get the dto instance from the JSON string
+     */
+    public static DamengSourceDTO getFromJson(@NotNull String extParams) {
+        try {
+            return JsonUtils.parseObject(extParams, DamengSourceDTO.class);
+        } catch (Exception e) {
+            throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT,
+                    String.format("parse extParams of DamengSourceDTO failure: %s", e.getMessage()));
+        }
+    }
+}