fluent / fluent/fluent-bit

out_kafka: support record-derived Kafka message headers

Open
#12,286 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
C
Stars
8.1k
Forks
2k
Avg merge
4d 20h
Merged PRs (30d)
71

Description

**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:

```yaml
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 #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).

Contributor guide

Open the contributing guide

Research direction

Start in plugins/out_kafka and its existing tests, tracing the record-mapping and serialization paths for the avro, gelf, json, msgpack, and raw formats, along with the librdkafka header APIs. Done means implementing headers_key and preserve_headers_key with the specified validation, payload-preservation, header-value, retry, size, and startup behaviors while rejecting unsupported OTLP formats and covering the listed cases.

Written by the indexing model from the issue text.

Assessment

Tech stack
c, kafka
Domain
stream-processing
Issue type
Feature
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Clearly specified
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.