diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendTableCompactSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendTableCompactSource.java index dae05ff60f1c..a9ec081c5ec1 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/AppendTableCompactSource.java +++ b/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 { 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 isAvailable() { + return availability.future(); + } + @Override public InputStatus pollNext(ReaderOutput readerOutput) throws Exception { boolean isEmpty; @@ -114,10 +138,17 @@ public InputStatus pollNext(ReaderOutput 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 buildSource( diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DelayedAvailability.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DelayedAvailability.java new file mode 100644 index 000000000000..6bb1c422e1eb --- /dev/null +++ b/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 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 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 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(); + } + } +} diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAppendCompactStreamSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAppendCompactStreamSource.java index 34e062801bf4..6edfd5ef12eb 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAppendCompactStreamSource.java +++ b/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 createReader private class Reader extends AbstractNonCoordinatedSourceReader { private MultiTableScanBase 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 isAvailable() { + return availability.future(); + } + @Override public InputStatus pollNext(ReaderOutput readerOutput) throws Exception { @@ -91,13 +100,15 @@ public InputStatus pollNext(ReaderOutput 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(); diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAwareStreamingSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAwareStreamingSource.java index cc5c86bbc404..a0dce404e972 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAwareStreamingSource.java +++ b/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, SimpleSourceSplit> createReader( private class Reader extends AbstractNonCoordinatedSourceReader> { private transient MultiTableScanBase> 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 isAvailable() { + return availability.future(); + } + @Override public InputStatus pollNext(ReaderOutput> readerOutput) throws Exception { @@ -94,13 +103,15 @@ public InputStatus pollNext(ReaderOutput> 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(); diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/AppendTableCompactSourceTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/AppendTableCompactSourceTest.java new file mode 100644 index 000000000000..11943912aa0e --- /dev/null +++ b/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 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(); + } + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DelayedAvailabilityTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DelayedAvailabilityTest.java new file mode 100644 index 000000000000..a1b89838471a --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DelayedAvailabilityTest.java @@ -0,0 +1,101 @@ +/* + * 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.flink.core.testutils.ManuallyTriggeredScheduledExecutorService; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Test for {@link DelayedAvailability}. */ +public class DelayedAvailabilityTest { + + @Test + public void testAvailableUntilDelayed() { + try (DelayedAvailability availability = + new DelayedAvailability(new ManuallyTriggeredScheduledExecutorService())) { + assertThat(availability.future()).isDone(); + } + } + + @Test + public void testDelayCompletesWhenTimerFires() { + ManuallyTriggeredScheduledExecutorService timer = + new ManuallyTriggeredScheduledExecutorService(); + try (DelayedAvailability availability = new DelayedAvailability(timer)) { + availability.delay(1_000L); + + assertThat(availability.future()).isNotDone(); + ScheduledFuture wakeUp = timer.getAllNonPeriodicScheduledTask().get(0); + assertThat(wakeUp.getDelay(TimeUnit.MILLISECONDS)).isEqualTo(1_000L); + + timer.triggerNonPeriodicScheduledTasks(); + + assertThat(availability.future()).isDone(); + } + } + + @Test + public void testConcurrentDelaysDoNotBlockEachOther() { + ManuallyTriggeredScheduledExecutorService timer = + new ManuallyTriggeredScheduledExecutorService(); + try (DelayedAvailability first = new DelayedAvailability(timer); + DelayedAvailability second = new DelayedAvailability(timer)) { + first.delay(1_000L); + second.delay(1_000L); + + // each wait is a queued timer task rather than a thread sleeping through the interval + assertThat(timer.getAllNonPeriodicScheduledTask()).hasSize(2); + assertThat(first.future()).isNotDone(); + assertThat(second.future()).isNotDone(); + + timer.triggerNonPeriodicScheduledTasks(); + + assertThat(first.future()).isDone(); + assertThat(second.future()).isDone(); + } + } + + @Test + public void testCloseCancelsThePendingDelay() { + ManuallyTriggeredScheduledExecutorService timer = + new ManuallyTriggeredScheduledExecutorService(); + try (DelayedAvailability availability = new DelayedAvailability(timer)) { + availability.delay(1_000L); + ScheduledFuture wakeUp = timer.getAllNonPeriodicScheduledTask().get(0); + + availability.close(); + + assertThat(wakeUp.isCancelled()).isTrue(); + assertThat(availability.future()).isDone(); + } + } + + @Test + public void testOwnedTimerCompletesTheDelay() throws Exception { + try (DelayedAvailability availability = + new DelayedAvailability("delayed-availability-test")) { + availability.delay(10L); + availability.future().get(10, TimeUnit.SECONDS); + } + } +}