[Improve][CDC] Set temporalAdjuster at defaultValue process

apache · Apr 25, 2024 · f16ca51 · f16ca51
1 parent a6941ec
commit f16ca51
Showing 1 changed file with 7 additions and 1 deletion.
diff --git a/...c/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlTypeUtils.java b/...c/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlTypeUtils.java
@@ -45,6 +45,10 @@ public static org.apache.seatunnel.api.table.catalog.Column convertToSeaTunnelCo
                 dbzConnectorConfig
                         .getConfig()
                         .getString(MySqlConnectorConfig.BIGINT_UNSIGNED_HANDLING_MODE);
+        final boolean timeAdjusterEnabled =
+                dbzConnectorConfig
+                        .getConfig()
+                        .getBoolean(MySqlConnectorConfig.ENABLE_TIME_ADJUSTER);
         MySqlConnectorConfig.BigIntUnsignedHandlingMode bigIntUnsignedHandlingMode =
                 MySqlConnectorConfig.BigIntUnsignedHandlingMode.parse(
                         bigIntUnsignedHandlingModeStr);
@@ -53,7 +57,9 @@ public static org.apache.seatunnel.api.table.catalog.Column convertToSeaTunnelCo
                         dbzConnectorConfig.getDecimalMode(),
                         dbzConnectorConfig.getTemporalPrecisionMode(),
                         bigIntUnsignedHandlingMode.asBigIntUnsignedMode(),
-                        dbzConnectorConfig.binaryHandlingMode());
+                        dbzConnectorConfig.binaryHandlingMode(),
+                        timeAdjusterEnabled ? MySqlValueConverters::adjustTemporal : (x) -> x,
+                        MySqlValueConverters::defaultParsingErrorHandler);
         MySqlDefaultValueConverter mySqlDefaultValueConverter =
                 new MySqlDefaultValueConverter(mySqlValueConverters);
         Object defaultValue =