out_kafka: support record-derived Kafka message headers

[kimonus]

Is your feature request related to a problem? Please describe.

The Kafka protocol and Fluent Bit's bundled librdkafka client support message headers, but the out_kafka plugin doesn't provide a general way to create those headers from a record field.

Some pipelines receive a variable set of per-record transport attributes, such as correlation identifiers, routing metadata, or content metadata. These values need to be Kafka message headers rather than fields in the serialized message body. Keeping them in the payload isn't equivalent because Kafka consumers, interceptors, and routing infrastructure access message headers separately from the value.

librdkafka already exposes the required producer APIs through rd_kafka_headers_new(), rd_kafka_header_add(), and RD_KAFKA_V_HEADERS. The missing part is the Fluent Bit out_kafka configuration and record-mapping layer.

Describe the solution you'd like

Add a headers_key option to out_kafka. Its value identifies a top-level record field whose MessagePack map is converted to Kafka message headers for that record.

Proposed behavior:

• An empty or ASCII-whitespace-only headers_key disables the feature.
• Nonblank field names are matched exactly and aren't trimmed.
• The selected field must be a MessagePack map.
• Header names must be strings.
• Header values can be strings, binary values, or null values.
• Empty values, null values, insertion order, and duplicate names are preserved on the Kafka record.
• Unsupported map entries are skipped with a warning.
• A valid selected map is removed from the serialized payload by default.
• A preserve_headers_key boolean option, defaulting to false, retains the original map in the payload when enabled.
• A missing selected field leaves payload processing unchanged.
• A selected field with a non-map value remains in the payload to avoid data loss.
• An empty map is valid and follows the configured preservation behavior.
• If duplicate top-level map fields match, the first map supplies the Kafka headers and later fields remain in the payload.
• The feature applies to the avro, gelf, json, msgpack, and raw formats. A nonblank headers_key is rejected for otlp_json and otlp_proto, whose serialization path doesn't expose an individual source map.
• Kafka 0.11-compatible brokers or later are required for message-header support.

Sanitized YAML example:

pipeline:
  inputs:
    - name: dummy
      tag: example.logs
      dummy: '{"message":"example","kafka_headers":{"trace-id":"abc-123","content-type":"application/json"}}'

  outputs:
    - name: kafka
      match: example.logs
      brokers: 192.0.2.10:9092
      topics: example-logs
      format: json
      headers_key: kafka_headers
      preserve_headers_key: false

This produces trace-id and content-type as Kafka message headers. The kafka_headers field isn't included in the JSON value unless preserve_headers_key is set to true.

Describe alternatives you've considered

• Keep the attributes in the serialized payload. This doesn't make them available through Kafka's message-header interface.
• Configure individual headers explicitly. PR out_kafka: Add dynamic/static headers support #8583 proposes repeated header options with static values or individual record-field references. That interface is useful when the header names are known in configuration, but it doesn't cover records that already contain a variable header map. The two interfaces are complementary.
• Transform the record with a filter or Lua script. Filters can reshape the payload, but they can't attach Kafka producer headers through the existing output interface.
• Patch librdkafka. This isn't necessary because librdkafka already implements the required protocol and ownership semantics.

Additional context

This request is a focused out_kafka implementation of the record-map use case discussed in #1673, #8334, and #9448. Those broader requests are closed as stale; #8334 and #9448 also cover both Kafka input and output, while this request deliberately defines and tests the output-side contract. It is intended to coexist with the explicit per-header configuration proposed in #8583:

• header: configure one known header at a time, optionally sourcing its value from a record field.
• headers_key: convert all supported entries from one record map at runtime.

The implementation can remain confined to Fluent Bit-owned plugins/out_kafka code and tests. It doesn't require changes to bundled librdkafka sources or to the Kafka wire protocol.

A tested implementation and separate documentation change are prepared locally. Coverage includes payload removal and preservation, empty/missing/non-map sources, null and binary values, duplicate names, MessagePack, raw, GELF, Avro with Schema Registry resolution, queue-full retry ownership, configured message-size rejection, startup validation, and strict Valgrind execution.

Related work: #1673, #8334, and #9448 (closed requests); #8583 and #9057 (open, conflicting implementations of individually configured headers).