Librdkafka: Add support for Record Headers

[exaurdon]

Description

Kafka 0.11.0.0 added the concept of record headers (separate from the payload):
https://issues.apache.org/jira/browse/KAFKA-4208

The Java API was updated to support this. This request is to add support for reading/sending this headers in librdkafka.

To vote for this issue add a 👍 reaction to this comment

[edenhill]

Header support is on the backlog but not yet prioritized.

It would be useful to know how you expect to use it, please fill out the following template (copy it to a comment):

• [ ] Add header
• [ ] Delete header (by name)
• [ ] Alter/Update header value (by name)
• [ ] Get/read header (by name)
• [ ] Enumerate all headers
• [ ] Enumerate headers for single name
• Typical number of headers per message:
• Typical size of header value:

[exaurdon]

We are using Kafka as part of a microservice architecture to publish events. Currently I am sending header information within the payload, with the main values being:

1. Correlation ID string
2. Event timestamp

• [X] Add header
• [ ] Delete header (by name)
• [ ] Alter/Update header value (by name)
• [X] Get/read header (by name)
• [X] Enumerate headers

Typical number of headers per message: 2-4

Thanks

[mjwarno]

We currently are sending an RTP header within the payload. Contains timestamp,sequence number, source of the payload.

Would like the following:
Add header
Delete header (by name)
Alter/Update header value (by name)
Get/read header (by name)
Enumerate headers
•Typical number of headers per message: 2
•Typical size of header value: 12 or 16 bytes, for another payload roughly 512

[gnuphie]

• [x] Add header
• [x] Delete header (by name)
• [x] Alter/Update header value (by name)
• [x] Get/read header (by name)
• [x] Enumerate headers

Typical number of headers per message: 2-4
Typical size of header value: 10-30 bytes

[javierholguera]

• [x] Add header
• [x] Delete header (by name)
• [x] Alter/Update header value (by name)
• [x] Get/read header (by name)
• [x] Enumerate headers
  Typical number of headers per message: 4-10
  Typical size of header value: not sure, not big

We're planning to use this for distributed tracing, among other scenarios.

[malejpavouk]

• [x] Add header

Typical number of headers per message: <5
Typical size of header value: ids, possibly uuids (nothing big)

[edenhill]

Headers support is now on master, please try it out.

[mjwarno]

Thank you very much. I will download and give her a shot.

Mike Warnock
[email protected]

[kparadkar]

When is the next official release expected to try headers. We can only upgrade to official release and not work with master.

[edenhill]

@kparadkar We're aiming for a release mid february the earliest.
The more people that gives new features a spin before that the better, making sure we cover all use-cases.

[exaurdon]

Thanks for implementing this. I have started work to use this new feature, and noted some difficulty in doing so as I have been using the C++ bindings.

Currently, using the C++ bindings, I cannot "safely" access the rd_kafka_message_t pointer needed to call the newly implemented C functions (the Message class does not offer a method for accessing rd_kafka_message_t). As a workaround, it appears I could reasonably safely cast Message to MessageImpl, and then access the public rd_kafka_message_t pointer from there.

Any thoughts on adding C++ support for headers? Or any other tips on how to safely utilize the existing api from C++?

[edenhill]

The cast workaround is fragile so I would not recommend it.

The plan is to add proper headers support to the C++ API, but before that happens we could expose a c_ptr() method on the Message object to return the rd_kafka_message_t pointer. I'm adding the same for rd_kafka_t and rd_kafka_topic_t just for this reason, where the C++ API is lagging behind the C API.

[exaurdon]

Yeah, I definitely saw the cast as fragile and unsafe. The c_ptr() method seems like a good way to permit access to the latest C features, before they have been fully encapsulated.

Thanks

[sobolevan]

Hi guys! Could someone share example of producing message with headers using C++ binding of the library? Thx in advance.

[edenhill]

Headers are not yet supported in the C++ API, only in C.
The C++ API typically lags behind the C API and features are added on community demand, or the community contributions.

[sobolevan]

@edenhill Thx for the answer. So, there is no way to pass headers to produced msg rather than extend C++ interface. Right? (even with workaround like for reading headers of consumed message poiting to C ptr).

[edenhill]

You can access the underlying C rd_kafka_t handle with c_ptr() and then call rd_kafka_producev() with a headers list or individual headers.

You can still use your existing DeliveryReport C++ classes to handle the responses and everything else.

[sobolevan]

I see. Thx for help!

[mjwarno]

Thank you again for providing this support. The C lib worked perfectly. After sending each message with a kafka header, I had to call rd_kafka_headers_destory following by rd_kafka_headers_new(1) in order to create new values and to avoid double frees.

[edenhill]

The produce headers are owned by the application.