[flink] Avoid having append compact sources always busy

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendTableCompactSource.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.flink.source;
 
+import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.append.AppendCompactCoordinator;
 import org.apache.paimon.append.AppendCompactTask;
 import org.apache.paimon.flink.sink.CompactionTaskTypeInfo;
@@ -40,6 +41,7 @@
 import javax.annotation.Nullable;
 
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * Source for unaware-bucket Compaction.
@@ -90,16 +92,38 @@ public static class CompactSourceReader
             extends AbstractNonCoordinatedSourceReader<AppendCompactTask> {
         private final AppendCompactCoordinator compactionCoordinator;
         private final long scanInterval;
+        private final DelayedAvailability availability;
 
         public CompactSourceReader(
                 FileStoreTable table,
                 boolean streaming,
                 PartitionPredicate partitions,
                 long scanInterval) {
+            this(
+                    table,
+                    streaming,
+                    partitions,
+                    scanInterval,
+                    new DelayedAvailability("append-compact-source-timer"));
+        }
+
+        @VisibleForTesting
+        CompactSourceReader(
+                FileStoreTable table,
+                boolean streaming,
+                PartitionPredicate partitions,
+                long scanInterval,
+                DelayedAvailability availability) {
             this.scanInterval = scanInterval;
+            this.availability = availability;
             compactionCoordinator = new AppendCompactCoordinator(table, streaming, partitions);
         }
 
+        @Override
+        public CompletableFuture<Void> isAvailable() {
+            return availability.future();
+        }
+
         @Override
         public InputStatus pollNext(ReaderOutput<AppendCompactTask> readerOutput) throws Exception {
             boolean isEmpty;
@@ -114,10 +138,17 @@ public InputStatus pollNext(ReaderOutput<AppendCompactTask> readerOutput) throws
             }
 
             if (isEmpty) {
-                Thread.sleep(scanInterval);
+                availability.delay(scanInterval);
+                return InputStatus.NOTHING_AVAILABLE;
             }
             return InputStatus.MORE_AVAILABLE;
         }
