Daishuyuan commented on code in PR #4320:
URL: https://github.com/apache/flink-cdc/pull/4320#discussion_r3071544600
##########
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:
Thanks for catching this. I fixed it in the latest update.
`scan.startup.mode=timestamp` is now explicitly validated together with
`scan.incremental.snapshot.enabled`, and the factory throws a
`ValidationException` for the legacy non-parallel path instead of letting it
fail later at runtime. I also added a test to cover that `timestamp` startup
requires incremental snapshot mode.
##########
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:
Thanks for pointing this out. I fixed it in the latest update.
The timestamp is now bound with an explicit `Calendar` created from the
configured `serverTimeZone`, so the JDBC parameter keeps the SQL Server
timezone semantics instead of falling back to the JVM default timezone. I also
added an integration test that sets the JVM default timezone differently from
the configured server timezone to verify the timestamp startup path.
--
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]
