[INLONG-9770][Manager] Unified compression type configuration

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

@@ -0,0 +1,72 @@
+/*
+ * 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.common.enums;
+
+import java.util.Objects;
+
+public enum InlongCompressType {
+
+    NONE(0, "NONE", "The message compressed with nothing"),
+    INLONG_GZ(1, "INLONG_GZ", "The message compressed with inlong gz"),
+    INLONG_SNAPPY(2, "INLONG_SNAPPY", "The message compressed with inlong snappy"),
+    UNKNOWN(99, "UNKNOWN", "Unknown compress type");
+
+    private final int id;
+    private final String name;
+    private final String desc;
+
+    InlongCompressType(int id, String name, String desc) {
+        this.id = id;
+        this.name = name;
+        this.desc = desc;
+    }
+
+    public static InlongCompressType valueOf(int value) {
+        for (InlongCompressType msgCompressType : InlongCompressType.values()) {
+            if (msgCompressType.getId() == value) {
+                return msgCompressType;
+            }
+        }
+        return UNKNOWN;
+    }
+
+    public static InlongCompressType forType(String type) {
+        for (InlongCompressType msgCompressType : InlongCompressType.values()) {
+            if (Objects.equals(msgCompressType.getName(), type)) {
+                return msgCompressType;
+            }
+        }
+        return UNKNOWN;
+    }
+
+    public int getId() {
+        return id;
+    }
+
+    public String getStrId() {
+        return String.valueOf(id);
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public String getDesc() {
+        return desc;
+    }
+}

inlong-manager/manager-dao/src/main/resources/mappers/ClusterSetMapper.xml

@@ -49,6 +49,7 @@
         select inlong_group_id,
                inlong_stream_id,
                data_type,
+               wrap_type,
                mq_resource as topic,
                ext_params
         from inlong_stream

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/dataproxy/InlongStreamId.java

@@ -26,16 +26,25 @@ public class InlongStreamId {
     private String inlongStreamId;
     private String topic;
     private String dataType;
+    private String wrapType;
     private String extParams;
 
     public String getDataType() {
         return dataType;
     }
 
+    public String getWrapType() {
+        return wrapType;
+    }
+
     public void setDataType(String dataType) {
         this.dataType = dataType;
     }
 
+    public void setWrapType(String wrapType) {
+        this.wrapType = wrapType;
+    }
+
     /**
      * get inlongGroupId
      *

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/stream/InlongStreamBriefInfo.java

@@ -85,6 +85,9 @@ public class InlongStreamBriefInfo {
     @ApiModelProperty(value = "The message body  wrap type, including: RAW, INLONG_MSG_V0, INLONG_MSG_V1, etc")
     private String wrapType;
 
+    @ApiModelProperty(value = "The compression type used for dataproxy and sort side data transmission to reduce the network IO overhead")
+    private String inlongCompressType;
+
     @ApiModelProperty(value = "Whether to ignore the parse errors of field value")
     private Boolean ignoreParseError;
 

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/stream/InlongStreamExtParam.java

@@ -48,6 +48,9 @@ public class InlongStreamExtParam implements Serializable {
     @ApiModelProperty(value = "If use extended fields")
     private Boolean useExtendedFields = false;
 
+    @ApiModelProperty(value = "The compression type used for dataproxy and sort side data transmission to reduce the network IO overhead")
+    private String inlongCompressType = "NONE";
+
     @ApiModelProperty(value = "Predefined fields")
     private String predefinedFields;
 

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/stream/InlongStreamInfo.java

@@ -133,6 +133,9 @@ public class InlongStreamInfo extends BaseInlongStream {
     @ApiModelProperty(value = "The message body wrap type, including: RAW, INLONG_MSG_V0, INLONG_MSG_V1, etc")
     private String wrapType;
 
+    @ApiModelProperty(value = "The compression type used for dataproxy and sort side data transmission to reduce the network IO overhead")
+    private String inlongCompressType;
+
     @ApiModelProperty(value = "If use extended fields")
     private Boolean useExtendedFields = false;
 

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/stream/InlongStreamRequest.java

@@ -121,6 +121,9 @@ public class InlongStreamRequest extends BaseInlongStream {
     @ApiModelProperty(value = "Whether to ignore the parse errors of field value")
     private boolean ignoreParseError = true;
 
+    @ApiModelProperty(value = "The compression type used for dataproxy and sort side data transmission to reduce the network IO overhead")
+    private String inlongCompressType;
+
     @ApiModelProperty(value = "If use extended fields")
     private Boolean useExtendedFields = false;
 

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/repository/DataProxyConfigRepository.java

@@ -92,6 +92,7 @@ public class DataProxyConfigRepository implements IRepository {
     public static final String KEY_NEW_TENANT_KEY = "pulsarTenant";
     public static final String KEY_OLD_TENANT_KEY = "tenant";
     public static final String KEY_DATA_TYPE = "dataType";
+    public static final String KEY_WRAP_TYPE = "wrapType";
     public static final String KEY_BACKUP_CLUSTER_TAG = "backup_cluster_tag";
     public static final String KEY_BACKUP_TOPIC = "backup_topic";
     public static final String KEY_SORT_TASK_NAME = "defaultSortTaskName";
@@ -376,6 +377,7 @@ private void reloadInlongId(Map<String, DataProxyCluster> proxyClusterMap) {
         streamIdMap.forEach((k, v) -> {
             Map<String, String> params = fromJsonToMap(v.getExtParams());
             params.computeIfAbsent(KEY_DATA_TYPE, type -> v.getDataType());
+            params.computeIfAbsent(KEY_WRAP_TYPE, type -> v.getWrapType());
             streamParams.put(k, params);
         });
         // reload inlong stream ext