first attempt at batch consumption of messages #282

tburmeister · 2017-12-04T23:48:30Z

This is a first attempt at adding batch consumption, a la #252

Have not done a ton of testing yet, but I am able to retrieve messages. Let me know what you think.

ghost · 2017-12-04T23:48:32Z

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

The purpose of a CLA is to ensure that the guardian of a project's outputs has the necessary ownership or grants of rights over all contributions to allow them to distribute under the chosen licence.
Wikipedia

You can read and sign our full Contributor License Agreement here.

Once you've signed reply with [clabot:check] to prove it.

Appreciation of efforts,

clabot

tburmeister · 2017-12-04T23:49:24Z

confluent_kafka/src/Consumer.c

@@ -15,6 +15,7 @@
 */

 #include "confluent_kafka.h"
+#include <stdio.h>


Had some debug printfs in earlier, will remove this.

tburmeister · 2017-12-04T23:49:43Z

confluent_kafka/src/Consumer.c

+					 (Py_ssize_t *)&num_messages, &tmout))
+		return NULL;
+
+        CallState_begin(self, &cs); // This unlocks GIL


The comment was for me; will remove.

tburmeister · 2017-12-04T23:50:46Z

confluent_kafka/src/Consumer.c

+
+        CallState_begin(self, &cs); // This unlocks GIL
+
+        rkqu = rd_kafka_queue_get_consumer(self->rk);


Think this returns NULL if consumer was not properly initialized, which leads to a segfault further down; maybe just need to check this and exit if NULL?

That should not happen, so no need to check

tburmeister · 2017-12-04T23:51:51Z

confluent_kafka/src/Consumer.c

+                                        PyObject *kwargs) {
+        size_t num_messages = 100;
+        double tmout = -1.0f;
+        static char *kws[] = { "timeout", "num_messages", NULL };


This is backwards, should be { "num_messages", "timeout", NULL }.

edenhill · 2017-12-05T10:31:34Z

confluent_kafka/src/Consumer.c

@@ -743,6 +743,67 @@ static PyObject *Consumer_poll (Handle *self, PyObject *args,
 }


+static PyObject *Consumer_consume (Handle *self, PyObject *args,
+                                        PyObject *kwargs) {
+        size_t num_messages = 100;


1 seems like a better default.

edenhill · 2017-12-05T10:32:02Z

confluent_kafka/src/Consumer.c

+
+        CallState_begin(self, &cs);
+
+        rkqu = rd_kafka_queue_get_consumer(self->rk);


optional future improvement: cache this on self

edenhill · 2017-12-05T10:33:25Z

confluent_kafka/src/Consumer.c

+                return NULL;
+        }
+
+        if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|nd", kws,


Use I for num_messages (unsigned int) to avoid checking for < 0.

edenhill · 2017-12-05T10:36:25Z

confluent_kafka/src/Consumer.c

+        }
+
+        if (n < 0) {
+                free(rkmessages);


Should raise exception from rd_kafka_last_error() here, something like:
cfl_PyErr_Format(rd_kafka_last_error(), "%s", rd_kafka_err2str(rd_kafka_last_error());

and return NULL (to raise the exception)

edenhill · 2017-12-05T10:38:26Z

confluent_kafka/src/Consumer.c

+        if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|nd", kws,
+					 (Py_ssize_t *)&num_messages, &tmout))
+		return NULL;
+


Might want to make sure num_messages is not completely out of bounds since we're using the value directly to allocate memory.
If it is higher than 1M messages, raise an ValueError with a proper error message

edenhill · 2017-12-05T10:39:15Z

confluent_kafka/src/Consumer.c

+        Py_ssize_t n = (Py_ssize_t)rd_kafka_consume_batch_queue(rkqu,
+                tmout >= 0 ? (int)(tmout * 1000.0f) : -1,
+                rkmessages,
+                num_messages > 0 ? (size_t)num_messages : 1);


can drop this check if using unsigned int

edenhill · 2017-12-05T10:40:38Z

confluent_kafka/src/Consumer.c

+	  "  .. note: Callbacks may be called from this method, "
+	  "such as ``on_assign``, ``on_revoke``, et.al.\n"
+	  "\n"
+	  "  :param int num_messages: Maximum number of messages to return.\n"


(default: 1)

edenhill · 2017-12-05T10:43:22Z

confluent_kafka/src/Consumer.c

+	  "\n"
+	  "  :param int num_messages: Maximum number of messages to return.\n"
+	  "  :param float timeout: Maximum time to block waiting for message, event or callback.\n"
+	  "  :returns: A list of Message objects or None on timeout\n"


