wenshao commented on code in PR #4362:
URL: https://github.com/apache/flink-cdc/pull/4362#discussion_r3043294320
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactory.java:
##########
@@ -110,6 +110,8 @@ public DynamicTableSource createDynamicTableSource(Context
context) {
boolean appendOnly =
config.get(MySqlSourceOptions.SCAN_READ_CHANGELOG_AS_APPEND_ONLY_ENABLED);
+ double recordsPerSecond =
config.get(MySqlSourceOptions.RECORDS_PER_SECOND);
Review Comment:
**[Critical]** `RECORDS_PER_SECOND` is not registered in `optionalOptions()`
for all 6 SQL TableSourceFactory implementations
All 6 SQL `DynamicTableSourceFactory` classes (`MySqlTableSourceFactory`,
`Db2TableSourceFactory`, `SqlServerTableSourceFactory`,
`OracleTableSourceFactory`, `PostgreSQLTableSourceFactory`,
`MongoDBTableSourceFactory`) read the value via
`config.get(RECORDS_PER_SECOND)` but never declare it in their
`optionalOptions()` method. Flink's `FactoryUtil` validates that all
user-provided options are declared — this means SQL/Table API users will get an
"Unsupported options found" validation error when using `records.per.second` in
`CREATE TABLE ... WITH ('records.per.second' = '100')`.
The Pipeline `DataSourceFactory` classes correctly register it via
`options.add(RECORDS_PER_SECOND)`, so the feature works for Pipeline API but is
broken for SQL/Table API.
**Suggested fix:** Add `options.add(RECORDS_PER_SECOND);` to
`optionalOptions()` in each of the 6 factory classes.
_— Qwen Code /review_
--
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]
