Debezium+Schema Registry下JdbcSink自动建表数据类型不匹配问题

源PostgreSQL表DDL #

Table "public.tbl1"
 Column |            Type             | Collation | Nullable |             Default              | Storage  | Compression | Stats target | Description
--------+-----------------------------+-----------+----------+----------------------------------+----------+-------------+--------------+-------------
 id     | integer                     |           | not null | nextval('tbl1_id_seq'::regclass) | plain    |             |              |
 name   | character varying(100)      |           |          |                                  | extended |             |              |
 col4   | numeric                     |           |          |                                  | main     |             |              |
 col5   | bigint                      |           |          |                                  | plain    |             |              |
 col6   | timestamp without time zone |           |          |                                  | plain    |             |              |
 col7   | timestamp with time zone    |           |          |                                  | plain    |             |              |
 col8   | boolean                     |           |          |                                  | plain    |             |              |
Indexes:
    "tbl1_pkey" PRIMARY KEY, btree (id)
Publications:
    "dbz_publication"
Access method: heap

Schema Registry中的Avro Schema #

{
  "type": "record",
  "name": "Value",
  "namespace": "test.public.tbl1",
  "fields": [
    {
      "name": "id",
      "type": {
        "type": "int",
        "connect.parameters": {
          "__debezium.source.column.type": "SERIAL",
          "__debezium.source.column.length": "10",
          "__debezium.source.column.scale": "0"
        },
        "connect.default": 0
      },
      "default": 0
    },
    {
      "name": "name",
      "type": [
        "null",
        {
          "type": "string",
          "connect.parameters": {
            "__debezium.source.column.type": "VARCHAR",
            "__debezium.source.column.length": "100",
            "__debezium.source.column.scale": "0"
          }
        }
      ],
      "default": null
    },
    {
      "name": "col4",
      "type": [
        "null",
        {
          "type": "double",
          "connect.parameters": {
            "__debezium.source.column.type": "NUMERIC",
            "__debezium.source.column.length": "0"
          }
        }
      ],
      "default": null
    },
    {
      "name": "col5",
      "type": [
        "null",
        {
          "type": "long",
          "connect.parameters": {
            "__debezium.source.column.type": "INT8",
            "__debezium.source.column.length": "19",
            "__debezium.source.column.scale": "0"
          }
        }
      ],
      "default": null
    },
    {
      "name": "col6",
      "type": [
        "null",
        {
          "type": "long",
          "connect.version": 1,
          "connect.parameters": {
            "__debezium.source.column.type": "TIMESTAMP",
            "__debezium.source.column.length": "29",
            "__debezium.source.column.scale": "6"
          },
          "connect.name": "io.debezium.time.MicroTimestamp"
        }
      ],
      "default": null
    },
    {
      "name": "col7",
      "type": [
        "null",
        {
          "type": "string",
          "connect.version": 1,
          "connect.parameters": {
            "__debezium.source.column.type": "TIMESTAMPTZ",
            "__debezium.source.column.length": "35",
            "__debezium.source.column.scale": "6"
          },
          "connect.name": "io.debezium.time.ZonedTimestamp"
        }
      ],
      "default": null
    },
    {
      "name": "col8",
      "type": [
        "null",
        {
          "type": "boolean",
          "connect.parameters": {
            "__debezium.source.column.type": "BOOL",
            "__debezium.source.column.length": "1",
            "__debezium.source.column.scale": "0"
          }
        }
      ],
      "default": null
    }
  ],
  "connect.name": "test.public.tbl1.Value"
}

目标PostgreSQL表DDL #

Table "public.tbl1"
 Column |       Type       | Collation | Nullable | Default | Storage  | Compression | Stats target | Description
--------+------------------+-----------+----------+---------+----------+-------------+--------------+-------------
 id     | text             |           | not null |         | extended |             |              |
 name   | text             |           |          |         | extended |             |              |
 col4   | double precision |           |          |         | plain    |             |              |
 col5   | bigint           |           |          |         | plain    |             |              |
 col6   | bigint           |           |          |         | plain    |             |              |
 col7   | text             |           |          |         | extended |             |              |
 col8   | boolean          |           |          |         | plain    |             |              |
Indexes:
    "tbl1_pkey" PRIMARY KEY, btree (id)
Access method: heap

JDBC Sink连接器配置 #

{
        "name": "t1-sink",
        "config": {
                "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
                "tasks.max": "1",
                "topics": "test.public.tbl1",
                "connection.url": "jdbc:postgresql://172.31.85.***:5432/test",
                "connection.user": "postgres",
                "connection.password": "***",
                "dialect.name": "PostgreSqlDatabaseDialect",
                "auto.create": "true",
                "insert.mode": "upsert",
                "delete.enabled": "true",
                "pk.fields": "id",
                "pk.mode": "record_key",
                "table.name.format": "tbl1",
                "key.converter": "org.apache.kafka.connect.storage.StringConverter",
                "key.converter.schemas.enable": "false",
                "internal.key.converter": "com.amazonaws.services.schemaregistry.kafkaconnect.AWSKafkaAvroConverter",
                "internal.key.converter.schemas.enable": "true",
                "internal.value.converter": "com.amazonaws.services.schemaregistry.kafkaconnect.AWSKafkaAvroConverter",
                "internal.value.converter.schemas.enable": "true",
                "value.converter": "com.amazonaws.services.schemaregistry.kafkaconnect.AWSKafkaAvroConverter",
                "value.converter.schemas.enable": "true",
                "value.converter.region": "us-east-1",
                "key.converter.region": "us-east-1",
                "key.converter.schemaAutoRegistrationEnabled": "true",
                "value.converter.schemaAutoRegistrationEnabled": "true",
                "key.converter.avroRecordType": "GENERIC_RECORD",
                "value.converter.avroRecordType": "GENERIC_RECORD",
                "key.converter.registry.name": "bhuvi-debezium",
                "value.converter.registry.name": "bhuvi-debezium",
                "value.converter.column.propagate.source.type": ".*",
                "value.converter.datatype.propagate.source.type": ".*"

        }
}

问题 #

使用PostgreSQL的Debezium源连接器，值转换器采用Avro并结合Schema Registry，但同步后目标PostgreSQL表的ID列、时间戳列（col6、col7）以及Decimal列（col4）的数据类型完全不匹配。想了解为何在使用Schema Registry的情况下，仍无法创建具有正确数据类型的目标表。

内容的提问来源于stack exchange，提问作者TheDataGuy