Kafka record headers in Stateful Functions: closing a five-year-old gap

Apache Stateful Functions never let a function see Kafka record headers, and never let it set them. Five years, no headers. StateFun Actors 3.4.0-KZM-3.4 closes that gap.

Read them on the way in, set them on the way out:

// in: the headers of the record that invoked you
String traceId = message.headers().stream()
    .filter(h -> h.key().equals("traceparent"))
    .findFirst()
    .map(MessageHeader::valueAsUtf8String)
    .orElse("no-trace");

// out: headers on the record you produce
context.send(
    KafkaEgressMessage.forEgress(EGRESS)
        .withTopic("orders.done")
        .withUtf8Key(orderId)
        .withValue(receipt)
        .withUtf8Header("traceparent", traceId)
        .withHeader("attempt", 2)     // a real binary int, not "2"
        .build());

Turn it on per topic in module.yaml:

kind: io.statefun.kafka.v1/ingress
spec:
  id: shop/orders-in
  address: kafka:9092
  forwardHeaders: true          # every topic below inherits this
  topics:
    - topic: orders
      valueType: shop/Order
      targets: [shop/order-fn]
    - topic: clickstream
      forwardHeaders: false     # noisy topic opts back out
      valueType: shop/Click
      targets: [shop/click-fn]

The details that matter:

• Null stays null. Kafka distinguishes a null header value from an empty one, and so do we, through the whole ingress, function, egress round trip.
• Reads never throw in production. Absent or undecodable values come back as null, not as an exception inside a live invocation.
• Off by default. Topics that don't opt in never capture header bytes, so header-free pipelines pay nothing.
• Additive protocol. Old SDKs and new runtimes interoperate. No lockstep upgrade, no savepoint impact.
• Tracing works now. A W3C traceparent header can finally ride through a StateFun function graph without living inside your payload schema.

Full guide: Kafka record headers · Design write-up: how it landed with zero flink-core changes