fixes based on pull request review feedback

Author: hqin

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 < (size_t)c_parts->cnt ; i++) {
+	for (i = 0 ; i < c_parts->cnt ; i++) {
 		const rd_kafka_topic_partition_t *rktpar = &c_parts->elems[i];
 		PyList_SET_ITEM(parts, i,
 				TopicPartition_new0(
@@ -778,7 +778,7 @@ rd_kafka_topic_partition_list_t *py_to_c_parts (PyObject *plist) {
 
 	c_parts = rd_kafka_topic_partition_list_new((int)PyList_Size(plist));
 
-	for (i = 0 ; i < (size_t)PyList_Size(plist) ; i++) {
+	for (i = 0 ; i < PyList_Size(plist) ; i++) {
 		TopicPartition *tp = (TopicPartition *)
 			PyList_GetItem(plist, i);
 
@@ -879,10 +879,10 @@ 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)
 		Py_DECREF(h->error_cb);
 
-	if (h->stats_cb) 
+	if (h->stats_cb)
 		Py_DECREF(h->stats_cb);
 
 	PyThread_delete_key(h->tlskey);
@@ -1143,7 +1143,16 @@ rd_kafka_conf_t *common_conf_setup (rd_kafka_type_t ktype,
 			Py_DECREF(ks);
 			continue;
 
-		} else if (!strcmp(k, "error_cb") && PyCallable_Check(vo)) {
+		} else if (!strcmp(k, "error_cb")) {
+			if (!PyCallable_Check(vo)) {
+				PyErr_SetString(PyExc_TypeError,
+						"expected error_cb property "
+						"as a callable function");
+				rd_kafka_topic_conf_destroy(tconf);
+				rd_kafka_conf_destroy(conf);
+				Py_DECREF(ks);
+				return NULL;
+                        }
 			if (h->error_cb) {
 				Py_DECREF(h->error_cb);
 				h->error_cb = NULL;
@@ -1154,7 +1163,17 @@ rd_kafka_conf_t *common_conf_setup (rd_kafka_type_t ktype,
 			}
 			Py_DECREF(ks);
 			continue;
-		} else if (!strcmp(k, "stats_cb") && PyCallable_Check(vo)) {
+		} else if (!strcmp(k, "stats_cb")) {
+			if (!PyCallable_Check(vo)) {
+				PyErr_SetString(PyExc_TypeError,
+						"expected stats_cb property "
+						"as a callable function");
+				rd_kafka_topic_conf_destroy(tconf);
+				rd_kafka_conf_destroy(conf);
+				Py_DECREF(ks);
+				return NULL;
+                        }
+			
 			if (h->stats_cb) {
 				Py_DECREF(h->stats_cb);
 				h->stats_cb = NULL;

examples/integration_test.py

@@ -36,7 +36,8 @@
 bootstrap_servers = 'localhost'
 
 
-
+# global variable to be set by stats_cb call back function
+good_stats_cb_result = False
 
 def error_cb (err):
     print('Error: %s' % err)
@@ -359,9 +360,13 @@ def verify_stats_cb():
     """ Verify stats_cb """
 
     def stats_cb(stats_json_str):
+        global good_stats_cb_result
         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']
+            if app_offset > 0:
+                print("# app_offset stats for topic test partition 0: %d" % app_offset)
+                good_stats_cb_result = True
 
     conf = {'bootstrap.servers': bootstrap_servers,
             'group.id': uuid.uuid1(),
@@ -388,7 +393,7 @@ def stats_cb(stats_json_str):
     else:
         bar = None
 
-    while True:
+    while not good_stats_cb_result:
         # Consume until EOF or error
 
         msg = c.poll(timeout=20.0)

tests/test_misc.py

@@ -16,11 +16,15 @@ def test_version():
     assert len(sver) > 0
     assert iver > 0
 
+# global variable for error_cb call back function
+seen_error_cb = False
+
 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
@@ -31,26 +35,23 @@ def error_cb(error_msg):
            }
 
     kc = confluent_kafka.Consumer(**conf)
-
     kc.subscribe(["test"])
-    kc.poll(timeout=0.001)
-    time.sleep(1)
-    kc.unsubscribe()
+    while not seen_error_cb:
+        kc.poll(timeout=1)
 
     kc.close()
 
+# global variable for stats_cb call back function
+seen_stats_cb = False
+
 def test_stats_cb():
     """ Tests stats_cb. """
 
     def stats_cb(stats_json_str):
-        # print(stats_json_str)
-        try:
-            stats_json = json.loads(stats_json_str)
-            if 'type' in stats_json:
-                print("stats_cb: type=%s" % stats_json['type'])
-            print('OK: stats_cb() called')
-        except Exception as e:
-            assert False
+        global seen_stats_cb
+        seen_stats_cb = True
+        stats_json = json.loads(stats_json_str)
+        assert len(stats_json['name']) > 0
 
     conf = {'group.id':'test',
             'socket.timeout.ms':'100',
@@ -62,8 +63,7 @@ def stats_cb(stats_json_str):
     kc = confluent_kafka.Consumer(**conf)
 
     kc.subscribe(["test"])
-    kc.poll(timeout=0.001)
-    time.sleep(1)
-
+    while  not seen_stats_cb:
+        kc.poll(timeout=1)
     kc.close()