wenshao commented on code in PR #4320:
URL: https://github.com/apache/flink-cdc/pull/4320#discussion_r3068994180
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTableFactory.java:
##########
@@ -232,13 +240,24 @@ private static StartupOptions
getStartupOptions(ReadableConfig config) {
case SCAN_STARTUP_MODE_VALUE_LATEST:
return StartupOptions.latest();
+ case SCAN_STARTUP_MODE_VALUE_TIMESTAMP:
+ Long startupTimestampMillis =
config.get(SCAN_STARTUP_TIMESTAMP_MILLIS);
+ if (startupTimestampMillis == null) {
+ throw new ValidationException(
+ String.format(
+ "To use timestamp startup mode, the
startup timestamp millis '%s' must be set.",
+ SCAN_STARTUP_TIMESTAMP_MILLIS.key()));
+ }
+ return StartupOptions.timestamp(startupTimestampMillis);
+
Review Comment:
**[Critical]** `scan.startup.mode=timestamp` is accepted here, but the
non-parallel execution path still delegates to
`SqlServerSource.Builder.build()`, whose startup-mode switch only supports
`INITIAL` and `LATEST_OFFSET` and otherwise throws
`UnsupportedOperationException`. As a result, a configuration that passes
validation can still fail at runtime when `scan.incremental.snapshot.enabled`
remains `false`. Please either reject `timestamp` mode for the non-parallel
path or add matching runtime support there.
_— gpt-5.4 via Qwen Code /review_
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/offset/LsnFactory.java:
##########
@@ -52,7 +78,68 @@ public Offset newOffset(Long position) {
@Override
public Offset createTimestampOffset(long timestampMillis) {
- throw new UnsupportedOperationException("not supported create new
Offset by timestamp.");
+ if (sourceConfig == null) {
+ throw new UnsupportedOperationException(
+ "Timestamp startup mode requires SqlServerSourceConfig in
LsnFactory.");
+ }
+ List<String> databaseList = sourceConfig.getDatabaseList();
+ if (databaseList == null || databaseList.isEmpty()) {
+ throw new IllegalStateException(
+ "Cannot create timestamp offset without configured SQL
Server database.");
+ }
+ final String databaseName = databaseList.get(0);
+ final ZoneId serverZoneId =
ZoneId.of(sourceConfig.getServerTimeZone());
+ final Timestamp startupTimestamp =
+ Timestamp.valueOf(
+ Instant.ofEpochMilli(timestampMillis)
+ .atZone(serverZoneId)
+ .toLocalDateTime());
Review Comment:
**[Critical]** `startupTimestamp` is converted using `serverTimeZone`, but
it is then bound with `PreparedStatement#setTimestamp(...)` without an explicit
`Calendar`. JDBC will interpret that timestamp in the JVM default timezone
instead of the configured SQL Server timezone, so `scan.startup.mode=timestamp`
can resolve to the wrong LSN when those timezones differ. Please bind the query
parameter with a timezone-aware `Calendar` (or otherwise preserve the
configured server timezone when passing the timestamp to JDBC).
_— gpt-5.4 via 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]
