[Bug][CDC][Postgres] Postgres-CDC fails with UnsupportedOperationException: Unsupported type TIMESTAMP_TZ for TIMESTAMP WITH TIME ZONE columns

Search before asking

• I had searched in the issues and found no similar issues.

What happened

When using Postgres-CDC on a table containing a PostgreSQL TIMESTAMP WITH TIME ZONE (TIMESTAMPTZ) column, SeaTunnel fails during source initialization.

The job never starts processing records and fails while building the CDC deserialization schema.

Actual behavior:

Source initialization fails with:

java.lang.UnsupportedOperationException: Unsupported type: TIMESTAMP_TZ Stack Trace Caused by: java.lang.UnsupportedOperationException: Unsupported type: TIMESTAMP_TZ at org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializationConverters.createNotNullConverter(SeaTunnelRowDebeziumDeserializationConverters.java:185) at org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializationConverters.createConverter(SeaTunnelRowDebeziumDeserializationConverters.java:119) at org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializeSchema.createTableRowConverters(SeaTunnelRowDebeziumDeserializeSchema.java:349) at org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializeSchema.(SeaTunnelRowDebeziumDeserializeSchema.java:94) at org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source.PostgresIncrementalSource.createDebeziumDeserializationSchema(PostgresIncrementalSource.java:113)

SeaTunnel Version

2.3.13

SeaTunnel Config

env {
  parallelism = 1
  job.mode = "STREAMING"
}

source {
  Postgres-CDC {
    plugin_output = "source_data"
    base-url = "jdbc:postgresql://<host>:5432/<database>"
    username = "<username>"
    password = "<password>"

    database-names = ["postgres"]
    schema-names = ["public"]
    table-names = ["public.cdc_timestamp_test"]

    slot.name = "seatunnel_timestamp_test"
  }
}

sink {
  Console {}
}

Running Command

cd {seatunnel_path}/ 
./bin/seatunnel.sh --config ./config/new_config/postgres_cdc_test -n test1

Error Exception

Caused by: java.lang.UnsupportedOperationException: Unsupported type: TIMESTAMP_TZ
        at org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializationConverters.createNotNullConverter(SeaTunnelRowDebeziumDeserializationConverters.java:185)
        at org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializationConverters.createConverter(SeaTunnelRowDebeziumDeserializationConverters.java:119)
        at org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializationConverters.lambda$new$0(SeaTunnelRowDebeziumDeserializationConverters.java:77)
        at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
        at java.base/java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:1024)
        at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
        at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
        at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:575)
        at java.base/java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260)
        at java.base/java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:616)
        at org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializationConverters.<init>(SeaTunnelRowDebeziumDeserializationConverters.java:79)
        at org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializeSchema.createTableRowConverters(SeaTunnelRowDebeziumDeserializeSchema.java:349)
        at org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializeSchema.<init>(SeaTunnelRowDebeziumDeserializeSchema.java:94)
        at org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializeSchema$Builder.build(SeaTunnelRowDebeziumDeserializeSchema.java:374)
        at org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source.PostgresIncrementalSource.createDebeziumDeserializationSchema(PostgresIncrementalSource.java:113)
        at org.apache.seatunnel.connectors.cdc.base.source.IncrementalSource.<init>(IncrementalSource.java:127)
        at org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source.PostgresIncrementalSource.<init>(PostgresIncrementalSource.java:63)
        at org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source.PostgresIncrementalSourceFactory.lambda$createSource$1(PostgresIncrementalSourceFactory.java:110)
        at org.apache.seatunnel.api.table.factory.FactoryUtil.createAndPrepareSource(FactoryUtil.java:180)
        at org.apache.seatunnel.api.table.factory.FactoryUtil.restoreAndPrepareSource(FactoryUtil.java:136)
        ... 8 more

Zeta or Flink or Spark Version

No response

Java or Scala Version

No response

Screenshots

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct