如何限制Kafka Connect JDBC源连接器的吞吐量以避免峰值

针对PostgreSQL大表导入Kafka时的速率控制需求，以下是具体的配置调整方案，包括你关注的参数和更精准的控速手段：

一、调整核心轮询参数 #

你提到的两个参数是控速的关键，具体影响和调整方式如下：

• poll.interval.ms：定义连接器两次轮询数据库的时间间隔（单位：毫秒）。当前值为10000（10秒），增大该值会减少单位时间内的轮询次数，直接降低整体导入速率。例如改为30000（30秒），让连接器每半分钟才拉取一次数据。
• batch.max.rows：定义每次轮询从数据库拉取的最大行数。当前值为100，减小该值会降低单次轮询的数据量。例如改为50，每次仅拉取50条记录。

二、补充底层数据拉取控制 #

除上述参数外，可通过fetch.size限制JDBC驱动的单次数据获取量（底层JDBC的fetch size），避免数据库一次性返回过多数据：

"fetch.size": "20"

这个参数控制JDBC驱动从数据库游标中每次获取的行数，配合batch.max.rows使用，能更精细地控制单次轮询的数据规模。

三、基于KIP-731的精准速率限制 #

如果你的Kafka Connect版本在2.8及以上（支持KIP-731），可以直接使用每秒记录数限制参数，这是最精准的控速方式，无需依赖轮询间隔和批次大小：

"source.max.records.per.second": "50"

设置该参数后，连接器会严格控制每秒发送到Kafka的记录数（示例为50条/秒），从根源避免数据摄入峰值。

四、调整后的完整配置示例 #

整合上述参数后的配置如下：

{
    "name": "my-connector",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "tasks.max": "1",
        "topic.prefix": "my-topic",
        "connection.url": "jdbc:postgresql://localhost:5432/my-db",
        "connection.user": "user",
        "connection.password": "password",
        "mode": "timestamp",
        "timestamp.column.name": "time",
        "poll.interval.ms": "30000",
        "batch.max.rows": "50",
        "fetch.size": "20",
        "source.max.records.per.second": "50",
        "query": "SELECT * FROM my-table",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter.schemas.enable": "false"
    }
}

五、配置生效方式 #

修改配置后，通过Kafka Connect的REST API提交更新即可生效：

curl -X PUT -H "Content-Type: application/json" --data @updated-config.json http://your-connect-host:8083/connectors/my-connector/config

内容的提问来源于stack exchange，提问作者Thomas