MQ plugin for Kafka: rdkafka config param log_level sets the log level for

the messaging plugin as well

trackrdrd/src/mq/kafka/README.rst

@@ -168,8 +168,6 @@ Parameter                           Description
 ``topic``                           Name of the Kafka topic to which messages
                                     are sent (required)
 ----------------------------------- --------------------------------------------
-``mq.debug``                        If set to true, then log at DEBUG level
------------------------------------ --------------------------------------------
 ``worker.shutdown.timeout.ms``      If non-zero, workers will wait this long
                                     before they shut down for acknowledgements
                                     that all of the messages that they produced
@@ -186,6 +184,12 @@ the ``rdkafka`` client, as documented at::
 
 	https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
 
+The ``rdkafka`` parameter ``log_level`` sets the log level for both
+the ``rdkafka`` client and the messaging plugin. Note that the value
+of this parameter is a numeric syslog(3) "serverity level" (cf. RFC
+5424 section 6.2.1, table 2), so INFO level is 6 (the default) and
+DEBUG level is 7.
+
 The following ``rdkafka`` parameters in the config file are ignored
 (they are set internally by the messaging plugin, or are only relevant
 to consumers):

trackrdrd/src/mq/kafka/config.c

@@ -122,17 +122,15 @@ CONF_Add(const char *lval, const char *rval)
             return EINVAL;
         return(0);
     }
-    /* XXX: use the rdkakfka param "log_level" instead */
-    if (strcmp(lval, "mq.debug") == 0) {
-        if (strcmp(rval, "1") == 0
-            || strcasecmp(rval, "true") == 0
-            || strcasecmp(rval, "yes") == 0
-            || strcasecmp(rval, "on") == 0)
-            loglvl = LOG_DEBUG;
-        else if (strcmp(rval, "0") != 0
-                 && strcasecmp(rval, "false") != 0
-                 && strcasecmp(rval, "no") != 0
-                 && strcasecmp(rval, "off") != 0)
+    if (strcmp(lval, "log_level") == 0) {
+        unsigned l;
+        if ((err = conf_getUnsignedInt(rval, &l)) != 0)
+            return(err);
+        if (loglvl > LOG_DEBUG)
+            return EINVAL;
+        loglvl = l;
+        result = rd_kafka_conf_set(conf, lval, rval, errstr, LINE_MAX);
+        if (result != RD_KAFKA_CONF_OK)
             return EINVAL;
         return(0);
     }
@@ -156,5 +154,4 @@ CONF_Dump(void)
     MQ_LOG_Log(LOG_DEBUG, "topic = %s", topic);
     MQ_LOG_Log(LOG_DEBUG, "worker.shutdown.timeout.ms = %u",
                wrk_shutdown_timeout);
-    // leaving out mq.debug for now
 }

trackrdrd/src/mq/kafka/test/kafka.conf

@@ -4,5 +4,5 @@ zookeeper.connect = localhost:2181
 zookeeper.connection.timeout.ms = 10000
 zookeeper.log = zoo.log
 topic = libtrackrdr_kafka_test
-mq.debug = true
+log_level = 7
 debug = all