Kafka Consumer因Null值序列化失败，如何修改value_serializer？

当PostgreSQL中的记录被删除时，同步到Kafka的消息value会是Null值，直接使用原有的value_serializer会因为x是NoneType而无法调用decode()方法，引发错误。你可以通过添加空值判断来改写序列化逻辑：

方式一：直接在lambda中添加条件判断 #

consumer = KafkaConsumer(
        topic,
        bootstrap_servers='localhost:9092',
        auto_offset_reset='earliest',
        value_serializer=lambda x: loads(x.decode('utf-8')) if x is not None else None
    )

方式二：使用独立函数（更易维护复杂逻辑） #

如果后续需要扩展空值处理逻辑，推荐定义单独的序列化函数：

def handle_value(x):
    if x is None:
        return None  # 可根据业务需求改为返回空字典{}或其他默认值
    return loads(x.decode('utf-8'))

consumer = KafkaConsumer(
        topic,
        bootstrap_servers='localhost:9092',
        auto_offset_reset='earliest',
        value_serializer=handle_value
    )

两种方式都会先检查输入是否为None，避免调用None的decode()方法，同时根据你的需求返回对应的空值处理结果。

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