[flink] Do not chain for AppendBypassCompactWorkerOperator (apache#4083)

tongcheng-elong · Aug 28, 2024 · 05d278e · 05d278e
1 parent 3efd2f3
commit 05d278e
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 1 deletion.
diff --git a/...-common/src/main/java/org/apache/paimon/flink/sink/AppendBypassCompactWorkerOperator.java b/...-common/src/main/java/org/apache/paimon/flink/sink/AppendBypassCompactWorkerOperator.java
@@ -21,6 +21,7 @@
 import org.apache.paimon.append.UnawareAppendCompactionTask;
 import org.apache.paimon.table.FileStoreTable;
 
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.types.Either;
 
@@ -30,6 +31,7 @@ public class AppendBypassCompactWorkerOperator
 
     public AppendBypassCompactWorkerOperator(FileStoreTable table, String commitUser) {
         super(table, commitUser);
+        this.chainingStrategy = ChainingStrategy.HEAD;
     }
 
     @Override

diff --git a/...k-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperator.java b/...k-common/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperator.java
@@ -67,7 +67,7 @@ public AppendBypassCoordinateOperator(
         this.table = table;
         this.processingTimeService = processingTimeService;
         this.mailbox = (MailboxExecutorImpl) mailbox;
-        this.chainingStrategy = ChainingStrategy.NEVER;
+        this.chainingStrategy = ChainingStrategy.HEAD;
     }
 
     @Override