[INLONG-11546][SDK] Support async and sync report dirty data

apache · Nov 26, 2024 · ccb31f2 · ccb31f2
1 parent 63a1173
commit ccb31f2
Show file tree

Hide file tree

Showing 2 changed files with 9 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
@@ -74,8 +74,14 @@ public void init() throws Exception {
         log.info("init InlongSdkDirtySink successfully, target group={}, stream={}", inlongGroupId, inlongStreamId);
     }
 
-    public void sendDirtyMessage(DirtyMessageWrapper messageWrapper) throws InterruptedException {
-        dirtyDataQueue.offer(messageWrapper, 10, TimeUnit.SECONDS);
+    public void sendDirtyMessageSync(DirtyMessageWrapper messageWrapper) throws InterruptedException {
+        dirtyDataQueue.put(messageWrapper);
+    }
+
+    public void sendDirtyMessageAsync(DirtyMessageWrapper messageWrapper) throws InterruptedException {
+        if(!dirtyDataQueue.offer(messageWrapper)) {
+            log.warn("the dirty data queue is full, you can increase the size of queue by configure maxCallbackSize");
+        }
     }
 
     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()) {