fapaul commented on code in PR #26846:
URL: https://github.com/apache/flink/pull/26846#discussion_r2288221116
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSinkITCase.java:
##########
@@ -584,34 +590,44 @@ private TestSource(List<Row> rows) {
}
@Override
- public ScanTableSource.ScanRuntimeProvider getScanRuntimeProvider(
- ScanTableSource.ScanContext context) {
- final DynamicTableSource.DataStructureConverter converter =
+ public ScanTableSource.ScanRuntimeProvider
getScanRuntimeProvider(ScanContext context) {
+ DynamicTableSource.DataStructureConverter converter =
context.createDataStructureConverter(
getFactoryContext().getPhysicalRowDataType());
- return SourceFunctionProvider.of(new TestSourceFunction(rows,
converter), false);
- }
- }
-
- private static class TestSourceFunction implements SourceFunction<RowData>
{
-
- private final List<Row> rows;
- private final DynamicTableSource.DataStructureConverter converter;
-
- public TestSourceFunction(
- List<Row> rows, DynamicTableSource.DataStructureConverter
converter) {
- this.rows = rows;
- this.converter = converter;
- }
+ // Can't use DataGeneratorSource distributes work across parallel
readers causing
+ // timestamp reordering
+ return SourceProvider.of(
+ new AbstractTestSource<>() {
+ @Override
+ public TestSourceReader<RowData>
createReader(SourceReaderContext ctx) {
+ return new TestSourceReader<>(ctx) {
+ final int subtask = ctx.getIndexOfSubtask();
+ boolean emitted = false;
Review Comment:
This mechanism looks a bit strange. Can we either use parallelism 1 for the
test source to ensure we have a single reader only or only create a single
split that the readers never get to pollNext?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
