add metadata field to topicpartition #1126

jliunyu · 2021-05-23T04:57:14Z

No description provided.

edenhill

Don't forget a CHANGELOG.md update

src/confluent_kafka/src/confluent_kafka.c

tests/test_TopicPartition.py

A Kafka offset commit message includes optional metadata, but python client doesn't support it yet.

edenhill

Metadata is binary and needs to be handled as such. The string operations will not work.

CHANGELOG.md

edenhill · 2021-06-04T08:13:12Z

src/confluent_kafka/src/confluent_kafka.c

 	return 0;
 }

 static void TopicPartition_setup (TopicPartition *self, const char *topic,
 				  int partition, long long offset,
+				  const char *metadata,


metadata is a binary blob, not a string, so we can't use strlen() to know the end of the data, the size must be passed explicitly. The corresponding Python type should be bytes (for py3, there are macros to help with py2 in confluent_kafka.h)

This does not seem to be fixed

src/confluent_kafka/src/confluent_kafka.c

tests/integration/consumer/test_consumer_topicpartition_metadata.py

src/confluent_kafka/src/confluent_kafka.c

edenhill

Looks good but needs some further tweaking

CHANGELOG.md

src/confluent_kafka/src/confluent_kafka.c

src/confluent_kafka/src/confluent_kafka.h

tests/integration/consumer/test_consumer_topicpartition_metadata.py

jliunyu · 2021-07-14T06:23:56Z

tests/integration/consumer/test_consumer_topicpartition_metadata.py

+    for tp in offsets:
+        assert tp.metadata is None
+
+    metadata = '\x01abcdefgafd\xff'


For string include byte works fine on linux, but report errors on mac, not sure if it's caused by different gcc or others, will waiting for ci to verify this.

Add some nul-characters into the string too.

And add the b byte prefix: metadata = b'...'

edenhill · 2021-09-20T08:09:42Z

src/confluent_kafka/src/confluent_kafka.c

 	return 0;
 }

 static void TopicPartition_setup (TopicPartition *self, const char *topic,
 				  int partition, long long offset,
+				  const char *metadata,


This does not seem to be fixed

edenhill · 2021-09-20T08:10:13Z

src/confluent_kafka/src/confluent_kafka.c

-	if (!PyArg_ParseTupleAndKeywords(args, kwargs, "s|iL", kws,
-					 &topic, &partition, &offset))
+
+	if (PyArg_ParseTupleAndKeywords (args, kwargs, "s|iLy", kws,


remove space before paren

The y formatter token docs says:

The bytes buffer must not contain embedded null bytes

But this is binary data, so it can contain anything. Need to use another formatter token here.

edenhill · 2021-09-20T08:13:13Z

src/confluent_kafka/src/confluent_kafka.c

 			       NULL };
-
-	if (!PyArg_ParseTupleAndKeywords(args, kwargs, "s|iL", kws,


We typically try to keep the happy-path (non-error path) unindented, so restore the original:

if (!Py..)) return -1; TopicPartition_setup(),..

edenhill · 2021-09-20T08:14:36Z

src/confluent_kafka/src/confluent_kafka.c

@@ -1084,6 +1106,7 @@ PyObject *c_parts_to_py (const rd_kafka_topic_partition_list_t *c_parts) {
 rd_kafka_topic_partition_list_t *py_to_c_parts (PyObject *plist) {
 	rd_kafka_topic_partition_list_t *c_parts;
 	size_t i;
+        rd_kafka_topic_partition_t *rktpar;


Since this variable is only used in the for-loop, put it there instead.

edenhill · 2021-09-20T08:15:45Z

src/confluent_kafka/src/confluent_kafka.c

+						       &metadata_size)) == -1) {
+ 				PyErr_SetString(PyExc_RuntimeError,
+						"Failed to parse metadata "
+						"object");


The partition list is leaked

edenhill · 2021-09-20T08:16:57Z

tests/integration/consumer/test_consumer_topicpartition_metadata.py

+        c.commit(offsets=[TopicPartition(topic, 0, 0)],
+                 asynchronous=False)
+    except KafkaException as e:
+        print('commit failed with %s' % e)


Not sure you want to catch this and just print, if the commit fails the test should fail, so just leave it without a try:

edenhill · 2021-09-20T08:18:05Z

tests/integration/consumer/test_consumer_topicpartition_metadata.py

+    for tp in offsets:
+        assert tp.metadata is None
+
+    metadata = '\x01abcdefgafd\xff'


Add some nul-characters into the string too.

And add the b byte prefix: metadata = b'...'

edenhill · 2021-09-20T08:18:12Z

tests/integration/consumer/test_consumer_topicpartition_metadata.py

+    try:
+        c.commit(offsets=[TopicPartition(topic, 0, 1, binarymetadata)],
+                 asynchronous=False)
+    except KafkaException as e:


edenhill · 2021-09-20T08:19:13Z

tests/integration/consumer/test_consumer_topicpartition_metadata.py

+        assert tp.metadata is None
+
+    metadata = '\x01abcdefgafd\xff'
+    binarymetadata = str.encode(metadata)


I think it is simpler to just define metadata as a byte string and not do any encoding.

edenhill · 2021-09-20T08:20:00Z

tests/integration/consumer/test_consumer_topicpartition_metadata.py

+    except KafkaException as e:
+        print('commit failed with %s' % e)
+
+    offsets = c.committed([TopicPartition(topic, 0)], timeout=100)


add: assert len(offsets) == 1

milindl · 2022-10-11T07:05:08Z

Closing this since #1410 is merged.

jliunyu marked this pull request as draft May 23, 2021 04:57

jliunyu force-pushed the metadataToTopicPartition branch 2 times, most recently from 9f53a6e to a000505 Compare May 26, 2021 02:09

jliunyu linked an issue May 26, 2021 that may be closed by this pull request

Is it possible to pass metadata when committing offsets? #1039

Closed

1 task

jliunyu requested review from edenhill and mhowlett May 26, 2021 02:11

jliunyu marked this pull request as ready for review May 26, 2021 02:11

jliunyu marked this pull request as draft May 26, 2021 07:28

edenhill suggested changes May 26, 2021

View reviewed changes

jliunyu force-pushed the metadataToTopicPartition branch 4 times, most recently from 8030780 to 27060b4 Compare June 2, 2021 07:58

add metadata field to topic partition

a0a30b7

A Kafka offset commit message includes optional metadata, but python client doesn't support it yet.

jliunyu force-pushed the metadataToTopicPartition branch from 27060b4 to a0a30b7 Compare June 2, 2021 08:00

jliunyu marked this pull request as ready for review June 2, 2021 08:01

edenhill suggested changes Jun 4, 2021

View reviewed changes

edenhill reviewed Jun 10, 2021

View reviewed changes

src/confluent_kafka/src/confluent_kafka.c Outdated Show resolved Hide resolved

jliunyu force-pushed the metadataToTopicPartition branch 2 times, most recently from dcc236d to d709176 Compare June 15, 2021 04:53

to binary

6db4f9c

jliunyu force-pushed the metadataToTopicPartition branch from d8a4a5e to 6db4f9c Compare June 15, 2021 06:41

jliunyu added 2 commits June 14, 2021 23:43

copyright

a6ff722

Update rktpar assignment

1d9148d

edenhill suggested changes Jun 30, 2021

View reviewed changes

jliunyu added 3 commits July 13, 2021 16:55

metadata_size remove

a65a1d6

format

4e369fe

fix integration test

ebeebad

jliunyu commented Jul 14, 2021

View reviewed changes

edenhill suggested changes Sep 20, 2021

View reviewed changes

milindl mentioned this pull request Aug 22, 2022

Add metadata to TopicPartition #1410

Merged

milindl closed this Oct 11, 2022

milindl mentioned this pull request Sep 25, 2023

[KIP-580] Exponential Backoff with Mock Broker Changes to Automate Testing. confluentinc/librdkafka#4422

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

add metadata field to topicpartition #1126

add metadata field to topicpartition #1126

jliunyu commented May 23, 2021

edenhill left a comment

edenhill left a comment

edenhill Jun 4, 2021

edenhill Sep 20, 2021

edenhill left a comment

jliunyu Jul 14, 2021

edenhill Sep 20, 2021

edenhill Sep 20, 2021

edenhill Sep 20, 2021

edenhill Sep 20, 2021

edenhill Sep 20, 2021

edenhill Sep 20, 2021

edenhill Sep 20, 2021

edenhill Sep 20, 2021

edenhill Sep 20, 2021

edenhill Sep 20, 2021

edenhill Sep 20, 2021

edenhill Sep 20, 2021

milindl commented Oct 11, 2022

		NULL };

		if (!PyArg_ParseTupleAndKeywords(args, kwargs, "s\|iL", kws,

add metadata field to topicpartition #1126

add metadata field to topicpartition #1126

Conversation

jliunyu commented May 23, 2021

edenhill left a comment

Choose a reason for hiding this comment

edenhill left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

edenhill left a 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

milindl commented Oct 11, 2022