Salesforce Bulk API Sink Connector生成过多批次的配置优化问询

你配置的参数为什么没用？ #

你加的confluent.topic.consumer.fetch.max.wait.ms和confluent.topic.consumer.fetch.min.bytes是管控Confluent内部配置主题（比如connect-configs）的消费者参数，和业务数据的消费批次逻辑完全无关，修改这些自然不会起作用。

正确的配置方法 #

要控制Salesforce批次生成，需要从Kafka消费者拉取策略和连接器自身批次提交规则两个层面配置：

1. 业务数据的消费者配置（使用consumer.前缀） #

将你想要设置的参数改为consumer.开头，添加到Sink连接器的配置中：

consumer.fetch.max.wait.ms=10000
consumer.fetch.min.bytes=1000

这两个参数的作用是：让Kafka消费者最多等待10秒，要么凑够1000字节的数据再拉取，要么到时间就触发拉取，避免每次只拉取单条记录。

2. Salesforce Bulk连接器专属批次控制参数（核心） #

这是直接控制Salesforce批次生成的关键配置，优先设置：

batch.size=1000
flush.timeout.ms=10000

• batch.size: 每个Salesforce批次最多包含1000条记录（可根据需求调整，最大不超过Salesforce Bulk API限制的10000条/批次）
• flush.timeout.ms: 即使未凑够batch.size的记录数，只要等待时间超过10秒，就会立即将当前积累的记录打包成一个Salesforce批次提交

3. 配置验证 #

修改完成后重启Sink连接器，再次测试间隔0.7秒的3次更新操作，应该会被合并为1个Salesforce批次。你可以通过Confluent Control Center查看连接器的运行指标，或者直接在Salesforce后台查看批次记录来确认效果。

额外提醒 #

• 若你的Salesforce Bulk Sink Connector版本过旧，可能不支持flush.timeout.ms参数，需要升级到新版连接器
• batch.size不要设置过大，避免Salesforce处理批次时出现超时错误
• 小流量场景下，主要依赖flush.timeout.ms来合并批次；大流量场景下，调大batch.size能有效减少批次生成数量

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