Kafka使用JSON序列化关闭schema后在消费端产生的一个问题

问题

• 消费端连接器的配置

[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

• 注册消费端连接器后会在目标库自动建表并导入数据，但是建表的时候出现以下错误

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

• 重新初始化数据到 Kafka 再消费到目标库

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

• 没有报错了