[INLONG-10208][Sort] ClsSink support unified configuration

inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/SortClusterConfig.java

@@ -17,6 +17,8 @@
 
 package org.apache.inlong.common.pojo.sort;
 
+import lombok.AllArgsConstructor;
+import lombok.NoArgsConstructor;
 import org.apache.inlong.common.pojo.sort.dataflow.DataFlowConfig;
 import org.apache.inlong.common.pojo.sort.mq.MqClusterConfig;
 
@@ -28,6 +30,8 @@
 
 @Data
 @Builder
+@AllArgsConstructor
+@NoArgsConstructor
 public class SortClusterConfig implements Serializable {
 
     private String clusterTag;

inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/SortTaskConfig.java

@@ -17,6 +17,8 @@
 
 package org.apache.inlong.common.pojo.sort;
 
+import lombok.AllArgsConstructor;
+import lombok.NoArgsConstructor;
 import org.apache.inlong.common.pojo.sort.node.NodeConfig;
 
 import lombok.Builder;
@@ -27,6 +29,8 @@
 
 @Data
 @Builder
+@AllArgsConstructor
+@NoArgsConstructor
 public class SortTaskConfig implements Serializable {
 
     private String sortTaskName;

inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/DataFlowConfig.java

@@ -17,6 +17,8 @@
 
 package org.apache.inlong.common.pojo.sort.dataflow;
 
+import lombok.AllArgsConstructor;
+import lombok.NoArgsConstructor;
 import org.apache.inlong.common.pojo.sort.dataflow.sink.SinkConfig;
 
 import lombok.Builder;
@@ -27,10 +29,14 @@
 
 @Data
 @Builder
+@AllArgsConstructor
+@NoArgsConstructor
 public class DataFlowConfig implements Serializable {
 
     private String dataflowId;
     private Integer version;
+    private String inlongGroupId;
+    private String inlongStreamId;
     private SourceConfig sourceConfig;
     private SinkConfig sinkConfig;
     private Map<String, String> properties;

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sort/DefaultSortConfigOperator.java

@@ -161,6 +161,8 @@ private DataFlowConfig getDataFlowConfig(InlongGroupInfo groupInfo, InlongStream
                 .dataflowId(String.valueOf(sink.getId()))
                 .sourceConfig(getSourceConfig(groupInfo, streamInfo, sink))
                 .sinkConfig(getSinkConfig(sink))
+                .inlongGroupId(groupInfo.getInlongGroupId())
+                .inlongStreamId(streamInfo.getInlongStreamId())
                 .build();
     }
 

inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/SinkContext.java

@@ -41,6 +41,7 @@
  * 
  * SinkContext
  */
+@Deprecated
 public class SinkContext {
 
     public static final Logger LOG = InlongLoggerFactory.getLogger(SinkContext.class);

inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/cls/ClsIdConfig.java

@@ -17,16 +17,26 @@
 
 package org.apache.inlong.sort.standalone.sink.cls;
 
+import org.apache.inlong.common.pojo.sort.dataflow.DataFlowConfig;
+import org.apache.inlong.common.pojo.sort.dataflow.field.FieldConfig;
+import org.apache.inlong.common.pojo.sort.dataflow.sink.ClsSinkConfig;
+import org.apache.inlong.common.pojo.sort.node.ClsNodeConfig;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
 import lombok.Data;
+import lombok.NoArgsConstructor;
 
-import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
+import java.util.stream.Collectors;
 
 /**
  * Cls config of each uid.
  */
 @Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
 public class ClsIdConfig {
 
     private String inlongGroupId;
@@ -36,23 +46,28 @@ public class ClsIdConfig {
     private String secretId;
     private String secretKey;
     private String topicId;
-    private String fieldNames;
+    private List<String> fieldList;
     private int fieldOffset = 2;
     private int contentOffset = 0;
-    private List<String> fieldList;
 
-    /**
-     * Parse fieldNames to list of fields.
-     * @return List of fields.
-     */
-    public List<String> getFieldList() {
-        if (fieldList == null) {
-            this.fieldList = new ArrayList<>();
-            if (fieldNames != null) {
-                String[] fieldNameArray = fieldNames.split("\\s+");
-                this.fieldList.addAll(Arrays.asList(fieldNameArray));
-            }
-        }
-        return fieldList;
+    public static ClsIdConfig create(DataFlowConfig dataFlowConfig, ClsNodeConfig nodeConfig) {
+        ClsSinkConfig sinkConfig = (ClsSinkConfig) dataFlowConfig.getSinkConfig();
+        List<String> fields = sinkConfig.getFieldConfigs()
+                .stream()
+                .map(FieldConfig::getName)
+                .collect(Collectors.toList());
+        return ClsIdConfig.builder()
+                .inlongGroupId(dataFlowConfig.getInlongGroupId())
+                .inlongStreamId(dataFlowConfig.getInlongStreamId())
+                .contentOffset(sinkConfig.getContentOffset())
+                .fieldOffset(sinkConfig.getFieldOffset())
+                .separator(sinkConfig.getSeparator())
+                .fieldList(fields)
+                .topicId(sinkConfig.getTopicId())
+                .endpoint(nodeConfig.getEndpoint())
+                .secretId(nodeConfig.getSendSecretId())
+                .secretKey(nodeConfig.getSendSecretKey())
+                .build();
     }
+
 }

inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/cls/ClsSink.java

@@ -27,13 +27,6 @@
 import java.util.ArrayList;
 import java.util.List;
 
-/**
- * Cls Sink implementation.
- *
- * <p>
- *     Response for initialization of {@link ClsChannelWorker}.
- * </p>
- */
 public class ClsSink extends AbstractSink implements Configurable {
 
     private static final Logger LOG = LoggerFactory.getLogger(ClsSink.class);
@@ -42,9 +35,6 @@ public class ClsSink extends AbstractSink implements Configurable {
     private ClsSinkContext context;
     private List<ClsChannelWorker> workers = new ArrayList<>();
 
-    /**
-     * Start {@link ClsChannelWorker}.
-     */
     @Override
     public void start() {
         super.start();
@@ -57,13 +47,10 @@ public void start() {
                 worker.start();
             }
         } catch (Exception e) {
-            LOG.error(e.getMessage(), e);
+            LOG.error("failed to start cls sink", e);
         }
     }
 
-    /**
-     * Stop {@link ClsChannelWorker}.
-     */
     @Override
     public void stop() {
         super.stop();
@@ -74,24 +61,15 @@ public void stop() {
             }
             this.workers.clear();
         } catch (Exception e) {
-            LOG.error(e.getMessage(), e);
+            LOG.error("failed to stop cls sink", e);
         }
     }
 
-    /**
-     * Process.
-     * @return Status
-     * @throws EventDeliveryException
-     */
     @Override
     public Status process() throws EventDeliveryException {
         return Status.BACKOFF;
     }
 
-    /**
-     * Config parent context.
-     * @param context Parent context.
-     */
     @Override
     public void configure(Context context) {
         LOG.info("start to configure:{}, context:{}.", this.getName(), context.toString());