freeze not working consumer based on new consumer api of rdlibkafka

Author: and.sergeev

kafka/consumer.lua

@@ -7,11 +7,15 @@ local ConsumerConfig = {}
 
 ConsumerConfig.__index = ConsumerConfig
 
-function ConsumerConfig.create(brokers_list)
+function ConsumerConfig.create(brokers_list, consumer_group, enable_auto_commit)
     assert(brokers_list ~= nil)
+    assert(consumer_group ~= nil)
+    assert(enable_auto_commit ~= nil)
 
     local config = {
         _brokers_list = brokers_list,
+        _consumer_group = consumer_group,
+        _enable_auto_commit = enable_auto_commit,
         _options = {},
     }
     setmetatable(config, ConsumerConfig)
@@ -22,6 +26,14 @@ function ConsumerConfig:get_brokers_list()
     return self._brokers_list
 end
 
+function ConsumerConfig:get_consumer_group()
+    return self._consumer_group
+end
+
+function ConsumerConfig:get_enable_auto_commit()
+    return self._enable_auto_commit
+end
+
 function ConsumerConfig:set_option(name, value)
     self._options[name] = value
 end
@@ -42,7 +54,7 @@ function ConsumerMessage.create(rd_message)
         _partition = nil,
         _offset = nil,
     }
-    ffi.gc(msg, function(...)
+    ffi.gc(msg._rd_message, function(...)
         librdkafka.rd_kafka_message_destroy(...)
     end)
     setmetatable(msg, ConsumerMessage)
@@ -102,24 +114,62 @@ function Consumer:_get_consumer_rd_config()
 --    end)
 
     local ERRLEN = 256
+    local errbuf = ffi.new("char[?]", ERRLEN) -- cdata objects are garbage collected
+    if librdkafka.rd_kafka_conf_set(rd_config, "group.id", tostring(self.config:get_consumer_group()), errbuf, ERRLEN) ~= librdkafka.RD_KAFKA_CONF_OK then
+        return nil, ffi.string(errbuf)
+    end
+
+    local enable_auto_commit
+    if self.config:get_enable_auto_commit() then
+        enable_auto_commit = "true"
+    else
+        enable_auto_commit = "false"
+    end
+
+    local ERRLEN = 256
+    local errbuf = ffi.new("char[?]", ERRLEN) -- cdata objects are garbage collected
+    if librdkafka.rd_kafka_conf_set(rd_config, "enable.auto.commit", enable_auto_commit, errbuf, ERRLEN) ~= librdkafka.RD_KAFKA_CONF_OK then
+        return nil, ffi.string(errbuf)
+    end
+
     for key, value in pairs(self.config:get_options()) do
         local errbuf = ffi.new("char[?]", ERRLEN) -- cdata objects are garbage collected
         if librdkafka.rd_kafka_conf_set(rd_config, key, tostring(value), errbuf, ERRLEN) ~= librdkafka.RD_KAFKA_CONF_OK then
             return nil, ffi.string(errbuf)
         end
     end
 
+    librdkafka.rd_kafka_conf_set_consume_cb(rd_config,
+        function(rkmessage)
+            print(rkmessage)
+            self._output_ch:put(ConsumerMessage.create(rkmessage))
+        end)
+
+    librdkafka.rd_kafka_conf_set_error_cb(rd_config,
+        function(rk, err, reason)
+            print("error", tonumber(err), ffi.string(reason))
+        end)
+
+
+    librdkafka.rd_kafka_conf_set_log_cb(rd_config,
+        function(rk, level, fac, buf)
+            print("log", tonumber(level), ffi.string(fac), ffi.string(buf))
+        end)
+
     return rd_config, nil
 end
 
 function Consumer:_poll()
     while true do
-        local rd_message = librdkafka.rd_kafka_consumer_poll(self._rd_consumer, 1)
-        if rd_message.err == librdkafka.RD_KAFKA_RESP_ERR_NO_ERROR then
-            self._output_ch:put(ConsumerMessage.create(rd_message))
-        else
-            fiber.yield()
+        librdkafka.rd_kafka_poll(self._rd_consumer, 10)
+        local rd_message = librdkafka.rd_kafka_consumer_poll(self._rd_consumer, 1000)
+        print(rd_message)
+        if rd_message ~= nil and rd_message.err ~= librdkafka.RD_KAFKA_RESP_ERR_NO_ERROR then
+            -- FIXME: properly log this
+            print(ffi.string(librdkafka.rd_kafka_err2str(rd_message.err)))
         end
+
+        fiber.yield()
     end
 end
 
