Kafka-connect无法发送非结构化JSON数据

在 Kafka-connect 配置文件中指定数据格式为 JSON，并在数据中添加 schema

配置文件示例：

{
  "name": "my-connector",
  "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
  "tasks.max": "1",
  "connection.url": "jdbc:mysql://localhost:3306/my_db",
  "table.whitelist": "my_table",
  "mode": "bulk",
  "topic.prefix": "my_topic",
  "key.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "key.converter.schemas.enable": "false",
  "value.converter.schemas.enable": "true"
}

在配置文件中，我们指定了 key 和 value 的转换器为 JsonConverter，并开启了 value 的 schema。在发送数据时，我们需要保证数据格式为 JSON，并正确地添加 schema 信息。示例代码如下：

{
  "name": "my_table",
  "schema": {
    "type": "struct",
    "fields": [
        {"name": "id", "type": "int32"},
        {"name": "name", "type": "string"},
        {"name": "age", "type": "int32"}
    ]
  },
  "payload": {"id": 1, "name": "John", "age": 30}
}

以上代码中，我们定义了一个包含 schema 的结构体，其中包含三个字段：id、name、和 age。我们要发送的数据也符合这个 schema，因此可以成功地将其发送到 Kafka。