如何在Python Flink Stateful Function的Kafka Ingress中访问Kafka Key

Flink Stateful Functions从3.2版本开始，Python SDK已经原生支持Kafka Ingress的元数据透传能力，可直接获取Kafka Key，具体操作步骤如下：

步骤1：调整Kafka Ingress配置 #

在服务部署的模块YAML配置中，给对应的Kafka Ingress添加metadata配置项，显式声明需要透传的Kafka元数据字段，包含kafka.key：

ingress:
  - kind: kafka
    id: "your-business-kafka-ingress"
    address: "kafka-broker-address:9092"
    consumerGroupId: "your-business-consumer-group"
    topics:
      - "your-business-topic"
    # 声明需要透传的Kafka元数据
    metadata:
      - "kafka.key"
      - "kafka.timestamp" # 可选，按业务需要添加其他元数据
    deserializer:
      valueFormat: "json" # 按你实际的value序列化格式调整

步骤2：Python函数中读取Kafka Key #

在状态函数的处理逻辑中，直接从调用上下文Context的message_headers属性中读取kafka.key对应的值即可，拿到的原始值为bytes类型，按需解码为业务需要的格式即可：

from statefun import StatefulFunctions, Context

functions = StatefulFunctions()

@functions.bind("your-namespace/your-business-function")
def process_kafka_msg(context: Context, message):
    # 读取Kafka Key
    kafka_key_bytes = context.message_headers.get("kafka.key")
    if kafka_key_bytes:
        # 按实际的key编码格式调整解码逻辑
        kafka_key = kafka_key_bytes.decode("utf-8")
        # 后续处理业务逻辑

补充说明 #

• 如果你使用的Flink Stateful Functions版本低于3.2，直接将Python SDK和集群版本升级到最新稳定版即可使用该能力，低版本没有原生的Kafka元数据透传支持。
• 如果你的Kafka Key采用了特殊的编码格式，或者需要和Value做联合反序列化，可以自行实现Java端的KafkaIngressDeserializer接口，在反序列化阶段将Kafka Key写入消息头或者和Value打包为复合结构，Python端可直接读取处理。

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