expose stats_cb #55

hqin · 2016-10-26T19:02:42Z

Resolves issue #43
@edenhill @ewencp Please review the changes

ghost · 2016-10-26T19:03:03Z

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

Appreciation of efforts,

clabot

edenhill

Code looks good but needs some styling.
Some minor points that should be addressed as well.

I'm missing unit-tests (e.g. tests/test_misc.py) that should verify the callback is called and the JSON is parsable.

Also want to see support for stats_cb in the integration test (examples/integration_test.py), it could extract some interesting value (such as avg rtt for brokers) and present that.

It also needs to be added to the documentation in docs/index.rst

edenhill · 2016-10-26T19:11:28Z

confluent_kafka/src/confluent_kafka.c

+	CallState *cs=NULL;
+
+	cs = CallState_get(h);
+	if (json_len== 0 || !h->stats_cb) {


We dont need to enable the C stats_cb if no Python stats_cb has been set, which remove the !h->stats_cb check

edenhill · 2016-10-26T19:13:59Z

confluent_kafka/src/confluent_kafka.c

+static int stats_cb(rd_kafka_t *rk, char *json, size_t json_len, void *opaque)
+{
+	Handle *h = opaque;
+	PyObject *eo=NULL, *result=NULL;


Style is inconsistent, let's try to maintain same style as in existing file to ease readability.

Functions: Open-brace on same line as function decl.
Variable declarations: type SPACE variablename SPACE = SPACE value;
If statements: if SPACE (EXPR) SPACE {
EXPR: lhs SPACE operator SPACE rhs (e.g., json_len == 0)

edenhill · 2016-10-26T19:16:12Z

confluent_kafka/src/confluent_kafka.c

@@ -857,6 +884,10 @@ void Handle_clear (Handle *h) {
 		Py_DECREF(h->error_cb);
 	}

+	if (h->stats_cb) {


style: remove braces for single-line clauses (you can remove the ones for error_cb above too, shouldnt be there)

edenhill · 2016-10-26T19:17:15Z

confluent_kafka/src/confluent_kafka.c

@@ -1123,6 +1157,17 @@ rd_kafka_conf_t *common_conf_setup (rd_kafka_type_t ktype,
 			}
 			Py_DECREF(ks);
 			continue;
+		} else if (!strcmp(k, "stats_cb")) {


We might want to check here and for error_cb if the object is callable (PyCallable_Check(vo)) (Py_None is okay too)

edenhill · 2016-10-26T19:17:26Z

confluent_kafka/src/confluent_kafka.c

@@ -1174,6 +1219,11 @@ rd_kafka_conf_t *common_conf_setup (rd_kafka_type_t ktype,

 	if (h->error_cb)
 		rd_kafka_conf_set_error_cb(conf, error_cb);
+
+	if (h->stats_cb) {


style: remove braces

edenhill · 2016-10-26T19:18:37Z

examples/consumer.py

+    sys.stderr.write('Usage: %s [options..] <bootstrap-brokers> <group> <topic1> <topic2> ..\n' % program_name)
+    options='''
+ Options:
+  -T <intvl>   Enable statistics from Kafka at specified interval (ms)   


the stats are from the client, not Kafka, so might want to rephrase that to something like:
Enable client statistics at specified interval (ms)

edenhill · 2016-10-26T19:19:20Z

examples/consumer.py


+
+if __name__ == '__main__':
+    optlist, argv = getopt.getopt(sys.argv[1:], 'T:')


edenhill · 2016-10-26T19:20:43Z

examples/producer.py

 from confluent_kafka import Producer
 import sys
+import getopt


I think we only the stats_cb example in of the examples (to keep the examples small).
Remove either this one or the other one.

edenhill

Almost there!

edenhill · 2016-10-27T06:47:02Z

confluent_kafka/src/confluent_kafka.c

@@ -1146,7 +1143,7 @@ rd_kafka_conf_t *common_conf_setup (rd_kafka_type_t ktype,
 			Py_DECREF(ks);
 			continue;

-		} else if (!strcmp(k, "error_cb")) {
+		} else if (!strcmp(k, "error_cb") && PyCallable_Check(vo)) {


It will need to check PyCallable inside error_cb if-clause.
E.g.:

if (k=="error_cb") { if (!callable) { error_out("\"error_cb\" must be a callable"); return.. } h->error_cb = ..; }.

Same thing for stats_cb

edenhill · 2016-10-27T06:49:08Z

confluent_kafka/src/confluent_kafka.c

@@ -880,13 +879,11 @@ static int stats_cb(rd_kafka_t *rk, char *json, size_t json_len, void *opaque)
 * Clear Python object references in Handle
 */
 void Handle_clear (Handle *h) {
-	if (h->error_cb) {
+	if (h->error_cb) 


nitpick: There's a trailing whitespace after ). This might seem silly but it will show up as a red blob in editors and git gui :)

edenhill · 2016-10-27T06:50:35Z

examples/integration_test.py

+    def stats_cb(stats_json_str):
+        stats_json = json.loads(stats_json_str)
+        if 'test' in stats_json['topics']:
+            print("# app_offset stats for topic test partition 0: %d" % stats_json['topics']['test']['partitions']['0']['app_offset'])


Maybe put this in a try block to give a meaningful error message if those dict keys are not available

edenhill · 2016-10-27T06:52:39Z

examples/integration_test.py

+    else:
+        bar = None
+
+    while True:


Looks good, but this loop doesnt actually verify that stats were received.
I think it could probably finish on the first complete stats seen to speed things up.

edenhill · 2016-10-27T06:55:06Z

tests/test_misc.py

+    kc = confluent_kafka.Consumer(**conf)
+
+    kc.subscribe(["test"])
+    kc.poll(timeout=0.001)


Does not verify that error_cb was actually called.
Maybe construct it like so, without sleep:

in test_error_cb: seen_error_cb = True .... while !seen_error_cb: kc.poll(timeout=1) kc.close()

Test timeouts will handle the case where no callback was seen

edenhill · 2016-10-27T06:55:48Z

tests/test_misc.py

+        # print(stats_json_str)
+        try:
+            stats_json = json.loads(stats_json_str)
+            if 'type' in stats_json:


Shouldnt this be an assertation for checking that the json doc actually contains something we know?

edenhill · 2016-10-27T06:56:19Z

tests/test_misc.py

+
+    kc = confluent_kafka.Consumer(**conf)
+
+    kc.subscribe(["test"])


Same thing as suggested for error_cb

edenhill · 2016-10-27T06:56:58Z

confluent_kafka/src/confluent_kafka.c

@@ -749,7 +749,7 @@ PyObject *c_parts_to_py (const rd_kafka_topic_partition_list_t *c_parts) {

 	parts = PyList_New(c_parts->cnt);

-	for (i = 0 ; i < c_parts->cnt ; i++) {
+	for (i = 0 ; i < (size_t)c_parts->cnt ; i++) {


Make these size_t changes a separate commit, for tracking.

edenhill

Almost perfect, just some small stuff now

edenhill · 2016-10-28T20:57:31Z

examples/integration_test.py

        stats_json = json.loads(stats_json_str)
        if 'test' in stats_json['topics']:
-            print("# app_offset stats for topic test partition 0: %d" % stats_json['topics']['test']['partitions']['0']['app_offset'])
+            app_offset = stats_json['topics']['test']['partitions']['0']['app_offset']


probably want to wrap this in a try: if any of the sub-dict keys are missing and just print the json blob on failure.

edenhill · 2016-10-28T20:58:59Z

tests/test_misc.py

 def test_error_cb():
    """ Tests error_cb. """

    def error_cb(error_msg):
-        print('OK: error_cb() called')
+        global seen_error_cb
+        seen_error_cb = True
        assert error_msg.code() in (confluent_kafka.KafkaError._TRANSPORT, confluent_kafka.KafkaError._ALL_BROKERS_DOWN)

    conf = {'bootstrap.servers': 'localhost:9093', # Purposely cause connection refused error


This is actually probably the second most likely port to find a broker on localhost, so maybe you want to connect to localhost:22 or somesuch that will definately not a Kafka broker.

edenhill · 2016-10-29T06:24:02Z

Also need to add documentation for stats_cb in docs/index.rst to (do a git grep error_cb in the repo root to see what and where needs to be added)

edenhill · 2016-11-02T17:59:30Z

docs/index.rst

@@ -81,6 +81,8 @@ The Python bindings also provide some additional configuration properties:
 * ``error_cb(kafka.KafkaError)``: Callback for generic/global error events. This callback is served by
  poll().

+* ``stats_cb(json_str)``: Callback for statistics data. This callback is triggered by poll() every ``statistics.interval.ms`` (needs to be configured separately). Function argument ``json_str`` is a str instance of a JSON document containing statistics data.


docs are good, but split up the lines at about 80 columns so it is readable in the terminal too.

edenhill · 2016-11-02T18:18:41Z

Awesome!

expose stats_cb

201b6b3

edenhill suggested changes Oct 26, 2016

View reviewed changes

hqin added 2 commits October 26, 2016 14:49

working in progress

8f88eb1

fixes based on pull request review feedback

73b363f

edenhill suggested changes Oct 27, 2016

View reviewed changes

hqin added 2 commits October 28, 2016 11:53

fixes based on pull request review feedback

d7ab475

removed extra space

eb31ace

edenhill suggested changes Oct 28, 2016

View reviewed changes

add documentation for stats_cb

90d11df

edenhill suggested changes Nov 2, 2016

View reviewed changes

reformatted to be more readable in terminal

495a836

edenhill approved these changes Nov 2, 2016

View reviewed changes

edenhill merged commit ff1f073 into confluentinc:master Nov 2, 2016

edenhill mentioned this pull request Nov 2, 2016

Jdennison/expose stats callback #7

Closed

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

expose stats_cb #55

expose stats_cb #55

hqin commented Oct 26, 2016 •

edited

Loading

ghost commented Oct 26, 2016

edenhill left a comment

edenhill Oct 26, 2016

edenhill Oct 26, 2016

edenhill Oct 26, 2016

edenhill Oct 26, 2016

edenhill Oct 26, 2016

edenhill Oct 26, 2016

edenhill Oct 26, 2016

edenhill Oct 26, 2016

edenhill left a comment

edenhill Oct 27, 2016

edenhill Oct 27, 2016

edenhill Oct 27, 2016

edenhill Oct 27, 2016

edenhill Oct 27, 2016

edenhill Oct 27, 2016

edenhill Oct 27, 2016

edenhill Oct 27, 2016

edenhill left a comment

edenhill Oct 28, 2016

edenhill Oct 28, 2016

edenhill commented Oct 29, 2016

edenhill Nov 2, 2016

edenhill commented Nov 2, 2016



		if __name__ == '__main__':
		optlist, argv = getopt.getopt(sys.argv[1:], 'T:')

expose stats_cb #55

expose stats_cb #55

Conversation

hqin commented Oct 26, 2016 • edited Loading

ghost 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

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

edenhill commented Oct 29, 2016

Choose a reason for hiding this comment

edenhill commented Nov 2, 2016

hqin commented Oct 26, 2016 •

edited

Loading