private DataStreamSource<RowData> createFLIP27Stream(
            StreamExecutionEnvironment env,
            TypeInformation<RowData> typeInfo) {
        SplitAssignerType assignerType =
                readableConfig.get(FlinkConfigOptions.TABLE_EXEC_SPLIT_ASSIGNER_TYPE);
        MetadataConverter[] converters = getMetadataConverters();
        IcebergSource<RowData> source =
                IcebergSource.forRowData()
                        .tableLoader(loader)
                        .assignerFactory(assignerType.factory())
                        .properties(properties)
                        .project(getProjectedSchema())
                        .limit(limit)
                        .filters(filters)
                        .flinkConfig(readableConfig)
                        .metadataConverters(converters)
                        .build();
        DataStreamSource<RowData> stream =
                env.fromSource(
                        source,
                        WatermarkStrategy.noWatermarks(),
                        source.name(),
                        typeInfo);
        return stream;
    }
Flink Iceberg table source.
Copy from iceberg-flink:iceberg-flink-1.15:1.3.1
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/source/IcebergTableSource.java