如何将Kafka旧数据迁移至MongoDB？解决旧主题轮询超时问题

针对更换主题后旧数据未同步、拉取旧主题时出现poll timeout的问题，以下是几种可落地的迁移方案：

方案一：调整消费者配置重新拉取 #

适合已有消费者代码的场景，核心解决poll超时问题并重置偏移量从头消费：

• 调整poll.timeout.ms参数：将消费者轮询超时时间从默认值（通常1000ms）调大，比如设为30000（30秒），避免因数据量过大或网络延迟导致超时
• 重置消费者偏移量：如果之前的消费者组已提交过旧主题偏移量，强制重置到最早位置，命令如下：

  kafka-consumer-groups.sh --bootstrap-server <kafka-broker>:9092 --group <你的消费者组ID> --reset-offsets --to-earliest --topic <旧主题名> --execute
  

• 修改消费者代码配置：确保auto.offset.reset设为earliest，临时关闭自动提交（enable.auto.commit=false），拉取成功后手动提交偏移量，防止中途失败重复消费
• 批量写入MongoDB：将拉取的消息攒成批量（比如1000条）后调用insertMany写入，减少数据库连接开销，提升效率

方案二：用Kafka Connect无代码迁移 #

适合无需自定义数据转换的场景，利用官方连接器快速批量迁移：

• 配置MongoDB Sink Connector：核心配置示例（写入connect-distributed.properties或单独的连接器配置文件）：

  name=mongo-sink-old-topic
  connector.class=com.mongodb.kafka.connect.MongoSinkConnector
  topics=<旧主题名>
  connection.uri=mongodb://<MongoDB地址>:27017
  database=<目标数据库名>
  collection=<目标集合名>
  consumer.auto.offset.reset=earliest
  batch.size=1000
  tasks.max=3  # 根据数据量设置并行任务数
  

• 启动连接器后，它会自动从头拉取旧主题数据并批量写入MongoDB，无需编写代码
• 迁移过程中可通过Kafka Connect的REST API（GET /connectors/mongo-sink-old-topic/status）监控进度

方案三：编写自定义迁移脚本 #

适合需要对数据进行复杂转换的场景，用代码灵活控制迁移流程：

• 以Python为例，核心逻辑如下（处理poll超时、批量写入）：

  from kafka import KafkaConsumer
  from pymongo import MongoClient
  import json
  
  # 初始化Kafka消费者
  consumer = KafkaConsumer(
      "<旧主题名>",
      bootstrap_servers=["<Kafka地址>:9092"],
      auto_offset_reset="earliest",
      enable_auto_commit=False,
      group_id="migration-temp-group",
      consumer_timeout_ms=60000  # 60秒超时，无消息则退出
  )
  
  # 初始化MongoDB连接
  mongo_client = MongoClient("mongodb://<MongoDB地址>:27017")
  target_coll = mongo_client["<目标数据库>"]["<目标集合>"]
  
  batch = []
  batch_size = 1000
  
  for msg in consumer:
      # 解析消息（根据你的消息格式调整）
      data = json.loads(msg.value.decode("utf-8"))
      batch.append(data)
      
      # 批量写入
      if len(batch) >= batch_size:
          target_coll.insert_many(batch)
          consumer.commit_sync()  # 手动提交偏移量
          batch = []
  
  # 处理剩余的最后一批消息
  if batch:
      target_coll.insert_many(batch)
      consumer.commit_sync()
  
  # 关闭连接
  consumer.close()
  mongo_client.close()
  

• 脚本中设置consumer_timeout_ms，超过60秒无新消息则自动退出，避免无限等待

方案四：镜像旧主题到新主题再同步 #

如果现有系统已有新主题到MongoDB的同步流程，可先把旧主题数据镜像到新主题：

• 使用Kafka MirrorMaker 2.0（推荐）进行主题镜像，示例命令：

  kafka-mirror-maker.sh --clusters source=<旧集群地址>,target=<新集群地址> --whitelist <旧主题名> --consumer.config source-consumer.properties --producer.config target-producer.properties
  

  其中source-consumer.properties需设置auto.offset.reset=earliest，确保从头拉取旧主题数据
• 镜像完成后，现有同步流程会自动将新主题中的旧数据同步到MongoDB，无需修改现有代码

通用注意事项 #

• 迁移前备份MongoDB目标集合，避免数据冲突或错误
• 用kafka-consumer-groups.sh --bootstrap-server <地址> --group <组ID> --describe监控消费者lag，确认所有数据都已消费完成
• 如果旧主题数据已超过Kafka的retention.ms被清理，需先从Kafka的快照或日志备份中恢复数据再迁移

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