+
+        @Override
+        public void close() throws Exception {
+            availability.close();
+            super.close();
+        }
     }
 
     public static DataStreamSource<AppendCompactTask> buildSource(

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DelayedAvailability.java

@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.source;
+
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.utils.ExecutorThreadFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Availability of a source reader that waits between scans. Sleeping in {@code pollNext} would
+ * block the task's mailbox thread and make the source report itself 100% busy, so the wait is a
+ * timer task that completes the future returned by {@link #future()} instead.
+ */
+public class DelayedAvailability implements AutoCloseable {
+
+    private final String timerName;
+
+    /** Created on the first wait, so a reader that never waits starts no thread. */
+    @Nullable private ScheduledExecutorService timer;
+
+    private final boolean ownsTimer;
+
+    private CompletableFuture<Void> future = CompletableFuture.completedFuture(null);
+
+    @Nullable private ScheduledFuture<?> pendingWakeUp;
+
+    public DelayedAvailability(String timerName) {
+        this.timerName = timerName;
+        this.ownsTimer = true;
+    }
+
+    @VisibleForTesting
+    public DelayedAvailability(ScheduledExecutorService timer) {
+        this.timerName = null;
+        this.timer = timer;
+        this.ownsTimer = false;
+    }
+
+    public CompletableFuture<Void> future() {
+        return future;
+    }
+
+    /** Makes the reader unavailable until {@code delayMillis} has elapsed. */
+    public void delay(long delayMillis) {
+        if (timer == null) {
+            timer =
+                    Executors.newSingleThreadScheduledExecutor(
+                            new ExecutorThreadFactory(timerName));
+        }
+        CompletableFuture<Void> next = new CompletableFuture<>();
+        future = next;
+        pendingWakeUp =
+                timer.schedule(() -> next.complete(null), delayMillis, TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public void close() {
+        if (pendingWakeUp != null) {
+            pendingWakeUp.cancel(false);
+            pendingWakeUp = null;
+        }
+        // unblock anyone still waiting on availability instead of leaving them hanging
+        future.complete(null);
+        if (ownsTimer && timer != null) {
+            timer.shutdownNow();
+        }
+    }
+}

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAppendCompactStreamSource.java

@@ -24,6 +24,7 @@
 import org.apache.paimon.flink.compact.MultiTableScanBase;
 import org.apache.paimon.flink.sink.MultiTableCompactionTaskTypeInfo;
 import org.apache.paimon.flink.source.AbstractNonCoordinatedSourceReader;
+import org.apache.paimon.flink.source.DelayedAvailability;
 import org.apache.paimon.flink.source.SimpleSourceSplit;
 
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
@@ -35,6 +36,7 @@
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 import java.util.regex.Pattern;
 
 import static org.apache.paimon.flink.compact.MultiTableScanBase.ScanResult.FINISHED;
@@ -69,6 +71,8 @@ public SourceReader<MultiTableAppendCompactTask, SimpleSourceSplit> createReader
 
     private class Reader extends AbstractNonCoordinatedSourceReader<MultiTableAppendCompactTask> {
         private MultiTableScanBase<MultiTableAppendCompactTask> tableScan;
+        private final DelayedAvailability availability =
+                new DelayedAvailability("combined-append-compact-source-timer");
 
         @Override
         public void start() {
@@ -83,6 +87,11 @@ public void start() {
                             tableOptions);
         }
 
+        @Override
+        public CompletableFuture<Void> isAvailable() {
+            return availability.future();
+        }
+
         @Override
         public InputStatus pollNext(ReaderOutput<MultiTableAppendCompactTask> readerOutput)
                 throws Exception {
@@ -91,13 +100,15 @@ public InputStatus pollNext(ReaderOutput<MultiTableAppendCompactTask> readerOutp
                 return InputStatus.END_OF_INPUT;
             }
             if (scanResult == IS_EMPTY) {
-                Thread.sleep(monitorInterval);
+                availability.delay(monitorInterval);
+                return InputStatus.NOTHING_AVAILABLE;
             }
             return InputStatus.MORE_AVAILABLE;
         }
 
         @Override
         public void close() throws Exception {
+            availability.close();
             super.close();
             if (tableScan != null) {
                 tableScan.close();

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAwareStreamingSource.java

@@ -22,6 +22,7 @@
 import org.apache.paimon.flink.compact.MultiAwareBucketTableScan;
 import org.apache.paimon.flink.compact.MultiTableScanBase;
 import org.apache.paimon.flink.source.AbstractNonCoordinatedSourceReader;
+import org.apache.paimon.flink.source.DelayedAvailability;
 import org.apache.paimon.flink.source.SimpleSourceSplit;
 import org.apache.paimon.flink.utils.JavaTypeInfo;
 import org.apache.paimon.table.source.DataSplit;
@@ -41,6 +42,7 @@
 import org.apache.flink.table.data.RowData;
 
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 import java.util.regex.Pattern;
 
 import static org.apache.paimon.flink.compact.MultiTableScanBase.ScanResult.FINISHED;
@@ -72,6 +74,8 @@ public SourceReader<Tuple2<Split, String>, SimpleSourceSplit> createReader(
 
     private class Reader extends AbstractNonCoordinatedSourceReader<Tuple2<Split, String>> {
         private transient MultiTableScanBase<Tuple2<Split, String>> tableScan;
+        private final DelayedAvailability availability =
+                new DelayedAvailability("combined-aware-compact-source-timer");
 
         @Override
         public void start() {
@@ -86,6 +90,11 @@ public void start() {
                             tableOptions);
         }
 
+        @Override
+        public CompletableFuture<Void> isAvailable() {
+            return availability.future();
+        }
+
         @Override
         public InputStatus pollNext(ReaderOutput<Tuple2<Split, String>> readerOutput)
                 throws Exception {
@@ -94,13 +103,15 @@ public InputStatus pollNext(ReaderOutput<Tuple2<Split, String>> readerOutput)
                 return InputStatus.END_OF_INPUT;
             }
             if (scanResult == IS_EMPTY) {
-                Thread.sleep(monitorInterval);
+                availability.delay(monitorInterval);
+                return InputStatus.NOTHING_AVAILABLE;
             }
             return InputStatus.MORE_AVAILABLE;
         }
 
         @Override
         public void close() throws Exception {
+            availability.close();
             super.close();
             if (tableScan != null) {
                 tableScan.close();

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/AppendTableCompactSourceTest.java

@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.source;
+
+import org.apache.paimon.append.AppendCompactTask;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.DataTypes;
+
+import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.core.testutils.ManuallyTriggeredScheduledExecutorService;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+import java.util.concurrent.ScheduledFuture;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link AppendTableCompactSource}. */
+public class AppendTableCompactSourceTest {
+
+    private static final long SCAN_INTERVAL_MS = 3_000L;
+
+    @TempDir Path tempDir;
+
+    private FileStoreTable table;
+
+    @BeforeEach
+    public void before() throws Exception {
+        Catalog catalog =
+                CatalogFactory.createCatalog(
+                        CatalogContext.create(new org.apache.paimon.fs.Path(tempDir.toUri())));
+        Schema schema =
+                Schema.newBuilder()
+                        .column("a", DataTypes.INT())
+                        .column("b", DataTypes.INT())
+                        .option("bucket", "-1")
+                        .build();
+        Identifier identifier = Identifier.create("default", "t");
+        catalog.createDatabase("default", false);
+        catalog.createTable(identifier, schema, false);
+        this.table = (FileStoreTable) catalog.getTable(identifier);
+    }
+
+    @Test
+    public void testPollWithoutCompactionTaskReportsNothingAvailable() throws Exception {
+        ManuallyTriggeredScheduledExecutorService timer =
+                new ManuallyTriggeredScheduledExecutorService();
+        try (AppendTableCompactSource.CompactSourceReader reader =
+                new AppendTableCompactSource.CompactSourceReader(
+                        table, true, null, SCAN_INTERVAL_MS, new DelayedAvailability(timer))) {
+            TestingReaderOutput<AppendCompactTask> output = new TestingReaderOutput<>();
+
+            assertThat(reader.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+            assertThat(output.getEmittedRecords()).isEmpty();
+            // the poll must not block the mailbox thread for the scan interval
+            assertThat(reader.isAvailable()).isNotDone();
+
+            timer.triggerNonPeriodicScheduledTasks();
+
+            assertThat(reader.isAvailable()).isDone();
+        }
+    }
+
+    @Test
+    public void testCloseCancelsThePendingWait() throws Exception {
+        ManuallyTriggeredScheduledExecutorService timer =
+                new ManuallyTriggeredScheduledExecutorService();
+        try (AppendTableCompactSource.CompactSourceReader reader =
+                new AppendTableCompactSource.CompactSourceReader(
+                        table, true, null, SCAN_INTERVAL_MS, new DelayedAvailability(timer))) {
+            assertThat(reader.pollNext(new TestingReaderOutput<>()))
+                    .isEqualTo(InputStatus.NOTHING_AVAILABLE);
+            assertThat(timer.getAllNonPeriodicScheduledTask()).hasSize(1);
+            ScheduledFuture<?> wakeUp = timer.getAllNonPeriodicScheduledTask().get(0);
+
+            reader.close();
+
+            assertThat(wakeUp.isCancelled()).isTrue();
+            assertThat(reader.isAvailable()).isDone();
+        }
+    }
+}