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

【问题标题】：Is there a way to update configuration of a running kafka connector有没有办法更新正在运行的 kafka 连接器的配置
【发布时间】：2022-01-29 05:07:24
【问题描述】：

我有一个正在运行的 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"}%

【问题讨论】：

标签： apache-kafka apache-kafka-connect

【解决方案1】：

要更新现有连接器的配置，您可以使用Connect REST Interface 的PUT /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"
}'

【讨论】：

{"error_code":405,"message":"HTTP 405 Method Not Allowed"}
是的，您必须将连接器名称添加到 url。检查我添加的示例
是的，连接器任务将在配置升级后重新启动
这取决于发生了什么问题：如果它是可重试的，那么你就按照你说的那样重新启动。请记住，重新启动之前 3.0.0 版本的连接器不会影响任务（检查 KIP-745 cwiki.apache.org/confluence/pages/…）
curl -s --location --request POST "http://10.24.18.167:80/connectors/xyz.connector/restart?includeTasks=true&onlyFailed=true" 我认为这应该可行。