[INLONG-4169][Sort] Add comments before classes (apache#4176)

inlong-sort-standalone/sort-standalone-common/src/main/java/org/apache/inlong/sort/standalone/utils/FlumeConfigGenerator.java

@@ -23,6 +23,9 @@
 import java.util.HashMap;
 import java.util.Map;
 
+/**
+ * generator for flume config
+ */
 public class FlumeConfigGenerator {
 
     public static final String KEY_TASK_NAME = "taskName";

inlong-sort/sort-api/src/main/java/org/apache/inlong/sort/ZkTools.java

@@ -39,6 +39,9 @@ public class ZkTools {
 
     private static final Logger LOG = LoggerFactory.getLogger(ZkTools.class);
 
+    /**
+     * add a dataFlow to sort cluster by creating a zk node
+     */
     public static void addDataFlowToCluster(
             String cluster,
             long dataFlowId,
@@ -59,6 +62,9 @@ public static void addDataFlowToCluster(
         LOG.info("Add dataFlow with id {} to cluster {} successfully", dataFlowId, cluster);
     }
 
+    /**
+     * remove a dataFlow from sort cluster
+     */
     public static void removeDataFlowFromCluster(
             String cluster,
             long dataFlowId,

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/configuration/Constants.java

@@ -21,6 +21,9 @@
 
 import static org.apache.inlong.sort.configuration.ConfigOptions.key;
 
+/**
+ * Constants used in sort
+ */
 public class Constants {
 
     public static final long UNKNOWN_DATAFLOW_ID = -1L;

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/configuration/FallbackKey.java

@@ -28,7 +28,6 @@ public class FallbackKey {
     // -------------------------
     //  Factory methods
     // -------------------------
-
     static FallbackKey createFallbackKey(String key) {
         return new FallbackKey(key, false);
     }

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/DataFlowStorageInfo.java

@@ -24,6 +24,10 @@
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
+/**
+ * defines where to get the data flow info from
+ * only supports zookeeper now
+ */
 public class DataFlowStorageInfo implements Serializable {
 
     private static final long serialVersionUID = -2785142086976967367L;

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/deserialization/AvroDeserializationInfo.java

@@ -18,6 +18,9 @@
 
 package org.apache.inlong.sort.protocol.deserialization;
 
+/**
+ * Avro deserialization info
+ */
 public class AvroDeserializationInfo implements DeserializationInfo {
 
     private static final long serialVersionUID = -5344203248610337314L;

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/deserialization/CanalDeserializationInfo.java

@@ -24,6 +24,9 @@
 
 import java.util.Objects;
 
+/**
+ * Canal deserialization info
+ */
 public class CanalDeserializationInfo implements DeserializationInfo {
 
     private static final long serialVersionUID = -5344203248610337314L;

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/deserialization/CsvDeserializationInfo.java

@@ -21,7 +21,7 @@
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 /**
- * .
+ * Csv deserialization info
  */
 public class CsvDeserializationInfo implements DeserializationInfo {
 

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/deserialization/DebeziumDeserializationInfo.java

@@ -23,6 +23,9 @@
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
+/**
+ * Debezium deserialization info
+ */
 public class DebeziumDeserializationInfo implements DeserializationInfo {
 
     private static final long serialVersionUID = 1L;

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/deserialization/JsonDeserializationInfo.java

@@ -18,6 +18,9 @@
 
 package org.apache.inlong.sort.protocol.deserialization;
 
+/**
+ * Json deserialization info
+ */
 public class JsonDeserializationInfo implements DeserializationInfo {
 
     private static final long serialVersionUID = -5344203248610337314L;

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/deserialization/KvDeserializationInfo.java

@@ -23,7 +23,7 @@
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 /**
- * .
+ * Kv deserialization info
  */
 public class KvDeserializationInfo implements DeserializationInfo {
 

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/ExtractNode.java

@@ -35,6 +35,9 @@
 import java.util.List;
 import java.util.Map;
 
+/**
+ * extract node extracts data from external system
+ */
 @JsonTypeInfo(
         use = JsonTypeInfo.Id.NAME,
         include = JsonTypeInfo.As.PROPERTY,

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/LoadNode.java

@@ -38,6 +38,9 @@
 import java.util.List;
 import java.util.Map;
 
+/**
+ * node for inserting data to external system
+ */
 @JsonTypeInfo(
         use = JsonTypeInfo.Id.NAME,
         include = JsonTypeInfo.As.PROPERTY,

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/Node.java

@@ -33,6 +33,9 @@
 import java.util.Map;
 import java.util.TreeMap;
 
+/**
+ * Base class for extract node \ load node \ transform node
+ */
 @JsonTypeInfo(
         use = JsonTypeInfo.Id.NAME,
         include = JsonTypeInfo.As.PROPERTY,

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/MySqlExtractNode.java

@@ -35,6 +35,9 @@
 import org.apache.inlong.sort.protocol.node.ExtractNode;
 import org.apache.inlong.sort.protocol.transformation.WatermarkField;
 
+/**
+ * Mysql extract node using debezium engine
+ */
 @EqualsAndHashCode(callSuper = true)
 @JsonTypeName("mysqlExtract")
 @Data

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/format/Format.java

@@ -24,6 +24,10 @@
 import java.util.HashMap;
 import java.util.Map;
 
+
+/**
+ * Interface class for data format
+ */
 @JsonTypeInfo(
         use = JsonTypeInfo.Id.NAME,
         include = JsonTypeInfo.As.PROPERTY,

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/HiveLoadNode.java

@@ -37,6 +37,9 @@
 import java.util.List;
 import java.util.Map;
 
+/**
+ * hive load node with flink connectors
+ */
 @EqualsAndHashCode(callSuper = true)
 @JsonTypeName("hiveLoad")
 @Data

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/KafkaLoadNode.java

@@ -43,6 +43,9 @@
 import java.util.List;
 import java.util.Map;
 
+/**
+ * Kafka load node using kafka connectors provided by flink
+ */
 @EqualsAndHashCode(callSuper = true)
 @JsonTypeName("kafkaLoad")
 @Data

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/transform/TransformNode.java

@@ -37,6 +37,9 @@
 import java.util.List;
 import java.util.Map;
 
+/**
+ * Base class for transform node, such as a distinct node
+ */
 @JsonTypeInfo(
         use = JsonTypeInfo.Id.NAME,
         include = JsonTypeInfo.As.PROPERTY,

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/serialization/AvroSerializationInfo.java

@@ -17,6 +17,9 @@
 
 package org.apache.inlong.sort.protocol.serialization;
 
+/**
+ * basic info for avro serialization
+ */
 public class AvroSerializationInfo implements SerializationInfo {
 
     private static final long serialVersionUID = 8446721117598285868L;

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/serialization/CanalSerializationInfo.java

@@ -17,6 +17,9 @@
 
 package org.apache.inlong.sort.protocol.serialization;
 
+/**
+ * basic info for canal serialization
+ */
 public class CanalSerializationInfo implements SerializationInfo {
 
     private static final long serialVersionUID = 479443152335788151L;

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/serialization/SerializationInfo.java

@@ -22,6 +22,9 @@
 
 import java.io.Serializable;
 
+/**
+ * interface for serialization infos
+ */
 @JsonTypeInfo(
         use = JsonTypeInfo.Id.NAME,
         include = JsonTypeInfo.As.PROPERTY,

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/CompareOperator.java

@@ -30,6 +30,9 @@
 import org.apache.inlong.sort.protocol.transformation.operator.NotEqualOperator;
 import org.apache.inlong.sort.protocol.transformation.operator.NotInOperator;
 
+/**
+ * interface for all compare operators
+ */
 @JsonTypeInfo(
         use = JsonTypeInfo.Id.NAME,
         include = JsonTypeInfo.As.PROPERTY,

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/FieldMappingRule.java

@@ -27,6 +27,9 @@
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.inlong.sort.protocol.FieldInfo;
 
+/**
+ * defines the mapping rules for data fields
+ */
 public class FieldMappingRule implements TransformationRule, Serializable {
 
     private static final long serialVersionUID = -3285027125865704371L;

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/FieldRelationShip.java

@@ -25,6 +25,11 @@
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
 import org.apache.inlong.sort.protocol.FieldInfo;
 
+
+/**
+ * defines the relationship between fields
+ * from input to output field
+ */
 @JsonTypeInfo(
         use = JsonTypeInfo.Id.NAME,
         include = JsonTypeInfo.As.PROPERTY,

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/FilterFunction.java

@@ -22,6 +22,9 @@
 import org.apache.inlong.sort.protocol.transformation.function.MultiValueFilterFunction;
 import org.apache.inlong.sort.protocol.transformation.function.SingleValueFilterFunction;
 
+/**
+ * interface for filter functions
+ */
 @JsonTypeInfo(
         use = JsonTypeInfo.Id.NAME,
         include = JsonTypeInfo.As.PROPERTY,

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/GroupTimeWindowFunction.java

@@ -23,6 +23,10 @@
 import org.apache.inlong.sort.protocol.transformation.function.SessionFunction;
 import org.apache.inlong.sort.protocol.transformation.function.TumbleFunction;
 
+/**
+ * top-level interface for window function in flink
+ * such as hop\session\tumble functions
+ */
 @JsonTypeInfo(
         use = JsonTypeInfo.Id.NAME,
         include = JsonTypeInfo.As.PROPERTY,

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/OrderDirection.java

@@ -17,6 +17,9 @@
 
 package org.apache.inlong.sort.protocol.transformation;
 
+/**
+ * order direction used in a flink sql
+ */
 public enum OrderDirection {
     /**
      * Asc for order direction

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/TimeWindowFunction.java

@@ -29,6 +29,9 @@
 import org.apache.inlong.sort.protocol.transformation.function.TumbleFunction;
 import org.apache.inlong.sort.protocol.transformation.function.TumbleStartFunction;
 
+/**
+ * interface for all the window functions appeared in flink
+ */
 @JsonTypeInfo(
         use = JsonTypeInfo.Id.NAME,
         include = JsonTypeInfo.As.PROPERTY,

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/WatermarkField.java

@@ -28,6 +28,9 @@
 import java.util.Arrays;
 import java.util.List;
 
+/**
+ * defines the field for generating watermarks
+ */
 @JsonTypeName("watermark")
 @Data
 @NoArgsConstructor

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/function/HopEndFunction.java

@@ -32,6 +32,9 @@
 import java.util.Arrays;
 import java.util.List;
 
+/**
+ * return hop end of a time window function
+ */
 @JsonTypeName("hopEnd")
 @Data
 @NoArgsConstructor

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/function/HopFunction.java

@@ -32,6 +32,14 @@
 import java.util.Arrays;
 import java.util.List;
 
+/**
+ * The HOP function assigns elements to windows of fixed length.
+ * Like a TUMBLE windowing function, the size of the windows is configured by the window size
+ * parameter. An additional window slide parameter controls how frequently a hopping
+ * window is started. Hence, hopping windows can be overlapping if the slide is smaller
+ * than the window size. In this case, elements are assigned to multiple windows.
+ * Hopping windows are also known as “sliding windows”.
+ */
 @JsonTypeName("hop")
 @Data
 @NoArgsConstructor

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/function/HopStartFunction.java

@@ -32,6 +32,9 @@
 import java.util.Arrays;
 import java.util.List;
 
+/**
+ * hop start of a window function
+ */
 @JsonTypeName("hopStart")
 @Data
 @NoArgsConstructor

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/function/MultiValueFilterFunction.java

@@ -33,6 +33,10 @@
 import java.util.List;
 import java.util.stream.Collectors;
 
+
+/**
+ * filter for filtering out data with multi-values
+ */
 @JsonTypeName("multiValueFilter")
 @Data
 @NoArgsConstructor

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/function/SessionEndFunction.java

@@ -32,6 +32,9 @@
 import java.util.Arrays;
 import java.util.List;
 
+/**
+ * identifier for the end of a session window
+ */
 @JsonTypeName("sessionEnd")
 @Data
 @NoArgsConstructor

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/function/SessionFunction.java

@@ -32,6 +32,14 @@
 import java.util.Arrays;
 import java.util.List;
 
+/**
+ * The sliding windows assigner assigns elements to windows of fixed length.
+ * Similar to a tumbling windows assigner, the size of the windows is configured
+ * by the window size parameter. An additional window slide parameter
+ * controls how frequently a sliding window is started. Hence,
+ * sliding windows can be overlapping if the slide is smaller than the window size.
+ * In this case elements are assigned to multiple windows.
+ */
 @JsonTypeName("session")
 @Data
 @NoArgsConstructor

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/transformation/function/SessionStartFunction.java

@@ -32,6 +32,9 @@
 import java.util.Arrays;
 import java.util.List;
 
+/**
+ * session start for a session window function
+ */
 @JsonTypeName("sessionStart")
 @Data
 @NoArgsConstructor