Error: Kafka target database批量录入错误（主键冲突为Null值）

Question

错误：批处理条目 0 INSERT INTO "person" ("ID","NAME") VALUES (NULL,'Alex') ON CONFLICT ("ID") DO UPDATE SET "NAME"=EXCLUDED."NAME" 被中止.调用 getNextException 查看原因。

来源：甲骨文数据库

目标：Postgres

使用Kafka复制数据

问题：接收器连接器在目标数据库中插入 ID 的NULL 值导致冲突。

源连接器：

{
    "name": "source",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "connection.url": "jdbc:oracle:thin:@192.168.91.138:1521/orcl1",
        "connection.user": "sys as sysdba",
        "connection.password": "oracle",
        "topic.prefix": "person",
        "mode": "incrementing",
        "poll.interval.ms": "1000",
        "incrementing.column.name":"ID",
        "query": "SELECT * from person",
        "numeric.mapping":"none",
        "include.schema.changes": "true",
        "validate.non.null": "false",
        "value.converter.schemas.enable": "true",
        "key.converter":"org.apache.kafka.connect.storage.StringConverter",
        "value.converter":"io.confluent.connect.avro.AvroConverter",
        "value.converter.schema.registry.url":"http://localhost:8081"
    }

}

水槽连接器：

{
    "name": "jdbc-sink",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "tasks.max": "1",
        "topics":"person",
        "connection.url": "jdbc:postgresql://192.168.91.229:5432/postgres?user=postgres&password=postgres ",
        "auto.create": "true",
        "insert.mode": "upsert",
        "pk.mode": "record_key",
        "pk.fields": "ID",
        "delete.enabled": "true",
        "key.converter":"org.apache.kafka.connect.storage.StringConverter",
        "value.converter":"io.confluent.connect.avro.AvroConverter",
        "value.converter.schema.registry.url":"http://localhost:8081"
    }
    }