如何处理kafka-python生产者过期批次，将记录转存本地而非直接删除？

1 核心实现思路 #

kafka-python原生没有自动落盘超时批次的能力，需要通过回调捕获超时异常，配合本地存储组件实现，核心逻辑是在发送回调中识别KafkaTimeoutError，将对应消息元数据和内容写入本地存储，不会修改原有超时阈值配置，也不影响生产者正常发送逻辑。

2 具体配置与改造步骤 #

2.1 基础生产者配置调整 #

首先调整生产者核心参数，可配置有限次重试优先规避瞬时网络波动，减少不必要的本地落盘：

from kafka import KafkaProducer
from kafka.errors import KafkaTimeoutError
import json
import threading
import queue
from datetime import datetime

# 生产者基础配置
producer_config = {
    'bootstrap_servers': '你的Broker地址:9092',
    'request_timeout_ms': 20000, # 保持原有20秒超时配置不变
    'retries': 2, # 可自定义重试次数，重试全部失败后再落地
    'retry_backoff_ms': 1000,
    'acks': 'all', # 可根据业务可靠性要求调整
    'value_serializer': lambda v: json.dumps(v).encode('utf-8')
}

2.2 实现异步本地持久化队列 #

发送回调运行在生产者内部的sender线程，不要在回调中直接做磁盘IO避免阻塞核心逻辑，单独启动线程处理本地落盘：

# 本地落盘队列，最大长度可根据服务器内存容量调整
local_persist_queue = queue.Queue(maxsize=10000)

# 本地落盘线程逻辑，此处示例写入JSONL文件，也可替换为RocksDB等嵌入式KV存储
def persist_worker():
    # 按天生成落盘文件，避免单个文件过大
    current_date = datetime.now().strftime("%Y%m%d")
    file_path = f"./kafka_expired_records_{current_date}.jsonl"
    while True:
        try:
            record = local_persist_queue.get(timeout=1)
            with open(file_path, 'a', encoding='utf-8') as f:
                f.write(json.dumps(record, ensure_ascii=False) + '\n')
            local_persist_queue.task_done()
            # 跨天自动切换落盘文件
            new_date = datetime.now().strftime("%Y%m%d")
            if new_date != current_date:
                current_date = new_date
                file_path = f"./kafka_expired_records_{current_date}.jsonl"
        except queue.Empty:
            continue

# 启动后台落盘线程
threading.Thread(target=persist_worker, daemon=True).start()

2.3 自定义发送回调捕获超时异常 #

发送消息时将消息内容、目标topic、分区等元数据作为上下文传入回调，识别到超时异常时写入落盘队列：

# 发送成功回调，可按需添加日志逻辑
def on_send_success(record_metadata):
    pass

# 发送失败回调
def on_send_error(excp, record, topic, partition=None):
    if isinstance(excp, KafkaTimeoutError):
        # 捕获到批次超时异常，组装元数据写入落盘队列
        persist_record = {
            'topic': topic,
            'partition': partition,
            'value': record,
            'error_time': datetime.now().isoformat()
        }
        try:
            local_persist_queue.put_nowait(persist_record)
        except queue.Full:
            # 队列满时降级同步写磁盘，避免数据丢失
            with open("./kafka_emergency_records.jsonl", 'a', encoding='utf-8') as f:
                f.write(json.dumps(persist_record, ensure_ascii=False) + '\n')
    # 其他类型异常可按需扩展处理逻辑

# 初始化生产者
producer = KafkaProducer(**producer_config)

# 封装发送方法
def send_record(topic, record, partition=None):
    future = producer.send(topic, value=record, partition=partition)
    future.add_callback(on_send_success)
    future.add_errback(on_send_error, record=record, topic=topic, partition=partition)

3 后续重发逻辑说明 #

远程Broker恢复可用后，单独编写重发脚本读取本地落盘文件，遍历每条记录调用生产者发送即可，因无需保证发送顺序，无需额外处理顺序逻辑，发送成功后可删除对应落盘文件或标记已发送状态。

4 注意事项 #

• 批量发送大流量场景下，可扩展落盘逻辑为批量写文件，进一步提升IO效率
• 本地存储目录要预留足够磁盘空间，可配置定期清理已成功重发的历史落盘文件
• 若需要更高的落盘可靠性，可替换JSONL文件为RocksDB等支持事务的嵌入式存储，避免进程崩溃时丢失未落盘的数据

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