This is an automated email from the ASF dual-hosted git repository. fhueske pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 34bda761e9c7d5f5bcdf4ca5cab2d71c76ab3ca7 Author: Fabian Hueske <[email protected]> AuthorDate: Wed Jul 8 20:14:01 2026 +0200 [FLINK-39785][table] Honor source.sleep-* in TestValues watermark-push-down source Wire the existing source.sleep-after-elements / source.sleep-time options into TestValuesScanTableSourceWithWatermarkPushDown Generated-By: Claude Opus 4.8 (1M context) --- .../factories/TestValuesRuntimeFunctions.java | 18 +++++++++++++++- .../planner/factories/TestValuesTableFactory.java | 24 ++++++++++++++++++---- 2 files changed, 37 insertions(+), 5 deletions(-) diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java index df439bd2c9e..851275d4ef8 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java @@ -257,15 +257,24 @@ public final class TestValuesRuntimeFunctions { private final TerminatingLogic terminating; + /** Sleep for {@link #sleepTimeMillis} after emitting every {@code sleepAfterElements}. */ + private final int sleepAfterElements; + + private final long sleepTimeMillis; + public FromElementSourceFunctionWithWatermark( String tableName, TypeSerializer<RowData> serializer, Iterable<RowData> elements, WatermarkStrategy<RowData> watermarkStrategy, - TerminatingLogic terminating) + TerminatingLogic terminating, + int sleepAfterElements, + long sleepTimeMillis) throws IOException { this.tableName = tableName; this.terminating = terminating; + this.sleepAfterElements = sleepAfterElements; + this.sleepTimeMillis = sleepTimeMillis; ByteArrayOutputStream baos = new ByteArrayOutputStream(); DataOutputViewStreamWrapper wrapper = new DataOutputViewStreamWrapper(baos); @@ -325,6 +334,13 @@ public final class TestValuesRuntimeFunctions { generator.onEvent(next, Long.MIN_VALUE, output); generator.onPeriodicEmit(output); } + + // If enabled, throttle emission of values + if (sleepAfterElements > 0 + && sleepTimeMillis > 0 + && numElementsEmitted % sleepAfterElements == 0) { + Thread.sleep(sleepTimeMillis); + } } if (terminating == TerminatingLogic.INFINITE) { diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java index 3387f522232..4e260a3e75f 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java @@ -716,7 +716,9 @@ public final class TestValuesTableFactory partitions, readableMetadata, null, - enableAggregatePushDown); + enableAggregatePushDown, + sleepAfterElements, + sleepTimeMillis); source.setEnableMetadataFilterPushDown(enableMetadataFilterPushDown); return source; } else { @@ -1750,6 +1752,8 @@ public final class TestValuesTableFactory extends TestValuesScanTableSource implements SupportsWatermarkPushDown, SupportsSourceWatermark { private final String tableName; + private final int sleepAfterElements; + private final long sleepTimeMillis; private WatermarkStrategy<RowData> watermarkStrategy = WatermarkStrategy.noWatermarks(); @@ -1771,7 +1775,9 @@ public final class TestValuesTableFactory List<Map<String, String>> allPartitions, Map<String, DataType> readableMetadata, @Nullable int[] projectedMetadataFields, - boolean enableAggregatePushDown) { + boolean enableAggregatePushDown, + int sleepAfterElements, + long sleepTimeMillis) { super( producedDataType, changelogMode, @@ -1792,6 +1798,8 @@ public final class TestValuesTableFactory projectedMetadataFields, enableAggregatePushDown); this.tableName = tableName; + this.sleepAfterElements = sleepAfterElements; + this.sleepTimeMillis = sleepTimeMillis; } @Override @@ -1817,7 +1825,13 @@ public final class TestValuesTableFactory try { return SourceFunctionProvider.of( new TestValuesRuntimeFunctions.FromElementSourceFunctionWithWatermark( - tableName, serializer, values, watermarkStrategy, terminating), + tableName, + serializer, + values, + watermarkStrategy, + terminating, + sleepAfterElements, + sleepTimeMillis), false); } catch (IOException e) { throw new TableException("Fail to init source function", e); @@ -1845,7 +1859,9 @@ public final class TestValuesTableFactory allPartitions, readableMetadata, projectedMetadataFields, - enableAggregatePushDown); + enableAggregatePushDown, + sleepAfterElements, + sleepTimeMillis); newSource.watermarkStrategy = watermarkStrategy; newSource.setEnableMetadataFilterPushDown(enableMetadataFilterPushDown); return newSource;
