Kafka连接JDBC错误，尽管给出了模式和有效负载

Question

当我运行kafka JDBC连接器到PSQL时，我收到以下错误：

JSON converter with schemas . enable需要“schema”和“payload”字段，不能包含其他字段。如果试图反序列化普通JSON数据，请在转换器配置中设置schemas.enable=false。

但是，我的主题包含以下消息结构，并添加了一个架构，就像在线显示的那样：

Rowtime： 2022/02/04 12:45:48.520 Z， key：， value：{"架构"：{"类型"："结构"，"字段"：[{"类型"："int"，"字段"："ID"，"可选"： false}，{"类型"："日期"，"字段"："日期"，"可选"： false}，{"类型"："varchar"，"字段"："ICD"，"可选"： false}，{"类型"："int"，"字段"："CPT"，"可选"： false}，{"类型"："双"，"字段"："成本"，"可选"： false}]，"可选"： false，"名称"："test"}，"有效载荷"：{"ID"："24427934"，"日期"："2019-05-22"，"ICD"："883.436"，"CPT"："60502"，"成本"："1374.36"}}"，分区： 0

我对连接器的配置是:

 curl -X PUT http://localhost:8083/connectors/claim_test/config \
    -H "Content-Type: application/json" \
    -d '{
     "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
     "connection.url":"jdbc:postgresql://localhost:5432/ae2772",
     "key.converter":"org.apache.kafka.connect.json.JsonConverter",
     "value.converter":"org.apache.kafka.connect.json.JsonConverter",
     "value.converter.schemas.enable":"true",
     "topics":"test_7",
     "auto.create":"true",
     "insert.mode":"insert"
    }'

经过一些更改后，我现在得到以下消息：

WorkerSinkTask｛id=claim_test｝在偏移量0和时间戳1644005137197处转换主题“test_9”分区0中的消息值时出错：未知架构类型：int

Answer 1

int 不是有效的架构类型。应为 int8、int16、int32 或 int64。

同样，date、varchar和Double也无效。

JSON中使用的类型不同于Postgres或任何特定于SQL的类型(日期应该转换为Unix Epoch int64时间或成为< code>string)。

您可以在这里找到支持的模式类型:https://github . com/Apache/Kafka/blob/trunk/connect/API/src/main/Java/org/Apache/Kafka/connect/data/schema . Java