Do we really need to use a different return type for timeout? An empty list seems easier to handle from the application's perspective.

edenhill · 2017-12-05T10:44:08Z

confluent_kafka/src/Consumer.c

+	  "such as ``on_assign``, ``on_revoke``, et.al.\n"
+	  "\n"
+	  "  :param int num_messages: Maximum number of messages to return.\n"
+	  "  :param float timeout: Maximum time to block waiting for message, event or callback.\n"


(default: infinite (-1))

edenhill · 2017-12-05T10:46:34Z

confluent_kafka/src/Consumer.c

+
+        CallState_begin(self, &cs); // This unlocks GIL
+
+        rkqu = rd_kafka_queue_get_consumer(self->rk);


That should not happen, so no need to check

edenhill

Good stuff!
What needs to be done:

minor edits from code review
unit test in tests/test_Consumer.py (to verify the API)
integration test in examples/integration_test.py (to verify it works in practice).

tburmeister · 2017-12-05T14:59:42Z

[clabot:check]

ghost · 2017-12-05T14:59:43Z

@confluentinc It looks like @tburmeister just signed our Contributor License Agreement. 👍

Always at your service,

clabot

tburmeister · 2017-12-05T15:46:17Z

Made suggested changes. Will work on unit tests and integration tests next, and might take a stab at caching the queue.

edenhill · 2017-12-05T16:02:30Z

Great! For caching the queue, to avoid locking issues, I suggest extracting the queue in Consumer_init() right after poll_set_consumer() and destryoing the queue in Consumer_close() after the rd_kafka_consumer_close() call, and in Consumer_dealloc() prior to rd_kafka_destroy() (and outside the Callstate )

edenhill · 2017-12-05T16:03:42Z

confluent_kafka/src/Consumer.c

+        }
+
+        if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|Id", kws,
+					 (unsigned int *)&num_messages, &tmout))


It is not safe to cast to integer pointers of possibly other size.
Use unsigned int for num_message's type directly to avoid the case.

edenhill · 2017-12-05T16:04:39Z

confluent_kafka/src/Consumer.c

+	  "  :param int num_messages: Maximum number of messages to return (default: 1).\n"
+	  "  :param float timeout: Maximum time to block waiting for message, event or callback (default: infinite (-1)).\n"
+	  "  :returns: A list of Message objects\n"
+	  "  :rtype: list(Message) or None\n"


Does not return None anymore.

edenhill · 2017-12-05T16:04:53Z

confluent_kafka/src/Consumer.c

+	  "\n"
+	  "  :param int num_messages: Maximum number of messages to return (default: 1).\n"
+	  "  :param float timeout: Maximum time to block waiting for message, event or callback (default: infinite (-1)).\n"
+	  "  :returns: A list of Message objects\n"


(possibly empty on timeout)

edenhill · 2017-12-05T16:05:20Z

confluent_kafka/src/Consumer.c

+	  "  :param float timeout: Maximum time to block waiting for message, event or callback (default: infinite (-1)).\n"
+	  "  :returns: A list of Message objects\n"
+	  "  :rtype: list(Message) or None\n"
+          "  :raises: RuntimeError if called on a closed consumer\n"


or KafkaError in case of internal error.

tburmeister · 2017-12-05T16:15:24Z

Missed your comment before pushing my last commit with caching, but that all makes sense.

tburmeister · 2017-12-05T19:14:03Z

I added some tests, but I'm having trouble running the tests because synchronous commits crash on my machine; commit_return.c_parts appears to be NULL in Consumer_commit.

edenhill · 2017-12-05T19:31:09Z

What librdkafka version?

tburmeister · 2017-12-05T19:44:23Z

Version 0.11.0.

Edit: upgraded to 0.11.3 but still segfaulting.

edenhill · 2017-12-05T20:20:19Z

Ack, do you have a gdb backtrace to share?
list and bt

edenhill

Getting closer! :)

edenhill · 2017-12-05T20:40:51Z

examples/integration_test.py

+
+            if msgcnt == 1:
+                t_first_msg = time.time()
+            if msgcnt >= max_msgcnt:


edenhill · 2017-12-05T20:41:13Z

examples/integration_test.py

+            if msgcnt >= max_msgcnt:
+                break
+
+        if msgcnt >= max_msgcnt:


move this to the while ..

edenhill · 2017-12-05T20:41:43Z

examples/integration_test.py

