[core] Remove EMIT_PER_BATCH in AppendBypassCoordinateOperator to avo…

…id slow compaction
apache · Sep 25, 2024 · 36c2756 · 36c2756
1 parent 3e87195
commit 36c2756
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 21 deletions.
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
@@ -23,14 +23,12 @@
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.utils.ExecutorUtils;
 
-import org.apache.flink.api.common.operators.MailboxExecutor;
 import org.apache.flink.api.common.operators.ProcessingTimeService.ProcessingTimeCallback;
 import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
 import org.apache.flink.streaming.api.operators.ChainingStrategy;
 import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
-import org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl;
 import org.apache.flink.types.Either;
 
 import java.util.List;
@@ -52,21 +50,16 @@ public class AppendBypassCoordinateOperator<CommitT>
                 ProcessingTimeCallback {
 
     private static final long MAX_PENDING_TASKS = 5000;
-    private static final long EMIT_PER_BATCH = 100;
 
     private final FileStoreTable table;
-    private final MailboxExecutorImpl mailbox;
 
     private transient ScheduledExecutorService executorService;
     private transient LinkedBlockingQueue<UnawareAppendCompactionTask> compactTasks;
 
     public AppendBypassCoordinateOperator(
-            FileStoreTable table,
-            ProcessingTimeService processingTimeService,
-            MailboxExecutor mailbox) {
+            FileStoreTable table, ProcessingTimeService processingTimeService) {
         this.table = table;
         this.processingTimeService = processingTimeService;
-        this.mailbox = (MailboxExecutorImpl) mailbox;
         this.chainingStrategy = ChainingStrategy.HEAD;
     }
 
@@ -100,14 +93,12 @@ private void asyncPlan(UnawareAppendTableCompactionCoordinator coordinator) {
 
     @Override
     public void onProcessingTime(long time) {
-        while (mailbox.isIdle()) {
-            for (int i = 0; i < EMIT_PER_BATCH; i++) {
-                UnawareAppendCompactionTask task = compactTasks.poll();
-                if (task == null) {
-                    return;
-                }
-                output.collect(new StreamRecord<>(Either.Right(task)));
+        while (true) {
+            UnawareAppendCompactionTask task = compactTasks.poll();
+            if (task == null) {
+                return;
             }
+            output.collect(new StreamRecord<>(Either.Right(task)));
         }
     }
 

diff --git a/...n/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperatorFactory.java b/...n/src/main/java/org/apache/paimon/flink/source/AppendBypassCoordinateOperatorFactory.java
@@ -25,15 +25,13 @@
 import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
 import org.apache.flink.streaming.api.operators.StreamOperator;
 import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
-import org.apache.flink.streaming.api.operators.YieldingOperatorFactory;
 import org.apache.flink.types.Either;
 
 /** Factory of {@link AppendBypassCoordinateOperator}. */
 public class AppendBypassCoordinateOperatorFactory<CommitT>
         extends AbstractStreamOperatorFactory<Either<CommitT, UnawareAppendCompactionTask>>
-        implements YieldingOperatorFactory<Either<CommitT, UnawareAppendCompactionTask>>,
-                OneInputStreamOperatorFactory<
-                        CommitT, Either<CommitT, UnawareAppendCompactionTask>> {
+        implements OneInputStreamOperatorFactory<
+                CommitT, Either<CommitT, UnawareAppendCompactionTask>> {
 
     private final FileStoreTable table;
 
@@ -47,8 +45,7 @@ T createStreamOperator(
                     StreamOperatorParameters<Either<CommitT, UnawareAppendCompactionTask>>
                             parameters) {
         AppendBypassCoordinateOperator<CommitT> operator =
-                new AppendBypassCoordinateOperator<>(
-                        table, processingTimeService, getMailboxExecutor());
+                new AppendBypassCoordinateOperator<>(table, processingTimeService);
         operator.setup(
                 parameters.getContainingTask(),
                 parameters.getStreamConfig(),