Kafka Python跨Topic数据迁移无输出问题排查及实现咨询

• 迁移逻辑执行顺序死锁：迁移服务的主函数先调用read_topic_data()，而KafkaConsumer的迭代器是永久阻塞的轮询逻辑，只要Kafka集群正常，这个循环永远不会退出，send_data_to_topic()永远不会被执行，自然没有数据写入output-topic
• 消费位点配置缺失：你没有给Consumer配置auto_offset_reset参数，默认值是latest，如果你是先启动Producer生产完数据，再启动迁移服务的话，Consumer默认只会消费启动后新产生的消息，不会读取之前已经存在的历史数据
• 生产分区配置不匹配：你描述input-topic有5个分区，但发数据的Producer里硬编码num_partitions=3，只会往0/1/2三个分区发数据，虽然不影响功能，但和你预期的分区分布不一致
• 内存攒批风险：当前逻辑是把所有读到的数据全放到内存list里排序再批量发，数据量稍大就会触发OOM，可用性极低

1. 基础功能修复（直接解决无数据问题） #

如果不需要全量数据排序，直接修改迁移服务代码为边读边写即可，参考代码如下：

from kafka import KafkaConsumer, KafkaProducer

topic_name_input  = "input-topic"
topic_name_output = "output-topic"

def bytes_to_int(bytes_obj):
    result = 0
    for b in bytes_obj:
        result = result * 256 + int(b)
    return result

consumer = KafkaConsumer(
     topic_name_input,
     bootstrap_servers=['localhost:9092'],
     group_id='my-group',
     auto_offset_reset='earliest', # 新增：从头消费历史数据
     value_deserializer=lambda x: bytes_to_int(x)
    )

# output-topic只有1个分区，不需要指定分区，默认会全发往唯一分区
producer1 = KafkaProducer(bootstrap_servers=['localhost:9092'], value_serializer=lambda x: bytes(x))

def migrate_data():
    print("start migrate...")
    for message in consumer:
        print(f"migrate value: {message.value}")
        # 边读边写，不需要攒全量数据到内存
        producer1.send(topic_name_output, value=message.value)
        # 可选：定期flush，也可以靠producer内部自动批量发送提升性能
        producer1.flush()

if __name__ == "__main__":
    migrate_data()

如果业务要求必须全量排序后再写入，需要先获取input-topic所有分区的最大偏移量，消费到所有分区都达到最大偏移量后，停止消费再做排序写入，避免无限循环。

2. 高效实现优化 #

如果是一次性历史数据迁移：

• 可以先暂停input-topic的写入，避免增量数据干扰排序结果，读完所有历史数据排序后一次性写入output-topic即可
• 可以开启多线程消费input-topic的多个分区，提升消费速度，最后汇总排序再写入
  如果是实时同步迁移：
• 不需要全量排序的场景直接用上述边读边写的逻辑即可，吞吐量可以达到Kafka单分区写入上限
• 可以开启Producer的批量发送、压缩配置进一步提升效率：

producer1 = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda x: bytes(x),
    batch_size=16384, # 16KB批量阈值
    linger_ms=5, # 最多等待5ms凑批量
    compression_type='lz4' # 开启lz4压缩降低传输带宽占用
)

如果你的场景只是简单的跨Topic同步，没有复杂的流处理需求（比如聚合、窗口计算、状态存储），完全不需要引入Kafka Streams，Python客户端的消费生产逻辑已经足够轻量高效。如果后续需要加入更复杂的流处理逻辑，优先考虑Python生态的Faust流处理框架即可，不需要额外引入JVM技术栈的Kafka Streams。

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