Ensure plugin path handled prior to plugin configs #404

rnpridgeon · 2018-06-18T15:32:10Z

No description provided.

edenhill

Great stuff, some comments!

edenhill · 2018-06-18T16:08:57Z

confluent_kafka/src/confluent_kafka.c

+
+                v = cfl_PyUnistr_AsUTF8(vs, &vs8);
+
+                if(rd_kafka_conf_set(conf, "plugin.library.paths", v, NULL, 0) != RD_KAFKA_CONF_OK) {


if (

keep under 80 columns wide

edenhill · 2018-06-18T16:10:07Z

tests/test_Consumer.py

+
+
+@pytest.skipif(len([True for x in (".so", ".dylib", ".dll") if os.path.exists("monitoring-interceptor" + x)]) > 0,
+               reason="requires confluent-librdkafka-plugins to be installed")


and copied to the current directory

edenhill · 2018-06-18T16:10:37Z

tests/test_Consumer.py

+    """
+    Consumer({'confluent.monitoring.interceptor.publishMs': 1000,
+              'confluent.monitoring.interceptor.sessionDurationMs': 1000,
+              'plugin.library.paths': 'monitoring-interceptor.dylib',


Skip the file extension, handled by the librd of kafka

edenhill · 2018-06-18T16:11:29Z

tests/test_Producer.py

@@ -191,3 +192,17 @@ def test_set_invalid_partitioner_murmur():
    with pytest.raises(KafkaException) as e:
        Producer({'partitioner': 'murmur'})
    assert 'Invalid value for configuration property "partitioner": murmur' in str(e)
+
+
+@pytest.skipif(len([True for x in (".so", ".dylib", ".dll") if os.path.exists("monitoring-interceptor" + x)]) > 0,


Instead of duplicating this test, move it to a common file (e.g., test_misc) and parameterize the client types: producer, consumer, admin

edenhill · 2018-06-18T16:12:58Z

confluent_kafka/src/confluent_kafka.c

@@ -1568,6 +1568,40 @@ rd_kafka_conf_t *common_conf_setup (rd_kafka_type_t ktype,
        /* Enable valid offsets in delivery reports */
        rd_kafka_topic_conf_set(tconf, "produce.offset.report", "true", NULL, 0);

+        /* Preload Plugins since dicts are unordered */


Preloading a library/plugin is something else, I think you can instead flesh out the comment a bit, something like:
Plugins must be configured prior to any configuration properties they support, since dicts are unordered
we explicitly check for, set, and delete the plugin paths here to make sure they are applied prior to
their configuration.

edenhill · 2018-06-19T09:50:19Z

confluent_kafka/src/confluent_kafka.c

@@ -1568,7 +1568,11 @@ rd_kafka_conf_t *common_conf_setup (rd_kafka_type_t ktype,
        /* Enable valid offsets in delivery reports */
        rd_kafka_topic_conf_set(tconf, "produce.offset.report", "true", NULL, 0);

-        /* Preload Plugins since dicts are unordered */
+       /*


Indent seems off, /* should probably be one space to the right

edenhill · 2018-06-19T09:51:08Z

confluent_kafka/src/confluent_kafka.c

-        /* Preload Plugins since dicts are unordered */
+       /*
+        * Plugins must be configured prior to handling any of their configuration properties
+        * Dicts are unordered so we explicitly check for, set, and delete the plugin paths here


end sentence with period.

edenhill · 2018-06-19T09:54:14Z

confluent_kafka/src/confluent_kafka.c

+       /*
+        * Plugins must be configured prior to handling any of their configuration properties
+        * Dicts are unordered so we explicitly check for, set, and delete the plugin paths here
+        * This ensures all configuration properties are handled in the right order


end sentence with period.

edenhill · 2018-06-19T09:54:29Z

confluent_kafka/src/confluent_kafka.c

+                if (rd_kafka_conf_set(conf, "plugin.library.paths", v, NULL, 0)
+                   != RD_KAFKA_CONF_OK) {
+                        cfl_PyErr_Format(RD_KAFKA_RESP_ERR__INVALID_ARG,
+                                "%s", NULL);


"%s", NULL will raise an exception with the message "(null)".
Instead pass errstr to conf_set, and then use that here.

Yep sorry that was left over from when I did the initial POC

edenhill · 2018-06-19T09:55:52Z

tests/test_misc.py

+    """
+    Interceptor configs can only be handled after the plugin has been loaded not before.
+    """
+    for fn in [confluent_kafka.Producer, confluent_kafka.Consumer, confluent_kafka.admin.AdminClient]:


I was thinking you could use parameterization: https://docs.pytest.org/en/latest/parametrize.html

edenhill · 2018-06-19T13:22:24Z

confluent_kafka/src/confluent_kafka.c

@@ -1568,6 +1568,47 @@ rd_kafka_conf_t *common_conf_setup (rd_kafka_type_t ktype,
        /* Enable valid offsets in delivery reports */
        rd_kafka_topic_conf_set(tconf, "produce.offset.report", "true", NULL, 0);

+       /*


indent should be 8 whitespaces, not 7 my man

edenhill · 2018-06-19T13:22:42Z

confluent_kafka/src/confluent_kafka.c

+                if (rd_kafka_conf_set(conf, "plugin.library.paths", v, errstr, sizeof(errstr))
+                   != RD_KAFKA_CONF_OK) {
+                        cfl_PyErr_Format(RD_KAFKA_RESP_ERR__INVALID_ARG,
+                                "%s", NULL);


pass errstr instead of NULL here

edenhill · 2018-06-19T13:24:05Z

confluent_kafka/src/confluent_kafka.c

+                Py_XDECREF(vs8);
+                Py_DECREF(vs);
+
+                PyDict_DelItemString(confdict, "plugin.library.paths");


ooh, come to think of it, we shouldnt mutate the passed dict.
So either we make a copy, or, just skip this DelItemString, which will make plugin.library.paths be applied twice but I believe that librdkafka will avoid re-loading a previously loaded plugin so it should work.

edenhill · 2018-06-19T13:24:37Z

tests/test_misc.py

+    (confluent_kafka.Producer),
+    (confluent_kafka.admin.AdminClient),
+])
+def test_unordered_dict(init_func):


rnpridgeon · 2018-06-20T13:50:38Z

@edenhill good call on mutating the application's dict. I have made the requested changes

edenhill · 2018-06-20T14:18:04Z

confluent_kafka/src/confluent_kafka.c

@@ -1530,7 +1530,7 @@ rd_kafka_conf_t *common_conf_setup (rd_kafka_type_t ktype,
         * precedence in case of duplicate keys.
         * All keys map to configuration properties.
         */


Extend this comment to mention why we make a copy of the dict

edenhill · 2018-06-20T14:19:04Z

confluent_kafka/src/confluent_kafka.c

+        /*
+         * Plugins must be configured prior to handling any of their configuration properties.
+         * Dicts are unordered so we explicitly check for, set, and delete the plugin paths here.
+         * This ensures all configuration properties are handled in the right order.


s/all/plugin/

edenhill · 2018-06-20T14:19:31Z

confluent_kafka/src/confluent_kafka.c

+                                "as type unicode string");
+                        rd_kafka_topic_conf_destroy(tconf);
+                        rd_kafka_conf_destroy(conf);
+                        return NULL;


confdict is leaked here, do Py_DECREF

edenhill · 2018-06-20T14:19:39Z

confluent_kafka/src/confluent_kafka.c

+                        Py_XDECREF(vs8);
+                        Py_XDECREF(vs);
+
+                        return NULL;


also check the other returns below

rnpridgeon · 2018-06-20T17:35:15Z

@edenhill , slightly less leaky. There are some whitespace changes but I believe they add to readability. If its a problem I can remove them .

edenhill · 2018-06-20T17:43:12Z

confluent_kafka/src/confluent_kafka.c

@@ -1529,6 +1529,8 @@ rd_kafka_conf_t *common_conf_setup (rd_kafka_type_t ktype,
         * When both args and kwargs are present the kwargs take
         * precedence in case of duplicate keys.
         * All keys map to configuration properties.
+         *
+         * Copy configuration dict to avoid manipulating application config


nit (but that's what you want..): How about a trailing period?

edenhill · 2018-06-20T18:37:01Z

confluent_kafka/src/confluent_kafka.c

@@ -1530,7 +1530,7 @@ rd_kafka_conf_t *common_conf_setup (rd_kafka_type_t ktype,
         * precedence in case of duplicate keys.
         * All keys map to configuration properties.
         *
-         * Copy configuration dict to avoid manipulating application config
+         * Copy configuration dict to avoid manipulating application config.


edenhill

LGTM!

.......

rnpridgeon requested a review from edenhill June 18, 2018 15:32

edenhill suggested changes Jun 18, 2018

View reviewed changes

edenhill suggested changes Jun 19, 2018

View reviewed changes

edenhill suggested changes Jun 20, 2018

View reviewed changes

rnpridgeon changed the title ~~Ensure Plugins are loaded prior to Plugin configs~~ Ensure Plugin path handled prior to Plugin configs Jun 20, 2018

edenhill suggested changes Jun 20, 2018

View reviewed changes

edenhill reviewed Jun 20, 2018

View reviewed changes

edenhill approved these changes Jun 20, 2018

View reviewed changes

rnpridgeon added 2 commits June 21, 2018 11:12

Ensure plugin configs handled in proper order

84bf3fd

Copy config dict priort to mutating contents

8ae36c2

.......

rnpridgeon changed the title ~~Ensure Plugin path handled prior to Plugin configs~~ Ensure plugin path handled prior to plugin configs Jun 21, 2018

rnpridgeon merged commit 1937de2 into confluentinc:master Jun 21, 2018

rnpridgeon deleted the ordered_dict branch July 3, 2018 23:15

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Ensure plugin path handled prior to plugin configs #404

Ensure plugin path handled prior to plugin configs #404

rnpridgeon commented Jun 18, 2018

edenhill left a comment

edenhill Jun 18, 2018

edenhill Jun 18, 2018

edenhill Jun 18, 2018

edenhill Jun 18, 2018

edenhill Jun 18, 2018

edenhill Jun 19, 2018

edenhill Jun 19, 2018

edenhill Jun 19, 2018

edenhill Jun 19, 2018

rnpridgeon Jun 19, 2018

edenhill Jun 19, 2018

edenhill Jun 19, 2018

edenhill Jun 19, 2018

edenhill Jun 19, 2018

edenhill Jun 19, 2018

rnpridgeon commented Jun 20, 2018

edenhill Jun 20, 2018

edenhill Jun 20, 2018

edenhill Jun 20, 2018

edenhill Jun 20, 2018

edenhill Jun 20, 2018

rnpridgeon commented Jun 20, 2018

edenhill Jun 20, 2018

rnpridgeon Jun 20, 2018

edenhill Jun 20, 2018

edenhill left a comment


		v = cfl_PyUnistr_AsUTF8(vs, &vs8);

		if(rd_kafka_conf_set(conf, "plugin.library.paths", v, NULL, 0) != RD_KAFKA_CONF_OK) {



		@pytest.skipif(len([True for x in (".so", ".dylib", ".dll") if os.path.exists("monitoring-interceptor" + x)]) > 0,
		reason="requires confluent-librdkafka-plugins to be installed")

Ensure plugin path handled prior to plugin configs #404

Ensure plugin path handled prior to plugin configs #404

Conversation

rnpridgeon commented Jun 18, 2018

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rnpridgeon commented Jun 20, 2018

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

rnpridgeon commented Jun 20, 2018

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