Message headers support (C) #1480

edenhill · 2017-10-23T13:30:00Z

No description provided.

mhowlett · 2017-11-02T16:47:23Z

examples/rdkafka_example.c

@@ -187,6 +189,27 @@ static void msg_consume (rd_kafka_message_t *rkmessage,
 				!timestamp ? 0 :
 				(int)time(NULL) - (int)(timestamp/1000));
 		}
+
+                if (!rd_kafka_message_headers(rkmessage, &hdrs)) {


some n00b commentary for your entertainment. I find this difficult to read, though I guess it's a Cism and so no problem. I'd have written this here: if (rd_kafka_message_headers(rkmessage, &hdrs) == RD_KAFKA_RESP_ERR_NO_ERROR).

mhowlett · 2017-11-02T20:59:53Z

examples/rdkafka_example.c

+
+                        fprintf(stdout, "%% Headers:");
+
+                        while (!rd_kafka_header_iter_all(hdrs, idx++,


I'd remove the word iter from this method name since you're iterating outside the method call (with the idx++). also the word all. perhaps just rd_kafka_header_get or something?

Okay, what about rd_kafka_header_get() for the key-specific one, and rd_kafka_header_enum() or ..list() or ..get_all() for the one that iterates all keys?

mhowlett · 2017-11-02T21:13:44Z

examples/rdkafka_example.c

+                        if (!hdrs)
+                                hdrs = rd_kafka_headers_new(8);
+
+                        err = rd_kafka_header_add(hdrs, name, name_sz, val, -1);


this is a bit tricky (dual meaning of the -1), but I can't think of anything I prefer and I guess it's ok.

mhowlett · 2017-11-02T21:39:25Z

src/rdkafka.h

+ *                   null-terminated string and the length is automatically
+ *                   acquired using strlen().
+ *
+ * @returns RD_KAFKA_RESP_ERR__READ_ONLY if the headers are read-only,


RD_KAFKA_RESP_ERR__READ_ONLY is never returned I think? do you have plans for this? is it a bit odd for a C api to try and enforce this sort of thing?

I got this from the Java API, but it turned out to be an internal API that we dont need to mimic. Will remove it.

Second thought I think we'll keep it so that headers can be "closed" when final message serialization has taken place, this would stop (future) late-stage interceptors from thinking they could modify the headers after serialization.

mhowlett · 2017-11-02T21:40:42Z

src/rdkafka.h

+/**
+ * @brief Remove all headers for the given key (if any).
+ *
+ * @returns RD_KAFKA_RESP_ERR__READ_ONLY if the headers are read-only,


same comment about readonly as above

mhowlett · 2017-11-02T21:44:12Z

src/rdkafka.h

+ * @param name   Header to find (last match).
+ * @param valuep (out) Set to a (null-terminated) const pointer to the value
+ *               (may be NULL).
+ * @param sizep  (out) Set to the value's size (not including null-terminator).


perhaps include a note about what it means if sizep == -1

sizep will not be set to -1 (null values have a size of 0)

mhowlett · 2017-11-02T21:49:09Z

src/rdkafka.h

+ * @param sizep  (out) Set to the value's size (not including null-terminator).
+ */
+RD_EXPORT rd_kafka_resp_err_t
+rd_kafka_header_iter (const rd_kafka_headers_t *hdrs, size_t idx,


don't like iter in the name here either. maybe rd_kafka_header_get is appropriate for this as well (not sure of the best name).

mhowlett · 2017-11-02T21:59:08Z

src/rdkafka_buf.h

@@ -195,8 +195,6 @@ rd_tmpabuf_write_str0 (const char *func, int line,
                                "expected %"PRIusz" bytes > %"PRIusz    \
                                " remaining bytes",                     \
                                __len0, rd_kafka_buf_read_remain(rkbuf)); \
-                        (rkbuf)->rkbuf_err = RD_KAFKA_RESP_ERR__BAD_MSG; \


i'm not going to bother understanding why you took this out, I assume you know what you're doing.

parse_fail() already sets it

mhowlett · 2017-11-02T22:13:24Z

src/rdkafka_header.c

+ * @brief header_t(name) to char * comparator
+ */
+static int rd_kafka_header_cmp_str (void *_a, void *_b) {
+        const rd_kafka_header_t *a = _a;


not sure why you have void* parameters then immediately cast?

The comparator signature is int () (void *, void *), passing a function with another signature would cause a compiler warning

mhowlett · 2017-11-02T22:50:24Z

LGTM after you've addressed comments as you see fit.

almazik · 2017-11-22T15:27:07Z

Is there a progress on this merge request? This feature is important for us...

johnistan · 2017-12-18T18:23:17Z

I have played around with python support for this feature. I haven't found a public API to find out how many headers an msg has. leading to hacks like:

        //HACK to figure out size of headers
        while (!rd_kafka_header_iter_all(hdrs, header_size++,
                                         &header_key, &header_value, &header_value_size)) {
        }

Can we add a rd_kafka_header_length to the public API? Or is there an existing approach I am not seeing.

edenhill · 2017-12-18T20:39:01Z

@johnistan Good idea, will add.

edenhill · 2018-01-02T17:45:31Z

@almazik This PR is pretty much ready to be merged, just waiting final review from @mhowlett on iter API renaming.

mhowlett · 2018-01-04T03:50:17Z

@edenhill - go for it. lgtm.

edenhill requested a review from mhowlett October 23, 2017 13:30

mhowlett reviewed Nov 2, 2017

View reviewed changes

jeroenheijmans mentioned this pull request Dec 7, 2017

Message headers support confluentinc/confluent-kafka-dotnet#393

Closed

johnistan mentioned this pull request Dec 13, 2017

Support for Message headers confluentinc/confluent-kafka-python#287

Closed

edenhill added 2 commits January 2, 2018 09:19

Message Headers support

4a86230

rdlist: add remove_elem() and remove_multi_cmp()

91872da

edenhill force-pushed the headers branch from 258610e to 91872da Compare January 2, 2018 08:19

Adds header count function to public api (#1614, @johnistan)

5c59bfc

edenhill merged commit c3d50eb into master Jan 4, 2018

olib963 mentioned this pull request May 8, 2018

Feature Request - Support for Message Headers Blizzard/node-rdkafka#414

Closed

alexander-alvarez mentioned this pull request Jun 26, 2018

C++ Headers Support #1861

Closed

iradul mentioned this pull request Jul 29, 2019

Looking for more Collaborators! Blizzard/node-rdkafka#628

Open

edenhill deleted the headers branch May 8, 2020 08:57

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Message headers support (C) #1480

Message headers support (C) #1480

edenhill commented Oct 23, 2017

mhowlett Nov 2, 2017

mhowlett Nov 2, 2017

edenhill Jan 2, 2018

mhowlett Nov 2, 2017

mhowlett Nov 2, 2017

edenhill Nov 3, 2017

edenhill Jan 2, 2018

mhowlett Nov 2, 2017

mhowlett Nov 2, 2017

edenhill Nov 3, 2017

mhowlett Nov 2, 2017

mhowlett Nov 2, 2017

edenhill Nov 3, 2017

mhowlett Nov 2, 2017

edenhill Nov 3, 2017

mhowlett commented Nov 2, 2017

almazik commented Nov 22, 2017

johnistan commented Dec 18, 2017 •

edited

Loading

edenhill commented Dec 18, 2017

edenhill commented Jan 2, 2018

mhowlett commented Jan 4, 2018


		fprintf(stdout, "%% Headers:");

		while (!rd_kafka_header_iter_all(hdrs, idx++,

Message headers support (C) #1480

Message headers support (C) #1480

Conversation

edenhill commented Oct 23, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mhowlett commented Nov 2, 2017

almazik commented Nov 22, 2017

johnistan commented Dec 18, 2017 • edited Loading

edenhill commented Dec 18, 2017

edenhill commented Jan 2, 2018

mhowlett commented Jan 4, 2018

johnistan commented Dec 18, 2017 •

edited

Loading