如何为Kafka Connector配置独立Broker存储偏移量至Kafka Topic

Kafka Connect完全支持这种分离部署的需求——将偏移量、配置及状态数据存储在你的Broker集群，同时将业务数据写入客户环境的Broker Topic。核心是通过拆分Connect Worker和连接器的配置，分别指向不同的Kafka集群：

1. 配置Connect Worker指向你的Broker集群（存储偏移量/配置/状态） #

在Connect Worker的配置文件（通常是connect-distributed.properties）中，修改以下参数，指定你的Broker信息：

• offset.storage.bootstrap.servers: 你的Broker集群地址（示例：our-broker-1:9092,our-broker-2:9092）
• offset.storage.topic: 你的集群中专门用于存储偏移量的Topic名称（示例：connect-offsets-our-cluster，建议提前创建并设置至少3个副本保证高可用）
• config.storage.bootstrap.servers: 同样设为你的Broker地址（用于存储连接器配置信息）
• config.storage.topic: 你的集群中存储配置的Topic（示例：connect-configs-our-cluster）
• status.storage.bootstrap.servers: 你的Broker地址（用于存储连接器运行状态）
• status.storage.topic: 你的集群中存储状态的Topic（示例：connect-statuses-our-cluster）

2. 配置Kinesis Source Connector指向客户的Broker集群（存储业务数据） #

在Kinesis连接器的配置JSON中，单独指定客户的Broker信息，确保业务数据写入客户集群：

{
  "name": "kinesis-to-customer-kafka",
  "config": {
    "connector.class": "io.confluent.connect.kinesis.KinesisSourceConnector",
    "tasks.max": "1",
    "kinesis.stream": "customer-kinesis-stream",
    "kinesis.region": "us-east-1",
    // 指向客户Broker的配置
    "bootstrap.servers": "customer-broker-1:9092,customer-broker-2:9092",
    "topic": "customer-target-topic",
    // 其他Kinesis连接器必要配置...
  }
}

关键注意事项 #

• 权限控制：确保Connect Worker进程有访问你的Broker集群的读写权限，同时连接器有访问客户Broker集群的写入权限（如果客户Broker启用了SASL/SSL等安全机制，需在连接器配置中添加对应的安全参数）
• 偏移量Topic可用性：你的偏移量Topic要配置足够的副本数和持久化策略，避免自身集群故障导致偏移量丢失
• 版本兼容性：确保你的Connect Worker版本与客户Broker版本兼容（尽量使用同大版本，比如都是2.8.x或3.3.x）

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