This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 8e9868d464 [flink] Avoid having QueryFileMonitor always busy (#9291)
8e9868d464 is described below
commit 8e9868d4645540864c2790959c072799b3db2bf9
Author: Eunbin Son <[email protected]>
AuthorDate: Mon Aug 24 00:27:06 2026 +0900
[flink] Avoid having QueryFileMonitor always busy (#9291)
---
.../paimon/flink/service/QueryFileMonitor.java | 61 +++++-
.../paimon/flink/service/QueryFileMonitorTest.java | 222 +++++++++++++++++++++
2 files changed, 281 insertions(+), 2 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryFileMonitor.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryFileMonitor.java
index 6688503778..ebc6f7bf3e 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryFileMonitor.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryFileMonitor.java
@@ -19,6 +19,7 @@
package org.apache.paimon.flink.service;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.flink.source.AbstractNonCoordinatedSource;
@@ -33,6 +34,7 @@ import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.StreamTableScan;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.table.system.FileMonitorTable;
+import org.apache.paimon.utils.ExecutorThreadFactory;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.connector.source.Boundedness;
@@ -45,6 +47,11 @@ import
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.ArrayList;
import java.util.List;
+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;
import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
@@ -79,12 +86,35 @@ public class QueryFileMonitor extends
AbstractNonCoordinatedSource<InternalRow>
@Override
public SourceReader<InternalRow, SimpleSourceSplit> createReader(
SourceReaderContext sourceReaderContext) throws Exception {
- return new Reader();
+ return new Reader(
+ Executors.newSingleThreadScheduledExecutor(
+ new ExecutorThreadFactory("query-file-monitor-timer")),
+ true);
+ }
+
+ @VisibleForTesting
+ SourceReader<InternalRow, SimpleSourceSplit> createReaderWithTimer(
+ ScheduledExecutorService timer) {
+ return new Reader(timer, false);
}
private class Reader extends
AbstractNonCoordinatedSourceReader<InternalRow> {
+
+ /** Fires the delayed wake-up without occupying a thread while the
delay elapses. */
+ private final ScheduledExecutorService timer;
+
+ /** Whether this reader created {@link #timer} and therefore has to
shut it down. */
+ private final boolean ownsTimer;
+
private transient StreamTableScan scan;
private transient TableRead read;
+ private CompletableFuture<Void> availableFuture =
CompletableFuture.completedFuture(null);
+ private ScheduledFuture<?> pendingWakeUp;
+
+ private Reader(ScheduledExecutorService timer, boolean ownsTimer) {
+ this.timer = timer;
+ this.ownsTimer = ownsTimer;
+ }
@Override
public void start() {
@@ -94,16 +124,43 @@ public class QueryFileMonitor extends
AbstractNonCoordinatedSource<InternalRow>
this.read = readBuilder.newRead();
}
+ @Override
+ public CompletableFuture<Void> isAvailable() {
+ return availableFuture;
+ }
+
@Override
public InputStatus pollNext(ReaderOutput<InternalRow> readerOutput)
throws Exception {
boolean isEmpty = doScan(readerOutput);
if (isEmpty) {
- Thread.sleep(monitorInterval);
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ availableFuture = future;
+ pendingWakeUp =
+ timer.schedule(
+ () -> {
+ future.complete(null);
+ },
+ monitorInterval,
+ TimeUnit.MILLISECONDS);
+ return InputStatus.NOTHING_AVAILABLE;
}
return InputStatus.MORE_AVAILABLE;
}
+ @Override
+ public void close() throws Exception {
+ if (pendingWakeUp != null) {
+ pendingWakeUp.cancel(false);
+ pendingWakeUp = null;
+ }
+ // unblock anyone still waiting on availability instead of leaving
them hanging
+ availableFuture.complete(null);
+ if (ownsTimer) {
+ timer.shutdownNow();
+ }
+ }
+
private boolean doScan(ReaderOutput<InternalRow> readerOutput) throws
Exception {
List<InternalRow> records = new ArrayList<>();
read.createReader(scan.plan()).forEachRemaining(records::add);
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/service/QueryFileMonitorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/service/QueryFileMonitorTest.java
new file mode 100644
index 0000000000..3a1af8ca27
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/service/QueryFileMonitorTest.java
@@ -0,0 +1,222 @@
+/*
+ * 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.service;
+
+import org.apache.paimon.CoreOptions;
+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.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.flink.source.SimpleSourceSplit;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.types.DataTypes;
+
+import org.apache.flink.api.common.eventtime.Watermark;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceOutput;
+import org.apache.flink.api.connector.source.SourceReader;
+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.ArrayList;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link QueryFileMonitor}. */
+public class QueryFileMonitorTest {
+
+ private static final long DISCOVERY_INTERVAL_MS = 3_000L;
+
+ @TempDir Path tempDir;
+
+ private Table 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())
+ .column("c", DataTypes.INT())
+ .primaryKey("a")
+ .option("bucket", "1")
+
.option(CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL.key(), "3 s")
+ .build();
+ Identifier identifier = Identifier.create("default", "t");
+ catalog.createDatabase("default", false);
+ catalog.createTable(identifier, schema, false);
+ this.table = catalog.getTable(identifier);
+ }
+
+ @Test
+ public void testPollWithoutNewFilesReportsNothingAvailable() throws
Exception {
+ SourceReader<InternalRow, SimpleSourceSplit> reader =
+ new QueryFileMonitor(table).createReader(null);
+ try {
+ reader.start();
+ TestingReaderOutput<InternalRow> output = new
TestingReaderOutput<>();
+
+ InputStatus status = reader.pollNext(output);
+
+ assertThat(status).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+ assertThat(output.getEmittedRecords()).isEmpty();
+ // the poll must not block the mailbox thread for the discovery
interval
+ assertThat(reader.isAvailable().isDone()).isFalse();
+ // availability is restored once the discovery interval has elapsed
+ reader.isAvailable().get(DISCOVERY_INTERVAL_MS * 10,
TimeUnit.MILLISECONDS);
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ public void testPollWithNewFilesReportsMoreAvailable() throws Exception {
+ writeToTable(1, 2, 3);
+ SourceReader<InternalRow, SimpleSourceSplit> reader =
+ new QueryFileMonitor(table).createReader(null);
+ try {
+ reader.start();
+ TestingReaderOutput<InternalRow> output = new
TestingReaderOutput<>();
+
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.MORE_AVAILABLE);
+ assertThat(output.getEmittedRecords()).isNotEmpty();
+ assertThat(reader.isAvailable().isDone()).isTrue();
+ } finally {
+ reader.close();
+ }
+ }
+
+ @Test
+ public void testConcurrentWaitsDoNotBlockEachOther() throws Exception {
+ ManuallyTriggeredScheduledExecutorService timer =
+ new ManuallyTriggeredScheduledExecutorService();
+ QueryFileMonitor monitor = new QueryFileMonitor(table);
+ SourceReader<InternalRow, SimpleSourceSplit> first =
monitor.createReaderWithTimer(timer);
+ SourceReader<InternalRow, SimpleSourceSplit> second =
monitor.createReaderWithTimer(timer);
+ try {
+ first.start();
+ second.start();
+ TestingReaderOutput<InternalRow> output = new
TestingReaderOutput<>();
+
+
assertThat(first.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+
assertThat(second.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+
+ // each wait is a queued timer task instead of a thread sleeping
for the interval, so
+ // the second reader's delay is already pending while the first
one has not elapsed
+ assertThat(timer.getAllNonPeriodicScheduledTask()).hasSize(2);
+ assertThat(first.isAvailable().isDone()).isFalse();
+ assertThat(second.isAvailable().isDone()).isFalse();
+
+ timer.triggerNonPeriodicScheduledTasks();
+
+ assertThat(first.isAvailable().isDone()).isTrue();
+ assertThat(second.isAvailable().isDone()).isTrue();
+ } finally {
+ first.close();
+ second.close();
+ }
+ }
+
+ @Test
+ public void testCloseCancelsThePendingWait() throws Exception {
+ ManuallyTriggeredScheduledExecutorService timer =
+ new ManuallyTriggeredScheduledExecutorService();
+ SourceReader<InternalRow, SimpleSourceSplit> reader =
+ new QueryFileMonitor(table).createReaderWithTimer(timer);
+ reader.start();
+
+ assertThat(reader.pollNext(new TestingReaderOutput<>()))
+ .isEqualTo(InputStatus.NOTHING_AVAILABLE);
+ ScheduledFuture<?> wakeUp =
timer.getAllNonPeriodicScheduledTask().get(0);
+ assertThat(wakeUp.isCancelled()).isFalse();
+
+ reader.close();
+
+ // the delay does not outlive the reader
+ assertThat(wakeUp.isCancelled()).isTrue();
+ assertThat(reader.isAvailable().isDone()).isTrue();
+ }
+
+ private void writeToTable(int a, int b, int c) throws Exception {
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ BatchTableWrite write = writeBuilder.newWrite();
+ write.write(GenericRow.of(a, b, c));
+ BatchTableCommit commit = writeBuilder.newCommit();
+ commit.commit(write.prepareCommit());
+ write.close();
+ commit.close();
+ }
+
+ private static final class TestingReaderOutput<E> implements
ReaderOutput<E> {
+
+ private final ArrayList<E> emittedRecords = new ArrayList<>();
+
+ @Override
+ public void collect(E record) {
+ emittedRecords.add(record);
+ }
+
+ @Override
+ public void collect(E record, long timestamp) {
+ collect(record);
+ }
+
+ @Override
+ public void emitWatermark(Watermark watermark) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void markIdle() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void markActive() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SourceOutput<E> createOutputForSplit(String splitId) {
+ return this;
+ }
+
+ @Override
+ public void releaseOutputForSplit(String splitId) {}
+
+ public ArrayList<E> getEmittedRecords() {
+ return emittedRecords;
+ }
+ }
+}
