問題

• 消費(fèi)端連接器的配置

[root@docker ~]# cat oracle-jdbc-sink.json
{
    "name": "oracle-datatype-sink",
        "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "connection.url": "jdbc:oracle:thin:@//172.17.0.2:1521/pdbtt",
        "connection.user": "test",
        "connection.password": "test",
        "tasks.max": "1",
        "topics.regex": "data_type.test.(.*)",
        "dialect.name": "OracleDatabaseDialect",
        "transforms": "unwrap,dropPrefix",
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
        "transforms.unwrap.drop.tombstones": "false",
        "transforms.dropPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.dropPrefix.regex":"data_type.test.(.*)",
        "transforms.dropPrefix.replacement":"$1",
        "auto.create": "true",
        "auto.evolve": "true",
        "delete.enabled": "true",
        "pk.mode": "record_key"
        }
}

curl -s -X DELETE localhost:8083/connectors/oracle-datatype-sink
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" 192.168.0.40:8083/connectors/ -d @oracle-jdbc-sink.json

• 注冊(cè)消費(fèi)端連接器后會(huì)在目標(biāo)庫自動(dòng)建表并導(dǎo)入數(shù)據(jù)，但是建表的時(shí)候出現(xiàn)以下錯(cuò)誤

Error: Sink connector 'oracle-datatype-sink2' is configured with 'delete.enabled=true' and 'pk.mode=record_key' 
and therefore requires records with a non-null key and non-null Struct or primitive key schema, 
but found record at (topic='tt_week',partition=0,offset=0,timestamp=1650770878211) with a HashMap key and null key schema.

嘗試處理

• 啟用 JSON 序列化的 schema

[root@docker ~]# docker inspect connect |grep "/kafka/config" -B1
                "Source": "/var/lib/docker/volumes/8905c22d93450573a179ec357a28dcd1051e658460388fbd6d61a419bbf6cf76/_data",
                "Destination": "/kafka/config",

[root@docker ~]# cd /var/lib/docker/volumes/8905c22d93450573a179ec357a28dcd1051e658460388fbd6d61a419bbf6cf76/_data
[root@docker _data]# vi connect-distributed.properties
key.converter.schemas.enable=true
value.converter.schemas.enable=true
[root@docker ~]# docker restart connect

• 重新初始化數(shù)據(jù)到 Kafka 再消費(fèi)到目標(biāo)庫

curl -s -X DELETE localhost:8083/connectors/pgsql_datatype_test
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" 192.168.0.40:8083/connectors/ -d @pgsql_test_2.json

curl -s -X DELETE localhost:8083/connectors/oracle-datatype-sink
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" 192.168.0.40:8083/connectors/ -d @oracle-jdbc-sink.json

• 沒有報(bào)錯(cuò)了