我正在使用jdbc源连接器从mysql获取数据，并将其放入kafka主题中。这是我的配置

{
"name": "test",
"config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:mysql://localhost:3306/sample",
    "connection.user": "root",
    "connection.password": "root@123",
    "topic.prefix": "",
    "poll.interval.ms": 3600000,
    "table.whitelist": "test",
    "schemas.enable": "true",
    "mode": "bulk",
    "schema:registry.url":"http://localhost:8081/subjects/",

}

}
这是avro模式

{
  "fields": [
    {
      "name": "name",
      "type": "string"
    },
    {
      "name": "venue",
      "type": "string"
    },
    {
      "name": "age",
      "type": "int"
    },
    {
      "name": "sal",
      "type": "long"
    }
  ],
  "name": "TestClass",
  "namespace": "com.ns.vishnu.Kafka_avro.model",
  "type": "record"
}

我必须从数据库中获取数据，并检查注册表中的模式，然后只需要放入主题！！
有人知道怎么做吗？