[INLONG-3719][Manager] Support data_transformation feature in Inlong (#3774)

* Add constructor to streamTransform

* Add unit tests for stream transform

* Update API doc

* Add processing_time for stream

* resolve circular dependency in manager service

* Update SQL file format

Co-authored-by: healchow <healchow@gmail.com>

Author: kipshi

inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/AutoPush2HiveExample.java

@@ -17,7 +17,7 @@
 
 package org.apache.inlong.manager.client;
 
-import org.apache.commons.compress.utils.Lists;
+import com.google.common.collect.Lists;
 import org.apache.inlong.manager.client.api.ClientConfiguration;
 import org.apache.inlong.manager.client.api.DataSeparator;
 import org.apache.inlong.manager.client.api.FlinkSortBaseConf;
@@ -29,7 +29,7 @@
 import org.apache.inlong.manager.client.api.InlongStreamConf;
 import org.apache.inlong.manager.client.api.PulsarBaseConf;
 import org.apache.inlong.manager.client.api.SinkField;
-import org.apache.inlong.manager.client.api.StreamField;
+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;

inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/File2HiveExample.java

@@ -17,7 +17,7 @@
 
 package org.apache.inlong.manager.client;
 
-import org.apache.commons.compress.utils.Lists;
+import com.google.common.collect.Lists;
 import org.apache.inlong.manager.client.api.ClientConfiguration;
 import org.apache.inlong.manager.client.api.DataSeparator;
 import org.apache.inlong.manager.client.api.FlinkSortBaseConf;
@@ -29,7 +29,7 @@
 import org.apache.inlong.manager.client.api.InlongStreamConf;
 import org.apache.inlong.manager.client.api.PulsarBaseConf;
 import org.apache.inlong.manager.client.api.SinkField;
-import org.apache.inlong.manager.client.api.StreamField;
+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;

inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/Kafka2HiveExample.java

@@ -17,8 +17,8 @@
 
 package org.apache.inlong.manager.client;
 
+import com.google.common.collect.Lists;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.compress.utils.Lists;
 import org.apache.inlong.manager.client.api.ClientConfiguration;
 import org.apache.inlong.manager.client.api.DataFormat;
 import org.apache.inlong.manager.client.api.DataSeparator;
@@ -31,7 +31,7 @@
 import org.apache.inlong.manager.client.api.InlongStreamConf;
 import org.apache.inlong.manager.client.api.PulsarBaseConf;
 import org.apache.inlong.manager.client.api.SinkField;
-import org.apache.inlong.manager.client.api.StreamField;
+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;

inlong-manager/manager-client-tools/src/main/java/org/apache/inlong/manager/client/cli/CreateGroupConf.java

@@ -20,7 +20,7 @@
 import lombok.Data;
 import org.apache.inlong.manager.client.api.InlongGroupConf;
 import org.apache.inlong.manager.client.api.InlongStreamConf;
-import org.apache.inlong.manager.client.api.StreamField;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
 import org.apache.inlong.manager.client.api.StreamSink;
 import org.apache.inlong.manager.client.api.StreamSource;
 

inlong-manager/manager-client/pom.xml

@@ -32,6 +32,12 @@
             <groupId>org.apache.inlong</groupId>
             <artifactId>manager-common</artifactId>
             <version>${project.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.commons</groupId>
+                    <artifactId>commons-compress</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
         <dependency>
             <groupId>org.projectlombok</groupId>

inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongGroup.java

@@ -53,12 +53,22 @@ public interface InlongGroup {
      */
     void update(InlongGroupConf conf) throws Exception;
 
+    /**
+     * ReInit inlong group after update configuration for group.
+     * Must be invoked when group is rejected,failed or started
+     *
+     * @return inlong group info
+     */
+    InlongGroupContext reInitOnUpdate(InlongGroupConf conf) throws Exception;
+
     /**
      * Init inlong group on updated conf.
      * Must be invoked when group is rejected,failed or started
+     * This method is deprecated, recommend to use reInitOnUpdate
      *
      * @return inlong group info
      */
+    @Deprecated
     InlongGroupContext initOnUpdate(InlongGroupConf conf) throws Exception;
 
     /**

inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongGroupConf.java

@@ -59,6 +59,9 @@ public class InlongGroupConf {
     @ApiModelProperty("Need zookeeper support")
     private boolean zookeeperEnabled = true;
 
-    @ApiModelProperty("data proxy cluster id")
+    @ApiModelProperty("Data proxy cluster id")
     private Integer proxyClusterId;
+
+    @ApiModelProperty("Use light weight group")
+    private boolean lightWeight = false;
 }

inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongStream.java

@@ -17,6 +17,9 @@
 
 package org.apache.inlong.manager.client.api;
 
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
+import org.apache.inlong.manager.common.pojo.stream.StreamPipeline;
+
 import java.util.List;
 import java.util.Map;
 
@@ -30,11 +33,19 @@ public abstract class InlongStream {
 
     public abstract Map<String, StreamSink> getSinks();
 
+    public abstract Map<String, StreamTransform> getTransforms();
+
     public abstract void addSource(StreamSource source);
 
     public abstract void addSink(StreamSink sink);
 
+    public abstract void addTransform(StreamTransform transform);
+
+    public abstract StreamPipeline createPipeline();
+
+    @Deprecated
     public abstract void updateSource(StreamSource source);
 
+    @Deprecated
     public abstract void updateSink(StreamSink sink);
 }

inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongStreamBuilder.java

@@ -17,6 +17,8 @@
 
 package org.apache.inlong.manager.client.api;
 
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
+
 import java.util.List;
 
 public abstract class InlongStreamBuilder {
@@ -43,6 +45,13 @@ public abstract class InlongStreamBuilder {
      */
     public abstract InlongStreamBuilder fields(List<StreamField> fieldList);
 
+    /**
+     * Create stream transform
+     *
+     * @return inlong stream builder
+     */
+    public abstract InlongStreamBuilder transform(StreamTransform streamTransform);
+
     /**
      * Create data stream by builder
      *

inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongStreamConf.java

@@ -22,6 +22,7 @@
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
 
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;

inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/SinkField.java

@@ -23,6 +23,7 @@
 import lombok.EqualsAndHashCode;
 import lombok.NoArgsConstructor;
 import org.apache.inlong.manager.common.enums.FieldType;
+import org.apache.inlong.manager.common.pojo.stream.StreamField;
 
 @Data
 @EqualsAndHashCode(callSuper = true)

inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/StreamSink.java

@@ -21,13 +21,14 @@
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 import org.apache.inlong.manager.common.enums.SinkType;
+import org.apache.inlong.manager.common.pojo.stream.StreamNode;
 
 import java.util.List;
 import java.util.Map;
 
 @Data
 @ApiModel("Stream sink configuration")
-public abstract class StreamSink {
+public abstract class StreamSink extends StreamNode {
 
     @ApiModelProperty(value = "DataSink name", required = true)
     private String sinkName;

inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/StreamSource.java

@@ -22,10 +22,11 @@
 import lombok.Data;
 import org.apache.inlong.manager.common.enums.SourceState;
 import org.apache.inlong.manager.common.enums.SourceType;
+import org.apache.inlong.manager.common.pojo.stream.StreamNode;
 
 @Data
 @ApiModel("Stream source configuration")
-public abstract class StreamSource {
+public abstract class StreamSource extends StreamNode {
 
     public enum State {
         INIT, NORMAL, FROZING, FROZEN, FAILED, DELETING, DELETE;

inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/StreamTransform.java

@@ -0,0 +1,35 @@
+/*
+ * 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.client.api;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+import org.apache.inlong.manager.common.pojo.stream.StreamNode;
+import org.apache.inlong.manager.common.pojo.transform.TransformDefinition;
+
+@Data
+@ApiModel("Stream Transform configuration")
+public abstract class StreamTransform extends StreamNode {
+
+    @ApiModelProperty(value = "Transform name", required = true)
+    protected String transformName;
+
+    @ApiModelProperty(value = "Transform name", required = true)
+    protected TransformDefinition transformDefinition;
+}

inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/BlankInlongGroup.java

@@ -48,6 +48,11 @@ public void update(InlongGroupConf conf) throws Exception {
         throw new UnsupportedOperationException("Inlong group is not exists");
     }
 
+    @Override
+    public InlongGroupContext reInitOnUpdate(InlongGroupConf conf) throws Exception {
+        throw new UnsupportedOperationException("Inlong group is not exists");
+    }
+
     @Override
     public InlongGroupContext initOnUpdate(InlongGroupConf conf) throws Exception {
         throw new UnsupportedOperationException("Inlong group is not exists");