[INLONG-11481][Sort] Tube Connector source supports dirty data achiev…

…ing (#11482)
apache · Nov 11, 2024 · ab59b91 · ab59b91
1 parent 87d0a92
commit ab59b91
Show file tree

Hide file tree

Showing 13 changed files with 243 additions and 135 deletions.
diff --git a/...sdk/dirty-data-sdk/src/main/java/org/apache/inlong/sdk/dirtydata/DirtyMessageWrapper.java b/...sdk/dirty-data-sdk/src/main/java/org/apache/inlong/sdk/dirtydata/DirtyMessageWrapper.java
@@ -18,12 +18,17 @@
 package org.apache.inlong.sdk.dirtydata;
 
 import lombok.Builder;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.text.StringEscapeUtils;
 
+import java.time.Instant;
 import java.time.LocalDateTime;
+import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
 import java.util.Base64;
 import java.util.StringJoiner;
 
+@Slf4j
 @Builder
 public class DirtyMessageWrapper {
 
@@ -32,16 +37,17 @@ public class DirtyMessageWrapper {
 
     private String inlongGroupId;
     private String inlongStreamId;
-    private String dataTime;
+    private long dataTime;
     private String dataflowId;
     private String serverType;
     private String dirtyType;
+    private String dirtyMessage;
     private String ext;
     private String data;
     private byte[] dataBytes;
 
     public String format() {
-        String now = LocalDateTime.now().format(dateTimeFormatter);
+        String reportTime = LocalDateTime.now().format(dateTimeFormatter);
         StringJoiner joiner = new StringJoiner(delimiter);
         String formatData = null;
         if (data != null) {
@@ -50,14 +56,19 @@ public String format() {
             formatData = Base64.getEncoder().encodeToString(dataBytes);
         }
 
-        return joiner.add(inlongGroupId)
-                .add(inlongStreamId)
-                .add(now)
-                .add(dataTime)
+        String dataTimeStr = LocalDateTime.ofInstant(Instant.ofEpochMilli(dataTime),
+                ZoneId.systemDefault()).format(dateTimeFormatter);
+        return joiner
                 .add(dataflowId)
+                .add(inlongGroupId)
+                .add(inlongStreamId)
+                .add(reportTime)
+                .add(dataTimeStr)
                 .add(serverType)
                 .add(dirtyType)
-                .add(ext)
-                .add(formatData).toString();
+                .add(StringEscapeUtils.escapeXSI(dirtyMessage))
+                .add(StringEscapeUtils.escapeXSI(ext))
+                .add(StringEscapeUtils.escapeXSI(formatData))
+                .toString();
     }
 }
diff --git a/...ong/sdk/dirtydata/InlongSdkDirtySink.java → ...g/sdk/dirtydata/InlongSdkDirtySender.java b/...ong/sdk/dirtydata/InlongSdkDirtySink.java → ...g/sdk/dirtydata/InlongSdkDirtySender.java
@@ -18,7 +18,6 @@
 package org.apache.inlong.sdk.dirtydata;
 
 import org.apache.inlong.sdk.dataproxy.DefaultMessageSender;
-import org.apache.inlong.sdk.dataproxy.MessageSender;
 import org.apache.inlong.sdk.dataproxy.ProxyClientConfig;
 import org.apache.inlong.sdk.dataproxy.common.SendMessageCallback;
 import org.apache.inlong.sdk.dataproxy.common.SendResult;
@@ -28,19 +27,22 @@
 import lombok.Builder;
 import lombok.extern.slf4j.Slf4j;
 
+import java.net.InetAddress;
+
 @Slf4j
 @Builder
-public class InlongSdkDirtySink {
+public class InlongSdkDirtySender {
 
     private String inlongGroupId;
     private String inlongStreamId;
     private String inlongManagerAddr;
+    private int inlongManagerPort;
     private String authId;
     private String authKey;
     private boolean ignoreErrors;
 
     private SendMessageCallback callback;
-    private MessageSender sender;
+    private DefaultMessageSender sender;
 
     public void init() throws Exception {
         Preconditions.checkNotNull(inlongGroupId, "inlongGroupId cannot be null");
@@ -51,8 +53,11 @@ public void init() throws Exception {
 
         this.callback = new LogCallBack();
         ProxyClientConfig proxyClientConfig =
-                new ProxyClientConfig(inlongManagerAddr, inlongGroupId, authId, authKey);
+                new ProxyClientConfig(InetAddress.getLocalHost().getHostAddress(), true,
+                        inlongManagerAddr, inlongManagerPort, inlongGroupId, authId, authKey);
+        proxyClientConfig.setReadProxyIPFromLocal(false);
         this.sender = DefaultMessageSender.generateSenderByClusterId(proxyClientConfig);
+        this.sender.setMsgtype(7);
         log.info("init InlongSdkDirtySink successfully, target group={}, stream={}", inlongGroupId, inlongStreamId);
     }
 
@@ -61,6 +66,12 @@ public void sendDirtyMessage(DirtyMessageWrapper messageWrapper)
         sender.asyncSendMessage(inlongGroupId, inlongStreamId, messageWrapper.format().getBytes(), callback);
     }
 
+    public void close() {
+        if (sender != null) {
+            sender.close();
+        }
+    }
+
     class LogCallBack implements SendMessageCallback {
 
         @Override

diff --git a/inlong-sort/sort-flink/base/pom.xml b/inlong-sort/sort-flink/base/pom.xml
@@ -38,6 +38,11 @@
             <artifactId>audit-sdk</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.inlong</groupId>
+            <artifactId>dirty-data-sdk</artifactId>
+            <version>${project.version}</version>
+        </dependency>
 
         <dependency>
             <groupId>org.apache.inlong</groupId>

diff --git a/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/Constants.java b/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/Constants.java
@@ -385,7 +385,7 @@ public final class Constants {
     public static final ConfigOption<String> DIRTY_SIDE_OUTPUT_FIELD_DELIMITER =
             ConfigOptions.key("dirty.side-output.field-delimiter")
                     .stringType()
-                    .defaultValue(",")
+                    .defaultValue("|")
                     .withDescription("The field-delimiter of dirty side-output");
     public static final ConfigOption<String> DIRTY_SIDE_OUTPUT_LINE_DELIMITER =
             ConfigOptions.key("dirty.side-output.line-delimiter")

diff --git a/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/dirty/DirtyData.java b/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/dirty/DirtyData.java
@@ -71,14 +71,22 @@ public class DirtyData<T> {
      * The row type of data, it is only used for 'RowData'
      */
     private @Nullable final LogicalType rowType;
+    /**
+     * Dirty message data time
+     */
+    private final long dataTime;
+    /**
+     * Dirty message ext params
+     */
+    private @Nullable final String extParams;
     /**
      * The real dirty data
      */
     private final T data;
 
     public DirtyData(T data, String identifier, String labels,
             String logTag, DirtyType dirtyType, String dirtyMessage,
-            @Nullable LogicalType rowType) {
+            @Nullable LogicalType rowType, long dataTime, String extParams) {
         this.data = data;
         this.dirtyType = dirtyType;
         this.dirtyMessage = dirtyMessage;
@@ -87,7 +95,8 @@ public DirtyData(T data, String identifier, String labels,
         this.labels = PatternReplaceUtils.replace(labels, paramMap);
         this.logTag = PatternReplaceUtils.replace(logTag, paramMap);
         this.identifier = PatternReplaceUtils.replace(identifier, paramMap);
-
+        this.dataTime = dataTime == 0 ? System.currentTimeMillis() : dataTime;
+        this.extParams = extParams;
     }
 
     public static <T> Builder<T> builder() {
@@ -122,6 +131,18 @@ public String getIdentifier() {
         return identifier;
     }
 
+    public long getDataTime() {
+        return dataTime;
+    }
+
+    public String getExtParams() {
+        return extParams;
+    }
+
+    public String getDirtyMessage() {
+        return dirtyMessage;
+    }
+
     @Nullable
     public LogicalType getRowType() {
         return rowType;
@@ -135,8 +156,20 @@ public static class Builder<T> {
         private DirtyType dirtyType = DirtyType.UNDEFINED;
         private String dirtyMessage;
         private LogicalType rowType;
+        private long dataTime;
+        private String extParams;
         private T data;
 
+        public Builder<T> setDirtyDataTime(long dataTime) {
+            this.dataTime = dataTime;
+            return this;
+        }
+
+        public Builder<T> setExtParams(String extParams) {
+            this.extParams = extParams;
+            return this;
+        }
+
         public Builder<T> setDirtyType(DirtyType dirtyType) {
             this.dirtyType = dirtyType;
             return this;
@@ -173,7 +206,8 @@ public Builder<T> setRowType(LogicalType rowType) {
         }
 
         public DirtyData<T> build() {
-            return new DirtyData<>(data, identifier, labels, logTag, dirtyType, dirtyMessage, rowType);
+            return new DirtyData<>(data, identifier, labels, logTag, dirtyType,
+                    dirtyMessage, rowType, dataTime, extParams);
         }
     }
 }
diff --git a/...base/dirty/sink/sdk/InlongSdkOptions.java → ...dirty/sink/sdk/InlongSdkDirtyOptions.java b/...base/dirty/sink/sdk/InlongSdkOptions.java → ...dirty/sink/sdk/InlongSdkDirtyOptions.java
@@ -26,7 +26,7 @@
 @Data
 @Builder
 @Getter
-public class InlongSdkOptions implements Serializable {
+public class InlongSdkDirtyOptions implements Serializable {
 
     private static final String DEFAULT_FORMAT = "csv";
 
@@ -36,9 +36,10 @@ public class InlongSdkOptions implements Serializable {
     private static final String DEFAULT_KV_FIELD_DELIMITER = "&";
     private static final String DEFAULT_KV_ENTRY_DELIMITER = "=";
 
-    private String inlongGroupId;
-    private String inlongStreamId;
+    private String sendToGroupId;
+    private String sendToStreamId;
     private String inlongManagerAddr;
+    private int inlongManagerPort;
     private String inlongManagerAuthKey;
     private String inlongManagerAuthId;
     private String format = DEFAULT_FORMAT;