如何使用在docker上运行的debezium和confluent-sink connector将所有更改从源数据库复制到目标数据库

version: '3'
services:
  debezium:
    image: debezium/connect:1.6
    ports:
      - 8083:8083
    environment:
      - GROUP_ID=1
      - CONFIG_STORAGE_TOPIC=my-connect-configs
      - OFFSET_STORAGE_TOPIC=my-connect-offsets
      - STATUS_STORAGE_TOPIC=my-connect-statuses
      - BOOTSTRAP_SERVERS=kafka:9092
  confluent-sink:
    image: confluentinc/connect-base:6.2.1
    ports:
      - 8084:8084
    environment:
      - CONNECT_BOOTSTRAP_SERVERS=kafka:9092
      - CONNECT_REST_ADVERTISED_HOST_NAME=confluent-sink
      - CONNECT_REST_PORT=8084
      - CONNECT_GROUP_ID=1
      - CONNECT_CONFIG_STORAGE_TOPIC=my-connect-configs
      - CONNECT_OFFSET_STORAGE_TOPIC=my-connect-offsets
      - CONNECT_STATUS_STORAGE_TOPIC=my-connect-statuses
      - CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR=1
      - CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR=1
      - CONNECT_STATUS_STORAGE_REPLICATION_FACTOR=1
      - CONNECT_PLUGIN_PATH=/usr/share/java,/usr/share/confluent-hub-components
    volumes:
      - ./confluent-sink-connector:/usr/share/confluent-hub-components

{
  "name": "mysql-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "source-db-host",
    "database.port": "3306",
    "database.user": "source-db-user",
    "database.password": "source-db-password",
    "database.server.id": "1",
    "database.server.name": "source-db",
    "database.whitelist": "your-database-name",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "dbhistory.source-db"
  }
}

docker-compose up -d

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
    http://localhost:8083/connectors/ -d @mysql-connector.json

{
  "name": "jdbc-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "your-database-name.public.table-name",
    "connection.url": "jdbc:your-target-database-connection-url",
    "connection.user": "your-target-db-user",
    "connection.password": "your-target-db-password",
    "auto.create": "true",
    "auto.evolve": "true",
    "insert.mode": "upsert",
    "pk.fields": "id",
    "pk.mode": "record_key",
    "transforms": "unwrap,insertTopic",
    "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
    "transforms.insertTopic.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.insertTopic.topic.field": "table"
  }
}

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
    http://localhost:8084/connectors/ -d @jdbc-sink-connector.json