lvyanquan commented on code in PR #4214: URL: https://github.com/apache/flink-cdc/pull/4214#discussion_r4091670476
########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/CustomPostgresValueConverter.java: ########## @@ -0,0 +1,113 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.debezium.connector.postgresql; + +import io.debezium.config.CommonConnectorConfig; +import io.debezium.jdbc.TemporalPrecisionMode; +import io.debezium.relational.Column; +import org.apache.kafka.connect.data.Field; + +import java.nio.charset.Charset; +import java.sql.Timestamp; +import java.time.ZoneOffset; + +/** + * A {@link PostgresValueConverter} that preserves the wall-clock fields of PostgreSQL {@code + * timestamp} values. + * + * <p>This converter only changes {@code timestamp} (without time zone) conversion. Timestamp values + * with time zone continue to use the parent implementation. The constructor and factory mirror the + * Debezium implementation and must remain aligned with it when the Debezium version changes. + */ +public class CustomPostgresValueConverter extends PostgresValueConverter { + protected CustomPostgresValueConverter( + Charset databaseCharset, + DecimalMode decimalMode, + TemporalPrecisionMode temporalPrecisionMode, + ZoneOffset defaultOffset, + BigIntUnsignedMode bigIntUnsignedMode, + boolean includeUnknownDatatypes, + TypeRegistry typeRegistry, + PostgresConnectorConfig.HStoreHandlingMode hStoreMode, + CommonConnectorConfig.BinaryHandlingMode binaryMode, + PostgresConnectorConfig.IntervalHandlingMode intervalMode, + byte[] toastPlaceholder, + int moneyFractionDigits) { + super( + databaseCharset, + decimalMode, + temporalPrecisionMode, + defaultOffset, + bigIntUnsignedMode, + includeUnknownDatatypes, + typeRegistry, + hStoreMode, + binaryMode, + intervalMode, + toastPlaceholder, + moneyFractionDigits); + } + + public static CustomPostgresValueConverter of( + PostgresConnectorConfig connectorConfig, + Charset databaseCharset, + TypeRegistry typeRegistry) { + return new CustomPostgresValueConverter( + databaseCharset, + connectorConfig.getDecimalMode(), + connectorConfig.getTemporalPrecisionMode(), + ZoneOffset.UTC, + null, + connectorConfig.includeUnknownDatatypes(), + typeRegistry, + connectorConfig.hStoreHandlingMode(), + connectorConfig.binaryHandlingMode(), + connectorConfig.intervalHandlingMode(), + connectorConfig.getUnavailableValuePlaceholder(), + connectorConfig.moneyFractionDigits()); + } + + /** + * Converts a PostgreSQL {@code timestamp} without passing through an instant and the JVM + * default time zone. + * + * <p>Debezium's implementation converts through {@link Timestamp#toInstant()}, which can apply + * inconsistent historical offsets to values before 1970 in regional time zones. The column and + * field definitions are intentionally unused because this conversion depends only on the JDBC + * value. + */ + @Override + protected Object convertTimestampToLocalDateTime(Column column, Field fieldDefn, Object data) { + if (data == null) { + return null; + } + if (!(data instanceof Timestamp)) { + return data; + } + final Timestamp timestamp = (Timestamp) data; + + if (POSITIVE_INFINITY_TIMESTAMP.equals(timestamp)) { + return POSITIVE_INFINITY_LOCAL_DATE_TIME; + } else if (NEGATIVE_INFINITY_TIMESTAMP.equals(timestamp)) { + return NEGATIVE_INFINITY_LOCAL_DATE_TIME; + } + + // Preserve the timestamp's wall-clock fields instead of applying historical zone offsets. + return timestamp.toLocalDateTime(); Review Comment: Good point, agreed — a fix in a minor release shouldn't silently change the output of jobs that already work. Added your option `scan.pre-epoch-timestamp.wall-clock-conversion.enabled` (default `false` = previous behavior kept, `true` = the wall clock stored in PostgreSQL is kept). It works in the SQL source, the DataStream source (same key in `debeziumProperties`) and the Postgres pipeline connector; en/zh docs updated. A unit test asserts that the default path still returns exactly what a stock `PostgresValueConverter` returns, and ITCases pin both modes to `Asia/Shanghai`. One limitation, now documented: the option only affects the incremental snapshot source. With `scan.incremental.snapshot.enabled=false` the record values are converted by Debezium's own `PostgresConnectorTask`, so the option is a no-op there — covering that would require shadowing more Debezium classes. Happy to include it here if you prefer. -- 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]
