[FLINK-38742][cdc/postgres] Fix integration tests to expect TIMESTAMP_TZ instead of TIMESTAMP_LTZ

Update PostgresFullTypesITCase to expect TIMESTAMP_TZ for PostgreSQL
TIMESTAMPTZ columns, matching the corrected schema inference behavior.

Author: Fluder-Paradyne

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresFullTypesITCase.java

@@ -23,10 +23,10 @@
 import org.apache.flink.cdc.common.data.ArrayData;
 import org.apache.flink.cdc.common.data.DateData;
 import org.apache.flink.cdc.common.data.DecimalData;
-import org.apache.flink.cdc.common.data.LocalZonedTimestampData;
 import org.apache.flink.cdc.common.data.RecordData;
 import org.apache.flink.cdc.common.data.TimeData;
 import org.apache.flink.cdc.common.data.TimestampData;
+import org.apache.flink.cdc.common.data.ZonedTimestampData;
 import org.apache.flink.cdc.common.data.binary.BinaryMapData;
 import org.apache.flink.cdc.common.data.binary.BinaryStringData;
 import org.apache.flink.cdc.common.event.CreateTableEvent;
@@ -68,6 +68,7 @@
 import java.time.Instant;
 import java.time.LocalDateTime;
 import java.time.LocalTime;
+import java.time.OffsetDateTime;
 import java.time.ZoneId;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -298,7 +299,8 @@ public void testTimeTypesWithTemporalModeAdaptive() throws Exception {
                     TimestampData.fromLocalDateTime(
                             LocalDateTime.parse("2020-07-17T18:00:22.123456")),
                     TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-07-17T18:00:22")),
-                    LocalZonedTimestampData.fromInstant(toInstant("2020-07-17 18:00:22")),
+                    ZonedTimestampData.fromOffsetDateTime(
+                            OffsetDateTime.parse("2020-07-17T10:00:22Z")),
                 };
 
         List<Event> snapshotResults = fetchResultsAndCreateTableEvent(events, 1).f0;
@@ -354,7 +356,8 @@ public void testTimeTypesWithTemporalModeMicroSeconds() throws Exception {
                     TimestampData.fromLocalDateTime(
                             LocalDateTime.parse("2020-07-17T18:00:22.123456")),
                     TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-07-17T18:00:22")),
-                    LocalZonedTimestampData.fromInstant(toInstant("2020-07-17 18:00:22")),
+                    ZonedTimestampData.fromOffsetDateTime(
+                            OffsetDateTime.parse("2020-07-17T10:00:22Z")),
                 };
 
         List<Event> snapshotResults = fetchResultsAndCreateTableEvent(events, 1).f0;
@@ -409,7 +412,8 @@ public void testTimeTypesWithTemporalModeConnect() throws Exception {
                     TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-07-17T18:00:22.123")),
                     TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-07-17T18:00:22.123")),
                     TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-07-17T18:00:22")),
-                    LocalZonedTimestampData.fromInstant(toInstant("2020-07-17 18:00:22")),
+                    ZonedTimestampData.fromOffsetDateTime(
+                            OffsetDateTime.parse("2020-07-17T10:00:22Z")),
                 };
 
         List<Event> snapshotResults = fetchResultsAndCreateTableEvent(events, 1).f0;
@@ -1042,7 +1046,7 @@ private Instant toInstant(String ts) {
                     DataTypes.TIMESTAMP(3),
                     DataTypes.TIMESTAMP(6),
                     DataTypes.TIMESTAMP(),
-                    DataTypes.TIMESTAMP_LTZ(0));
+                    DataTypes.TIMESTAMP_TZ(0));
 
     private static final RowType HSTORE_TYPES_WITH_ADAPTIVE =
             RowType.of(DataTypes.INT(), DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()));

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java

@@ -377,6 +377,9 @@ protected Object convertToZonedTimestamp(Object dbzObj, Schema schema) {
             String str = (String) dbzObj;
             // ZonedTimestamp type is encoded in string type with timezone offset
             // Format: ISO-8601 with timezone offset (e.g., "2020-07-17T18:00:22+00:00")
+            // According to Debezium documentation, PostgreSQL TIMESTAMPTZ is ALWAYS encoded as
+            // String
+            // with ZonedTimestamp.SCHEMA_NAME, regardless of time.precision.mode
             if (ZonedTimestamp.SCHEMA_NAME.equals(schema.name())) {
                 // Parse using Debezium's ZonedTimestamp formatter
                 OffsetDateTime offsetDateTime = OffsetDateTime.parse(str, ZonedTimestamp.FORMATTER);
@@ -391,7 +394,11 @@ protected Object convertToZonedTimestamp(Object dbzObj, Schema schema) {
                 "Unable to convert to TIMESTAMP WITH TIME ZONE from unexpected value '"
                         + dbzObj
                         + "' of type "
-                        + dbzObj.getClass().getName());
+                        + dbzObj.getClass().getName()
+                        + " with schema name '"
+                        + (schema != null ? schema.name() : "null")
+                        + "'. PostgreSQL TIMESTAMPTZ should always be encoded as String with "
+                        + ZonedTimestamp.SCHEMA_NAME);
     }
 
     protected Object convertToLocalTimeZoneTimestamp(Object dbzObj, Schema schema) {