+
+    if msgcnt > 0:
+        t_spent = time.time() - t_first_msg
+        print('%d messages (%.2fMb) consumed in %.3fs: %d msgs/s, %.2f Mb/s' %


what thruput are you getting with the batch interface, compared to the standard one?

Having trouble running the integration test, but running a quick test I wrote on an actual Kafka stream I got:

consume took 1.33 seconds 1000000 messages, 662.56 MB, 497.54 MB/sec poll tool 2.17 seconds 1000000 messages, 662.62 MB, 304.57 MB/sec

So definitely an improvement, though I found I had to tune the batch size to get the best results, e.g. 1k messages performs better than 10k messages or 100 messages.

edenhill · 2017-12-05T20:47:28Z

examples/integration_test.py

+        # Consume until EOF or error
+
+        # Consume message (error()==0) or event (error()!=0)
+        msglist = c.consume(max_msgcnt, 1.0)


Knowing that there is at least N messages in the topic (thanks to the produce tests), and where N > max_msgcnt (which it should be), we should check that we actually get max_msgcnt back for each call until we've hit our expected message count (which is some arbitrary value <=N).
Relying on EOF is not safe since the topic may feature multiple partitions and we don't know what the message distribution is like.

edenhill · 2017-12-05T20:49:42Z

tests/test_Consumer.py

@@ -41,6 +41,17 @@ def dummy_assign_revoke(consumer, partitions):
    if msg is not None:
        assert msg.timestamp() == (TIMESTAMP_NOT_AVAILABLE, -1)

+    msglist = kc.consume(num_messages=10, timeout=0.001)
+    if len(msglist) == 0:


aren't we expecting exactly 0 messages here? use assert len(msglist) == 0, "expected 0 messages, not %d" % len(msglist)

edenhill · 2017-12-05T20:53:40Z

confluent_kafka/src/Consumer.c

@@ -756,6 +815,11 @@ static PyObject *Consumer_close (Handle *self, PyObject *ignore) {

        rd_kafka_consumer_close(self->rk);

+        if (self->u.Consumer.rkqu) {
+                rd_kafka_queue_destroy(self->u.Consumer.rkqu);


indent looks weird.

Will fix. I noticed that there are a bunch of tabs in the code currently, and I gather from the majority of the code and the librdkafka style guide that spaces are preferred - I can clean those up as well.

Yeah, the tabs are part of the dark past.
But you don't need to change whitespace of existing code, we'll do that later in one big go across the code base as we also instate coding guidelines. But thanks anyway!

edenhill · 2017-12-05T20:54:44Z

tests/test_Consumer.py

+                break
+
+        print('OK: consumed messages')
+


Please add a failing test for invalid number of num_messages (e.g., -100 and > 1M)

tburmeister · 2017-12-05T20:56:03Z

Struggling to get debug symbols working properly on my Mac, so the best I can give you right now is:

(gdb) bt
#0  0x0000000101be67e0 in c_parts_to_py () from /Users/tburmeister/anaconda/lib/python3.6/site-packages/confluent_kafka/cimpl.cpython-36m-darwin.so
#1  0x0000000101becceb in Consumer_commit () from /Users/tburmeister/anaconda/lib/python3.6/site-packages/confluent_kafka/cimpl.cpython-36m-darwin.so
#2  0x000000010005894a in _PyCFunction_FastCallDict () from /Users/tburmeister/anaconda/lib/libpython3.6m.dylib
#3  0x0000000100058b5b in _PyCFunction_FastCallKeywords () from /Users/tburmeister/anaconda/lib/libpython3.6m.dylib
#4  0x00000001000da787 in call_function () from /Users/tburmeister/anaconda/lib/libpython3.6m.dylib
#5  0x00000001000d7001 in _PyEval_EvalFrameDefault () from /Users/tburmeister/anaconda/lib/libpython3.6m.dylib
#6  0x00000001000db9ed in fast_function () from /Users/tburmeister/anaconda/lib/libpython3.6m.dylib
#7  0x00000001000da761 in call_function () from /Users/tburmeister/anaconda/lib/libpython3.6m.dylib
#8  0x00000001000d6f62 in _PyEval_EvalFrameDefault () from /Users/tburmeister/anaconda/lib/libpython3.6m.dylib
#9  0x00000001000db1ef in _PyEval_EvalCodeWithName () from /Users/tburmeister/anaconda/lib/libpython3.6m.dylib
#10 0x00000001000d03d4 in PyEval_EvalCode () from /Users/tburmeister/anaconda/lib/libpython3.6m.dylib
#11 0x00000001001060f1 in PyRun_FileExFlags () from /Users/tburmeister/anaconda/lib/libpython3.6m.dylib
#12 0x00000001001058a3 in PyRun_SimpleFileExFlags () from /Users/tburmeister/anaconda/lib/libpython3.6m.dylib
#13 0x000000010011e70f in Py_Main () from /Users/tburmeister/anaconda/lib/libpython3.6m.dylib
#14 0x0000000100000df8 in main ()

I do not have this issue on my Ubuntu dev machine.

edenhill · 2017-12-05T21:01:32Z

I added some tests, but I'm having trouble running the tests because synchronous commits crash on my machine; commit_return.c_parts appears to be NULL in Consumer_commit.

That is weird since there is assert just above that checks that commit_return.c_parts is not NULL:
https://github.com/confluentinc/confluent-kafka-python/blob/master/confluent_kafka/src/Consumer.c#L480

Can you add some printfs around that part to figure out what is going on?

tburmeister · 2017-12-05T21:28:19Z

The assert seems to not actually be checked. I added:

                /* sync commit returns the topic,partition,offset,err list */
                assert(commit_return.c_parts);
                printf("commit_return.c_parts: %p\n", commit_return.c_parts);

at line 477, and get:

============================== Verifying Consumer ==============================
test[0]@193: key=b'0', value=b'Message #0', tstype=1, timestamp=1512499907092
test[0]@194: key=b'1', value=b'Message #1', tstype=1, timestamp=1512499907092
test[0]@195: key=b'2', value=b'Message #2', tstype=1, timestamp=1512499907092
test[0]@196: key=b'3', value=b'Message #3', tstype=1, timestamp=1512499907092
commit_return.c_parts: 0x0
Segmentation fault: 11

edenhill

only nits left now!

edenhill · 2017-12-06T08:13:13Z

examples/integration_test.py

@@ -589,21 +589,17 @@ def verify_batch_consumer():
    max_msgcnt = 100
    msgcnt = 0

-    while True:
+    while msgcnt < max_msgcnt:


let max_msgcnt=1000 and batch_cnt=100

edenhill · 2017-12-06T08:14:23Z

examples/integration_test.py

@@ -715,12 +706,9 @@ def my_on_revoke(consumer, partitions):

            if msgcnt == 1:
                t_first_msg = time.time()
-            if msgcnt >= max_msgcnt:
+            elif msgcnt >= max_msgcnt:


can get rid of this one now with the while loop

edenhill

The crash might be that no offsets were committed, it is most likely not related to your changes though so we can fix them separately, you can ignore it for now (you can change it to return None if c_parts is empty in your local code).

edenhill · 2017-12-06T15:15:38Z

examples/integration_test.py

-        # Consume message (error()==0) or event (error()!=0)
-        msglist = c.consume(max_msgcnt, 1.0)
+        # Consume messages (error()==0) or event (error()!=0)
+        msglist = c.consume(batch_cnt, 1.0)
        assert len(msglist) == max_msgcnt


This test is failing, right?
the assert should check for batch_cnt, not max_msgcnt

Yep, good catch.

edenhill · 2017-12-06T15:16:29Z

examples/integration_test.py

-        # Consume message (error()==0) or event (error()!=0)
-        msglist = c.consume(max_msgcnt, 1.0)
+        # Consume messages (error()==0) or event (error()!=0)
+        msglist = c.consume(batch_cnt, 1.0)


Make the test more robust by setting timeout to >session.timeout.ms, such as 10

edenhill · 2017-12-06T20:18:24Z

examples/integration_test.py

+
+        # Consume messages (error()==0) or event (error()!=0)
+        msglist = c.consume(batch_cnt, 1.0)
+        assert len(msglist) == max_msgcnt


this should compare to batch_cnt

edenhill · 2017-12-06T20:19:01Z

examples/integration_test.py

+
+            msgcnt += 1
+
+        print('max_msgcnt %d reached' % msgcnt)


wrong indent, should be outside while loop.

edenhill · 2017-12-06T20:20:28Z

examples/integration_test.py

+
+    if msgcnt > 0:
+        t_spent = time.time() - t_first_msg
+        print('%d messages (%.2fMb) consumed in %.3fs: %d msgs/s, %.2f Mb/s' %


tburmeister · 2017-12-06T21:35:17Z

Ohhhh I figured out why commits were crashing and it is my fault - will push a fix soon.

tburmeister · 2017-12-06T21:44:18Z

I can now run the full integration test locally, without errors. I really should have confirmed the tests worked locally on master first.

tburmeister · 2017-12-07T14:54:58Z

So what needs to be done to get this merged? Looking at the CI tests, it looks like it's passing the actual tests but failing because of pep8 violations, which I believe are addressed in #279

edenhill · 2017-12-07T15:17:41Z

Huge thanks for doing this! 😍

tburmeister · 2017-12-07T15:30:23Z

Awesome! Thanks! This feature will be super useful to me. By the way, we are big librdkafka users at AppNexus.

isamaru · 2017-12-13T18:15:04Z

Thanks! This is very useful to us. Any timeline for when this becomes part of a pypi release?

edenhill · 2017-12-15T09:27:15Z

@isamaru We're aiming for a feature release end of january.

messense · 2017-12-15T09:33:02Z

@edenhill Maybe you can upload a pre-release version to PyPi so we can try it? Thanks!

https://packaging.python.org/tutorials/distributing-packages/#pre-release-versioning

aswinjoseroy · 2018-02-22T09:37:41Z

Awesome work adding this feature! Any updates as to when the pypi release will happen?

edenhill · 2018-02-22T09:38:32Z

Final release in about 2 weeks, we'll have an rc out hopefully this week

aswinjoseroy · 2018-07-26T14:37:30Z

Is the batch consumer released yet? Any links to usage / any documentation etc are highly appreciated! Thanks!

alannesta · 2018-10-12T18:30:33Z

Is the batch consumer released yet? Any links to usage / any documentation etc are highly appreciated! Thanks!

@aswinjoseroy Take a look at the consume method in the doc. That's the new function added in this PR
https://docs.confluent.io/current/clients/confluent-kafka-python/#confluent_kafka.Consumer.consume

first attempt at batch consumption of messages

9089894

tburmeister commented Dec 4, 2017

View reviewed changes

small fixes

fca622d

edenhill reviewed Dec 5, 2017

View reviewed changes

edenhill suggested changes Dec 5, 2017

View reviewed changes

made suggested changes

1aaf4fe

edenhill suggested changes Dec 5, 2017

View reviewed changes

cache consumer queue in consumer

3867ea0

Taylor Burmeister added 2 commits December 5, 2017 11:38

some changes to consumer queue caching and other suggested changes

e091bab

added unit and integration tests for batch consumption

d38691b

small fix

eb88d46

edenhill suggested changes Dec 5, 2017

View reviewed changes

more suggested changes

9a5444b

edenhill suggested changes Dec 6, 2017

View reviewed changes

some more suggested changes

c212230

edenhill reviewed Dec 6, 2017

View reviewed changes

edenhill suggested changes Dec 6, 2017

View reviewed changes

a few more fixes

64a325a

fixed commit crash and build on Ubuntu

316f5ed

edenhill approved these changes Dec 6, 2017

View reviewed changes

edenhill merged commit c864b64 into confluentinc:master Dec 7, 2017


		CallState_begin(self, &cs); // This unlocks GIL

		rkqu = rd_kafka_queue_get_consumer(self->rk);


		CallState_begin(self, &cs);

		rkqu = rd_kafka_queue_get_consumer(self->rk);

first attempt at batch consumption of messages #282

first attempt at batch consumption of messages #282

Conversation

tburmeister commented Dec 4, 2017

ghost commented Dec 4, 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

edenhill left a comment

Choose a reason for hiding this comment

tburmeister commented Dec 5, 2017

ghost commented Dec 5, 2017

tburmeister commented Dec 5, 2017

edenhill commented Dec 5, 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

tburmeister commented Dec 5, 2017

tburmeister commented Dec 5, 2017

edenhill commented Dec 5, 2017

tburmeister commented Dec 5, 2017 • edited

edenhill commented Dec 5, 2017

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

tburmeister commented Dec 5, 2017 • edited

edenhill commented Dec 5, 2017

tburmeister commented Dec 5, 2017 • edited

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

tburmeister commented Dec 6, 2017

tburmeister commented Dec 6, 2017

tburmeister commented Dec 7, 2017

edenhill commented Dec 7, 2017

tburmeister commented Dec 7, 2017

isamaru commented Dec 13, 2017 • edited

edenhill commented Dec 15, 2017

messense commented Dec 15, 2017

aswinjoseroy commented Feb 22, 2018

edenhill commented Feb 22, 2018

aswinjoseroy commented Jul 26, 2018

alannesta commented Oct 12, 2018 • edited

tburmeister commented Dec 5, 2017 •

edited

tburmeister commented Dec 5, 2017 •

edited

tburmeister commented Dec 5, 2017 •

edited

isamaru commented Dec 13, 2017 •

edited

alannesta commented Oct 12, 2018 •

edited