Copilot commented on code in PR #4214: URL: https://github.com/apache/flink-cdc/pull/4214#discussion_r2697883457
########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/CustomPostgresValueConverter.java: ########## @@ -0,0 +1,99 @@ +/* + * 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 custom PostgresValueConverter that correctly handles timestamp conversion to LocalDateTime for + * dates before 1970-01-01. + */ +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()); + } + + @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; + } + + return timestamp.toLocalDateTime(); + } Review Comment: The CustomPostgresValueConverter class lacks dedicated unit tests. While integration tests are added, unit tests should verify the convertTimestampToLocalDateTime method's behavior with various edge cases including null values, non-Timestamp objects, infinity timestamps, and crucially, timestamps both before and after 1970-01-01 to ensure the timezone handling fix works correctly. ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/CustomPostgresValueConverter.java: ########## @@ -0,0 +1,99 @@ +/* + * 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 custom PostgresValueConverter that correctly handles timestamp conversion to LocalDateTime for + * dates before 1970-01-01. + */ +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()); + } + + @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; + } + + return timestamp.toLocalDateTime(); Review Comment: The overridden method doesn't use the column and fieldDefn parameters. If the parent class implementation uses these parameters for specific logic (such as handling different column types or field definitions), removing this logic could cause unexpected behavior. Consider adding a comment explaining why these parameters can be safely ignored in this implementation. ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLConnectorITCase.java: ########## @@ -689,18 +689,21 @@ void testAllTypes(boolean parallelismSnapshot) throws Throwable { // generate WAL try (Connection connection = getJdbcConnection(POSTGIS_CONTAINER); Statement statement = connection.createStatement()) { - statement.execute("UPDATE inventory.full_types SET small_c=0 WHERE id=1;"); + statement.execute("UPDATE inventory.full_types SET small_c=0 WHERE id<=2;"); } - waitForSinkSize("sink", 3); + waitForSinkSize("sink", 6); List<String> expected = Arrays.asList( "+I(1,[50],32767,65535,2147483647,5.5,6.6,123.12345,404.4,true,Hello World,a,abc,abcd..xyz,2020-07-17T18:00:22.123,2020-07-17T18:00:22.123456,2020-07-17,18:00:22,500,{\"hexewkb\":\"0105000020e610000001000000010200000002000000a779c7293a2465400b462575025a46c0c66d3480b7fc6440c3d32b65195246c0\",\"srid\":4326},{\"hexewkb\":\"0101000020730c00001c7c613255de6540787aa52c435c42c0\",\"srid\":3187})", + "+I(2,[50],32767,65535,2147483647,5.5,6.6,123.12345,404.4,true,Hello World,a,abc,abcd..xyz,1900-01-01T00:00:00.123,1900-01-01T00:00:00.123456,1900-01-01,18:00:22,500,{\"hexewkb\":\"0105000020e610000001000000010200000002000000a779c7293a2465400b462575025a46c0c66d3480b7fc6440c3d32b65195246c0\",\"srid\":4326},{\"hexewkb\":\"0101000020730c00001c7c613255de6540787aa52c435c42c0\",\"srid\":3187})", "-D(1,[50],32767,65535,2147483647,5.5,6.6,123.12345,404.4,true,Hello World,a,abc,abcd..xyz,2020-07-17T18:00:22.123,2020-07-17T18:00:22.123456,2020-07-17,18:00:22,500,{\"hexewkb\":\"0105000020e610000001000000010200000002000000a779c7293a2465400b462575025a46c0c66d3480b7fc6440c3d32b65195246c0\",\"srid\":4326},{\"hexewkb\":\"0101000020730c00001c7c613255de6540787aa52c435c42c0\",\"srid\":3187})", - "+I(1,[50],0,65535,2147483647,5.5,6.6,123.12345,404.4,true,Hello World,a,abc,abcd..xyz,2020-07-17T18:00:22.123,2020-07-17T18:00:22.123456,2020-07-17,18:00:22,500,{\"hexewkb\":\"0105000020e610000001000000010200000002000000a779c7293a2465400b462575025a46c0c66d3480b7fc6440c3d32b65195246c0\",\"srid\":4326},{\"hexewkb\":\"0101000020730c00001c7c613255de6540787aa52c435c42c0\",\"srid\":3187})"); + "-D(2,[50],32767,65535,2147483647,5.5,6.6,123.12345,404.4,true,Hello World,a,abc,abcd..xyz,1900-01-01T00:00:00.123,1900-01-01T00:00:00.123456,1900-01-01,18:00:22,500,{\"hexewkb\":\"0105000020e610000001000000010200000002000000a779c7293a2465400b462575025a46c0c66d3480b7fc6440c3d32b65195246c0\",\"srid\":4326},{\"hexewkb\":\"0101000020730c00001c7c613255de6540787aa52c435c42c0\",\"srid\":3187})", + "+I(1,[50],0,65535,2147483647,5.5,6.6,123.12345,404.4,true,Hello World,a,abc,abcd..xyz,2020-07-17T18:00:22.123,2020-07-17T18:00:22.123456,2020-07-17,18:00:22,500,{\"hexewkb\":\"0105000020e610000001000000010200000002000000a779c7293a2465400b462575025a46c0c66d3480b7fc6440c3d32b65195246c0\",\"srid\":4326},{\"hexewkb\":\"0101000020730c00001c7c613255de6540787aa52c435c42c0\",\"srid\":3187})", + "+I(2,[50],0,65535,2147483647,5.5,6.6,123.12345,404.4,true,Hello World,a,abc,abcd..xyz,1900-01-01T00:00:00.123,1900-01-01T00:00:00.123456,1900-01-01,18:00:22,500,{\"hexewkb\":\"0105000020e610000001000000010200000002000000a779c7293a2465400b462575025a46c0c66d3480b7fc6440c3d32b65195246c0\",\"srid\":4326},{\"hexewkb\":\"0101000020730c00001c7c613255de6540787aa52c435c42c0\",\"srid\":3187})"); List<String> actual = TestValuesTableFactory.getRawResultsAsStrings("sink"); - Assertions.assertThat(actual).isEqualTo(expected); + Assertions.assertThat(actual).containsExactlyInAnyOrderElementsOf(expected); Review Comment: The change from isEqualTo to containsExactlyInAnyOrderElementsOf weakens the test by no longer verifying event ordering. While this is appropriate for parallel processing scenarios (as indicated by the parallelismSnapshot parameter), consider documenting why order verification was removed or adding a separate test that verifies correct ordering when parallelismSnapshot is false. ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/CustomPostgresValueConverter.java: ########## @@ -0,0 +1,99 @@ +/* + * 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 custom PostgresValueConverter that correctly handles timestamp conversion to LocalDateTime for + * dates before 1970-01-01. + */ +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()); + } + Review Comment: The overridden method convertTimestampToLocalDateTime should include a JavaDoc comment explaining why this override is necessary and how it differs from the parent implementation. Specifically, it should document that this method uses Timestamp.toLocalDateTime() to properly handle dates before 1970-01-01 by avoiding timezone offset inconsistencies. ```suggestion /** * Overrides {@link PostgresValueConverter#convertTimestampToLocalDateTime(Column, Field, Object)} * to use {@link Timestamp#toLocalDateTime()} directly. * * <p>The default implementation derives the {@code LocalDateTime} value using epoch-based * calculations, which can apply timezone offsets inconsistently for timestamps before * {@code 1970-01-01}. By delegating to {@code Timestamp.toLocalDateTime()}, this override * avoids those pre-epoch timezone offset inconsistencies while preserving the behavior for * timestamps on or after the epoch.</p> */ ``` -- 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]
