有没有办法更新正在运行的 kafka 连接器的配置

Question

我有一个正在运行的 Kafka debezium 连接器，现在我想更新以下参数的值，

heartbeat.interval.ms
snapshot.mode

当前配置：

{
  "connector.class": "io.debezium.connector.mysql.MySqlConnector",
  "snapshot.locking.mode": "minimal",
  "database.user": "cdc_user",
  "tasks.max": "3",
  "database.history.kafka.bootstrap.servers": "XXX:9092",
  "database.history.kafka.topic": "history.cdc.fkw.supply.marketplace.fk_sp_generic_checklist",
  "database.server.name": "cdc.fkw.supply.marketplace.fk_sp_generic_checklist",
  "heartbeat.interval.ms": "5000",
  "database.port": "3306",
  "table.whitelist": "fk_sp_generic_checklist.entity_checklist",
  "database.hostname": "xyzcloud.in",
  "database.password": "ACCSSDD",
  "database.history.kafka.recovery.poll.interval.ms": "5000",
  "name": "cdc.fkw.supply.marketplace1.fk_sp_generic_checklist.connector",
  "database.history.skip.unparseable.ddl": "true",
  "errors.tolerance": "all",
  "database.whitelist": "fk_sp_generic_checklist",
  "snapshot.mode": "when_needed"
}

This is what is get:

curl --location --request PUT 'http://XX.XX.7/connectors/' \
--header 'Content-Type: application/json' \
--data-raw '{
    "name": "cdc.fkw.supply.marketplace.fk_sp_generic_checklist.connector",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "database.hostname": "abc.cloud.in",
        "database.port": "3306",
        "database.user": "XXXXX",
        "database.password": "XXXXX",
        "database.server.name": "cdc.fkw.supply.marketplace.fk_sp_generic_checklist",
        "database.whitelist": "pno",
        "table.whitelist": "fk_sp_generic_checklist.entity_checklist",
        "database.history.kafka.bootstrap.servers": "X9:9092",
        "database.history.kafka.topic": "history.cdc.fkw.supply.marketplace.fk_sp_generic_checklist",
        "include.schema.changes": "false",
        "snapshot.mode": "when_needed",
        "errors.tolerance": "all"
    }
}'

Output: {"error_code":405,"message":"HTTP 405 Method Not Allowed"}

我尝试了上述方法，但它给了我不允许的方法。

版本：

{"version":"6.1.1-ccs","commit":"c209f70c6c2e52ae","kafka_cluster_id":"snBlf-kfTdCYWEO9IIEXTA"}%

Answer 1

要更新现有连接器的配置，您可以使用Connect REST InterfacePUT /connectors/(string:name)/config的方法。

curl -XPUT <host>:<port>/connectors/cdc.fkw.supply.marketplace1.fk_sp_generic_checklist.connector/config \
  --header 'Accept: application/json' \
  --header 'Content-Type: application/json' -d '{
  "connector.class": "io.debezium.connector.mysql.MySqlConnector",
  "snapshot.locking.mode": "minimal",
  "database.user": "cdc_user",
  "tasks.max": "3",
  "database.history.kafka.bootstrap.servers": "XXX:9092",
  "database.history.kafka.topic": "history.cdc.fkw.supply.marketplace.fk_sp_generic_checklist",
  "database.server.name": "cdc.fkw.supply.marketplace.fk_sp_generic_checklist",
  "heartbeat.interval.ms": "5000",
  "database.port": "3306",
  "table.whitelist": "fk_sp_generic_checklist.entity_checklist",
  "database.hostname": "xyzcloud.in",
  "database.password": "ACCSSDD",
  "database.history.kafka.recovery.poll.interval.ms": "5000",
  "name": "cdc.fkw.supply.marketplace1.fk_sp_generic_checklist.connector",
  "database.history.skip.unparseable.ddl": "true",
  "errors.tolerance": "all",
  "database.whitelist": "fk_sp_generic_checklist",
  "snapshot.mode": "when_needed"
}'