Add timestamp() to librdkafka messages #50

qix · 2016-09-30T01:32:38Z

tstype still needs to be exposed somehow, not sure of the best way to do that.

The whitespace in confluent_kafka.c is a mix of tabs/spaces as well as spaces in blank lines which my editor automatically removed.

ghost · 2016-09-30T01:32:40Z

It looks like @qix hasn't signed our Contributor License Agreement, yet.

Appreciation of efforts,

clabot

qix · 2016-09-30T01:35:52Z

CLA signed.

ewencp

Seems reasonable. One nit, and of course the timestamp type is still missing. @edenhill do you want to take a look?

ewencp · 2016-10-06T23:53:37Z

confluent_kafka/src/confluent_kafka.c

@@ -326,6 +326,14 @@ static PyObject *Message_offset (Message *self, PyObject *ignore) {
 }


+static PyObject *Message_timestamp (Message *self, PyObject *ignore) {
+	if (self->timestamp > 0)


Should this be >=?

It should depend on the tstype, not the value of the timestamp.
E.g., even a seemingly invalid timestamp with a proper tstype is valid (because it might mean something in the future)

edenhill

Also needs tests:

add to tests/test_Consumer.py (these are broker-less unit tests so just check that the API works and returns tstype none)
add to integration tests in examples/integration_test.py (this should check that the returned timestamp actually makes sense (which requires a recent enough broker..), e.g. tstype is append and timestamp is reasonably within range to time.time()

edenhill · 2016-10-07T07:19:21Z

confluent_kafka/src/confluent_kafka.c

@@ -362,6 +370,11 @@ static PyMethodDef Message_methods[] = {
 	  "  :rtype: int or None\n"
 	  "\n"
 	},
+	{ "timestamp", (PyCFunction)Message_timestamp, METH_NOARGS,


It needs to return the timestamp type as well (none, append, create).
Im not sure whether it should be some class level constants or strings, probably the former. @ewencp?

Please add the timestamp type as a class level constant, e.g.:
TIMESTAMP_NOT_AVAILABLE, TIMESTAMP_CREATE_TIME, TIMESTAMP_APPEND_TIME
Map it directly to the librdkafka counterparts.

edenhill · 2016-10-07T07:20:26Z

confluent_kafka/src/confluent_kafka.c

@@ -495,6 +508,12 @@ PyObject *Message_new0 (const rd_kafka_message_t *rkm) {
 	self->partition = rkm->partition;
 	self->offset = rkm->offset;

+	rd_kafka_timestamp_type_t tstype;


Put variable declarations at the beginning of the scope to be C89 safe

edenhill · 2016-10-07T07:21:55Z

confluent_kafka/src/confluent_kafka.c

@@ -326,6 +326,14 @@ static PyObject *Message_offset (Message *self, PyObject *ignore) {
 }


+static PyObject *Message_timestamp (Message *self, PyObject *ignore) {
+	if (self->timestamp > 0)


It should depend on the tstype, not the value of the timestamp.
E.g., even a seemingly invalid timestamp with a proper tstype is valid (because it might mean something in the future)

qix · 2016-10-26T19:17:25Z

Sorry about the delay, this was blocked on a librdkafka bug: confluentinc/librdkafka#858

I added a tstype() function but didn't add the class level constants. I feel like that is the correct approach, but didn't know what to name them and that can be done in a separate commit.
Message object wasn't available in test_Consumer so couldn't add a test there without large changes.
The integration "test" doesn't have any assertions and isn't written as a test. I added tstype / timestamp to the print statement and manually verified.

edenhill

Thank you, almost there.

Some remaining actions:

Change timestamp() to return a tstype, timestamp tuple.
Define tstype constants
Add unit-tests
Improve documentation

edenhill · 2016-11-08T10:13:06Z

confluent_kafka/src/confluent_kafka.c

@@ -327,13 +327,18 @@ static PyObject *Message_offset (Message *self, PyObject *ignore) {


 static PyObject *Message_timestamp (Message *self, PyObject *ignore) {


I think it would be better to have a single timestamp() method that returns a tuple: tstype, timestamp

edenhill · 2016-11-08T10:17:19Z

confluent_kafka/src/confluent_kafka.c

@@ -362,6 +370,11 @@ static PyMethodDef Message_methods[] = {
 	  "  :rtype: int or None\n"
 	  "\n"
 	},
+	{ "timestamp", (PyCFunction)Message_timestamp, METH_NOARGS,


Please add the timestamp type as a class level constant, e.g.:
TIMESTAMP_NOT_AVAILABLE, TIMESTAMP_CREATE_TIME, TIMESTAMP_APPEND_TIME
Map it directly to the librdkafka counterparts.

edenhill · 2016-11-08T10:19:59Z

confluent_kafka/src/confluent_kafka.c

@@ -362,6 +370,11 @@ static PyMethodDef Message_methods[] = {
 	  "  :rtype: int or None\n"
 	  "\n"
 	},
+	{ "timestamp", (PyCFunction)Message_timestamp, METH_NOARGS,
+	  "  :returns: message timestamp or None if not available.\n"
+	  "  :rtype: int or None\n"


It should probably return a local timestamp in Python format (e.g float), and the doc string should be updated to reflect what is being returned.
The timestamp() method should also return the tstype (tstype,timestamp) tuple

qix · 2016-11-08T19:15:56Z

Hi @edenhill, I've made most of the changes you requested, but don't have too much time to work on the very specifics especially if there's going to be a lot of back and forth. Feel free to make the remaining changes yourself if you want.

Some thoughts/issues:

You do not have any unit tests that actually get a Message object, so adding that is a much larger change. I got _Consumer working but that meant commenting out the .unsubscribe line and tweaking it a lot.
This package has minimal documentation already. I had to browse the source to figure out how to use it, I don't know where you're expecting me to add documentation.
I don't believe converting it to a float is future safe. If you're expecting other uses of the timestamp field based on tstype they might not all be milliseconds so / 1000 isn't a safe assumption. I decided to leave as is with the tuple.
I made the api return the tuple even if tstype=0. Might want to consider returning None in that case, but that will cause (tstype, timestamp) = msg.timestamp() to fail.

qix · 2016-11-08T19:16:50Z

p.s. We released https://github.com/smyte/kafka_store today which uses a fork based on my original commit. As soon as we get this merged in and a new release is out I'll remove the fork and move it back over.

edenhill · 2016-11-09T12:34:26Z

Thanks a lot for your effort!

PR #50 message timestamps - additional docs, etc

updated readme

ewencp reviewed Oct 6, 2016

View reviewed changes

edenhill suggested changes Oct 7, 2016

View reviewed changes

Add timestamp() to librdkafka messages

2e6c56d

qix force-pushed the master branch from f00ec40 to c520205 Compare October 26, 2016 19:11

Updates for requested changes

339cb33

qix force-pushed the master branch from c520205 to 339cb33 Compare October 26, 2016 19:20

edenhill suggested changes Nov 8, 2016

View reviewed changes

Include constants, add unit test, use tuple

a8d2701

edenhill changed the base branch from master to PR50_msg_timestamps November 9, 2016 12:34

edenhill merged commit 2252d24 into confluentinc:PR50_msg_timestamps Nov 9, 2016

edenhill added a commit that referenced this pull request Nov 9, 2016

Merge pull request #61 from confluentinc/PR50_msg_timestamps

312bd1f

PR #50 message timestamps - additional docs, etc

dtheodor pushed a commit to dtheodor/confluent-kafka-python that referenced this pull request Sep 4, 2018

Merge pull request confluentinc#50 from mhowlett/rm

181a992

updated readme

odemkovych mentioned this pull request Jun 8, 2023

Consumer close() function hungs and prevent app exit #1573

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add timestamp() to librdkafka messages #50

Add timestamp() to librdkafka messages #50

qix commented Sep 30, 2016

ghost commented Sep 30, 2016

qix commented Sep 30, 2016

ewencp left a comment

ewencp Oct 6, 2016

edenhill Oct 7, 2016

edenhill left a comment

edenhill Oct 7, 2016

edenhill Nov 8, 2016

edenhill Oct 7, 2016

edenhill Oct 7, 2016

qix commented Oct 26, 2016

edenhill left a comment

edenhill Nov 8, 2016

edenhill Nov 8, 2016

edenhill Nov 8, 2016

qix commented Nov 8, 2016

qix commented Nov 8, 2016

edenhill commented Nov 9, 2016

		@@ -327,13 +327,18 @@ static PyObject Message_offset (Message self, PyObject *ignore) {


		static PyObject Message_timestamp (Message self, PyObject *ignore) {

Add timestamp() to librdkafka messages #50

Add timestamp() to librdkafka messages #50

Conversation

qix commented Sep 30, 2016

ghost commented Sep 30, 2016

qix commented Sep 30, 2016

ewencp 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

qix commented Oct 26, 2016

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

qix commented Nov 8, 2016

qix commented Nov 8, 2016

edenhill commented Nov 9, 2016