[INLONG-11546][SDK] Support async and sync report dirty data (#11547)

* [INLONG-11546][SDK] Support async and sync report dirty data
apache · Nov 26, 2024 · 031732a · 031732a
1 parent 4c475e9
commit 031732a
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 3 deletions.
diff --git a/...dk/dirty-data-sdk/src/main/java/org/apache/inlong/sdk/dirtydata/InlongSdkDirtySender.java b/...dk/dirty-data-sdk/src/main/java/org/apache/inlong/sdk/dirtydata/InlongSdkDirtySender.java
@@ -31,7 +31,6 @@
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
 
 @Slf4j
 @Builder
@@ -75,7 +74,11 @@ public void init() throws Exception {
     }
 
     public void sendDirtyMessage(DirtyMessageWrapper messageWrapper) throws InterruptedException {
-        dirtyDataQueue.offer(messageWrapper, 10, TimeUnit.SECONDS);
+        dirtyDataQueue.put(messageWrapper);
+    }
+
+    public boolean sendDirtyMessageAsync(DirtyMessageWrapper messageWrapper) {
+        return dirtyDataQueue.offer(messageWrapper);
     }
 
     private void doSendDirtyMessage() {

diff --git a/...ink/base/src/main/java/org/apache/inlong/sort/base/dirty/sink/sdk/InlongSdkDirtySink.java b/...ink/base/src/main/java/org/apache/inlong/sort/base/dirty/sink/sdk/InlongSdkDirtySink.java
@@ -74,7 +74,7 @@ public void invoke(DirtyData<T> dirtyData) throws Exception {
                     .data(dirtyMessage)
                     .build();
 
-            dirtySender.sendDirtyMessage(wrapper);
+            dirtySender.sendDirtyMessageAsync(wrapper);
         } catch (Throwable t) {
             log.error("failed to send dirty message to inlong sdk", t);
             if (!options.isIgnoreSideOutputErrors()) {