1.创建新的source端connector
curl -s -X POST -H "Content-Type: application/json" -i 'http://localhost:8083/connectors' --data
'{"name": "test-oracle-kafka-connector",
"config":
{"connector.class":"io.confluent.connect.jdbc.JdbcSourceConnector",
"connection.url":"jdbc:oracle:thin:@localhost:1521:××××××",
"connection.user":"nk_zffx",
"connection.password":"××××××",
"table.whitelist":"TEST_USERS",
"mode":"timestamp",
"timestamp.column.name":"CREATE_TIME",
"topic.prefix":"my_oracle_",
"poll.interval.ms":"5000",
"validate.non.null":"false",
"timestamp.initial":"1593532297000",
"db.timezone":"Asia/Shanghai"
}
}'
2.创建新的Sink端connector
curl -s -X POST -H "Content-Type: application/json" -i 'http://localhost:8083/connectors' --data
'{"name": "test-oracle-kafka-connector_sink2",
"config":
{"connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector",
"connection.url":"jdbc:oracle:thin:@localhost:1521:orcl",
"connection.user":"nk_zffx",
"connection.password":"password",
"tasks.max": "1",
"topics":"my_oracle_TEST_USERS",
"table.name.format":"TEST_USERS_BAK",
"auto.create": false,
"auto.evolve": false,
"insert.mode": "upsert",
"batch.size": 3000,
"delete.enabled": false,
"pk.mode":"record_value",
"pk.fields":"USER_UUID"
}
}'
3.删除连接器
curl -X DELETE http://localhost:8083/connectors/<connector-name>