Docker环境下Kafka Consumer无法将消费数据写入列表的问题

核心问题根源 #

KafkaConsumer是阻塞式的无限迭代器——默认情况下，它会一直等待新消息，不会自动停止迭代。当Producer发送完所有消息后，Consumer仍会持续阻塞，导致[msg.value for msg in consumer]这个列表推导式永远无法执行完成，自然无法生成完整的消息列表。

排查步骤 #

1. 验证阻塞状态：在列表推导式前后添加日志，比如logger.info("Starting to consume messages...")和logger.info(f"Received {len(msg_values)} messages")，观察日志是否停在"Starting to consume messages..."，确认Consumer是否卡在迭代环节。
2. 手动验证消息存在性：进入broker容器，用Kafka自带的命令行消费者确认消息已成功写入Topic：

   docker exec -it broker kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_test --from-beginning
   

3. 检查Consumer配置：确认是否有配置能让Consumer在读取完现有消息后自动停止迭代。

解决方案 #

方案1：添加超时配置（最简单） #

在初始化KafkaConsumer时，添加consumer_timeout_ms参数，指定无新消息时的等待超时时间，超时后Consumer会自动停止迭代：

def consume_data_with_kafka():
    """
    Kafka Consumer read data from topic.
    Args: None 
    Returns:
        data: dataframe: input dataframe
    """
    try:
        logger.info("Kafka Consumer setting up...")
        consumer = KafkaConsumer(
                "topic_test",
                bootstrap_servers=['broker:9092'],
                value_deserializer=lambda x: json.loads(x.decode("utf-8")),
                auto_offset_reset='earliest',
                enable_auto_commit=False,
                consumer_timeout_ms=5000  # 5秒无新消息则停止迭代，可按需调整
            )   
        # 初始化时已指定topic，无需重复subscribe，去掉该行避免警告
        # consumer.subscribe(["topic_test"])
        
        logger.info("Starting to consume messages...")
        msg_values = [msg.value for msg in consumer]
        logger.info(f"Received {len(msg_values)} messages")
        
        result = pd.DataFrame(
            msg_values,
            columns=my_cols,
        )
        consumer.commit()
        return result
    except Exception as e:
        # 修正异常信息（原错误写成了Producer）
        logger.error(f"Error in Kafka Consumer: {str(e)}")
        raise Exception(f"Error in Kafka Consumer: {str(e)}")

方案2：手动控制迭代次数（适合已知消息数量的场景） #

如果提前知道Producer发送的消息总数，可以循环指定次数读取：

def consume_data_with_kafka(expected_count):
    try:
        # ... 初始化Consumer代码 ...
        msg_values = []
        for _ in range(expected_count):
            msg = next(consumer)
            msg_values.append(msg.value)
        # ... 后续DataFrame创建代码 ...
    except StopIteration:
        logger.warning(f"Received fewer messages than expected: {len(msg_values)} vs {expected_count}")

方案3：使用poll()手动拉取消息（更灵活） #

用poll()方法批量拉取消息，超时后自动退出循环：

def consume_data_with_kafka():
    try:
        # ... 初始化Consumer代码 ...
        msg_values = []
        while True:
            # 3秒超时，拉取当前可用的所有消息
            records = consumer.poll(timeout_ms=3000)
            if not records:
                # 无消息则退出循环
                break
            # 遍历所有分区的消息
            for partition, msgs in records.items():
                for msg in msgs:
                    msg_values.append(msg.value)
        # ... 后续DataFrame创建代码 ...
    except Exception as e:
        # ... 异常处理 ...

额外优化点 #

1. 移除重复订阅：初始化KafkaConsumer时已传入"topic_test"，无需再调用consumer.subscribe(["topic_test"])，否则会触发日志中的subscription unchanged警告。
2. 修正异常信息：原代码中异常提示为"Error when setting up Kafka Producer!"，实际应为Consumer相关错误，避免排查混淆。

内容的提问来源于stack exchange，提问作者Ugur Selim Ozen