@@ -134,13 +184,20 @@ function Consumer:start()
     local ERRLEN = 256
     local errbuf = ffi.new("char[?]", ERRLEN) -- cdata objects are garbage collected
     local rd_consumer = librdkafka.rd_kafka_new(librdkafka.RD_KAFKA_CONSUMER, rd_config, errbuf, ERRLEN)
-
     if rd_consumer == nil then
         return ffi.string(errbuf)
     end
 
+    -- redirect all events polling to rd_kafka_consumer_poll function
+    local err = librdkafka.rd_kafka_poll_set_consumer(rd_consumer)
+    if err ~= librdkafka.RD_KAFKA_RESP_ERR_NO_ERROR then
+        return ffi.string(librdkafka.rd_kafka_err2str(err))
+    end
+
     for _, broker in ipairs(self.config:get_brokers_list()) do
-        librdkafka.rd_kafka_brokers_add(rd_consumer, broker)
+        if librdkafka.rd_kafka_brokers_add(rd_consumer, broker) < 1 then
+            return "no valid brokers specified"
+        end
     end
 
     self._rd_consumer = rd_consumer
@@ -181,6 +238,8 @@ function Consumer:subscribe(topics)
 
     local list = librdkafka.rd_kafka_topic_partition_list_new(#topics)
     for _, topic in ipairs(topics) do
+        print(topic, librdkafka.RD_KAFKA_PARTITION_UA)
+--        librdkafka.rd_kafka_topic_partition_list_add(list, topic, librdkafka.RD_KAFKA_PARTITION_UA)
         librdkafka.rd_kafka_topic_partition_list_add(list, topic, 0)
     end
 

kafka/librdkafka.lua

@@ -1,6 +1,8 @@
 local ffi = require 'ffi'
 
 ffi.cdef[[
+    static const int32_t RD_KAFKA_PARTITION_UA = ((int32_t)-1);
+
     typedef struct rd_kafka_s rd_kafka_t;
     typedef struct rd_kafka_conf_s rd_kafka_conf_t;
     typedef struct rd_kafka_topic_s rd_kafka_topic_t;
@@ -46,6 +48,8 @@ ffi.cdef[[
             char *errstr, size_t errstr_size);
     void rd_kafka_conf_set_dr_msg_cb (rd_kafka_conf_t *conf, void (*dr_msg_cb) (rd_kafka_t *rk,
             const rd_kafka_message_t *rkmessage, void *opaque));
+    void rd_kafka_conf_set_consume_cb (rd_kafka_conf_t *conf, void (*consume_cb) (rd_kafka_message_t *rkmessage,
+            void *opaque));
     void rd_kafka_conf_set_error_cb (rd_kafka_conf_t *conf, void  (*error_cb) (rd_kafka_t *rk, int err,
             const char *reason, void *opaque));
     void rd_kafka_conf_set_stats_cb (rd_kafka_conf_t *conf, int (*stats_cb) (rd_kafka_t *rk, char *json,
@@ -115,6 +119,8 @@ ffi.cdef[[
 
     rd_kafka_resp_err_t rd_kafka_commit (rd_kafka_t *rk, const rd_kafka_topic_partition_list_t *offsets, int async);
     rd_kafka_resp_err_t rd_kafka_commit_message (rd_kafka_t *rk, const rd_kafka_message_t *rkmessage, int async);
+
+    rd_kafka_resp_err_t rd_kafka_poll_set_consumer (rd_kafka_t *rk);
 ]]
 
 local librdkafka = ffi.load("librdkafka.so.1")

tests/consumer.lua

@@ -1,19 +1,28 @@
 local box = require('box')
+local os = require("os")
 local fiber = require('fiber')
 local kafka_consumer = require('kafka.consumer')
 
 box.cfg{}
 
 local BROKERS_ADDRESS = { "kafka" }
-local TOPIC_NAME = "test_consumer"
+local TOPIC_NAME = "test_producer"
 
-local config = kafka_consumer.ConsumerConfig.create(BROKERS_ADDRESS)
+local config = kafka_consumer.ConsumerConfig.create(BROKERS_ADDRESS, "test_consumer5", false)
 
 local consumer = kafka_consumer.Consumer.create(config)
 
-consumer:start()
+local err = consumer:start()
+if err ~= nil then
+    print(err)
+    os.exit(1)
+end
 
-consumer:subscribe({TOPIC_NAME})
+local err = consumer:subscribe({TOPIC_NAME})
+if err ~= nil then
+    print(err)
+    os.exit(1)
+end
 
 for i = 0, 1 do
     fiber.create(function()
@@ -36,6 +45,6 @@ for i = 0, 1 do
     end)
 end
 
-fiber.sleep(2)
+fiber.sleep(10)
 
 consumer:stop()