added base consumer implementation

Author: and.sergeev

docker/Dockerfile

@@ -6,4 +6,4 @@ ADD . /opt/tarantool
 
 WORKDIR /opt/tarantool
 
-ENTRYPOINT tarantool /opt/tarantool/tests/producer.lua
+ENTRYPOINT tarantool /opt/tarantool/tests/consumer.lua

kafka/consumer.lua

@@ -0,0 +1,222 @@
+local ffi = require('ffi')
+local box = require('box')
+local fiber = require('fiber')
+local librdkafka = require('kafka.librdkafka')
+
+local ConsumerConfig = {}
+
+ConsumerConfig.__index = ConsumerConfig
+
+function ConsumerConfig.create(brokers_list)
+    assert(brokers_list ~= nil)
+
+    local config = {
+        _brokers_list = brokers_list,
+        _options = {},
+    }
+    setmetatable(config, ConsumerConfig)
+    return config
+end
+
+function ConsumerConfig:get_brokers_list()
+    return self._brokers_list
+end
+
+function ConsumerConfig:set_option(name, value)
+    self._options[name] = value
+end
+
+function ConsumerConfig:get_options()
+    return self._options
+end
+
+local ConsumerMessage = {}
+
+ConsumerMessage.__index = ConsumerMessage
+
+function ConsumerMessage.create(rd_message)
+    local msg = {
+        _rd_message = rd_message,
+        _value = nil,
+        _topic = nil,
+        _partition = nil,
+        _offset = nil,
+    }
+    ffi.gc(msg, function(...)
+        librdkafka.rd_kafka_message_destroy(...)
+    end)
+    setmetatable(msg, ConsumerMessage)
+    return msg
+end
+
+function ConsumerMessage:value()
+    if self._value == nil then
+        self._value = ffi.string(self._rd_message.payload)
+    end
+    return self._value
+end
+
+function ConsumerMessage:topic()
+    if self._topic == nil then
+        self._topic = ffi.string(librdkafka.rd_kafka_topic_name(self._rd_message.rkt))
+    end
+    return self._topic
+end
+
+function ConsumerMessage:partition()
+    if self._partition == nil then
+        self._partition = 1
+    end
+    return self._partition
+end
+
+function ConsumerMessage:offset()
+    if self._offset == nil then
+        self._offset = 1
+    end
+    return self._offset
+end
+
+local Consumer = {}
+
+Consumer.__index = Consumer
+
+function Consumer.create(config)
+    assert(config ~= nil)
+
+    local consumer = {
+        config = config,
+        _rd_consumer = {},
+        _output_ch = nil,
+    }
+    setmetatable(consumer, Consumer)
+    return consumer
+end
+
+function Consumer:_get_consumer_rd_config()
+    local rd_config = librdkafka.rd_kafka_conf_new()
+
+-- FIXME: почему мы здесь получаем segfault, а в продьюсере с таким же кодом все ок?
+--    ffi.gc(rd_config, function (rd_config)
+--        librdkafka.rd_kafka_conf_destroy(rd_config)
+--    end)
+
+    local ERRLEN = 256
+    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
+
+    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()
+        end
+    end
+end
+
+jit.off(Consumer._poll)
+
+function Consumer:start()
+    local rd_config, err = self:_get_consumer_rd_config()
+    if err ~= nil then
+        return err
+    end
+
+    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
+
+    for _, broker in ipairs(self.config:get_brokers_list()) do
+        librdkafka.rd_kafka_brokers_add(rd_consumer, broker)
+    end
+
+    self._rd_consumer = rd_consumer
+
+    self._output_ch = fiber.channel(100)
+
+    self._poll_fiber = fiber.create(function()
+        self:_poll()
+    end)
+end
+
+function Consumer:stop(timeout_ms)
+    if self._rd_consumer == nil then
+        return "'stop' method must be called only after consumer was started "
+    end
+
+    if timeout_ms == nil then
+        timeout_ms = 1000
+    end
+
+    self._poll_fiber:cancel()
+    self._output_ch:close()
+
+    -- FIXME: handle this error
+    local err = librdkafka.rd_kafka_consumer_close(self._rd_consumer)
+
+    librdkafka.rd_kafka_destroy(self._rd_consumer)
+    librdkafka.rd_kafka_wait_destroyed(timeout_ms)
+    self._rd_consumer = nil
+
+    return nil
+end
+
+function Consumer:subscribe(topics)
+    if self._rd_consumer == nil then
+        return "'add_topic' method must be called only after consumer was started "
+    end
+
+    local list = librdkafka.rd_kafka_topic_partition_list_new(#topics)
+    for _, topic in ipairs(topics) do
+        librdkafka.rd_kafka_topic_partition_list_add(list, topic, 0)
+    end
+
+    local err = nil
+    local err_no = librdkafka.rd_kafka_subscribe(self._rd_consumer, list)
+    if err_no ~= librdkafka.RD_KAFKA_RESP_ERR_NO_ERROR then
+        err = ffi.string(librdkafka.rd_kafka_err2str(err_no))
+    end
+
+    librdkafka.rd_kafka_topic_partition_list_destroy(list)
+
+    return err
+end
+
+function Consumer:output()
+    if self._rd_consumer == nil then
+        return nil, "'output' method must be called only after consumer was started "
+    end
+
+    return self._output_ch, nil
+end
+
+function Consumer:commit_async(message)
+    if self._rd_consumer == nil then
+        return "'commit' method must be called only after consumer was started "
+    end
+
+    local err_no = librdkafka.rd_kafka_commit_message(self._rd_consumer, message._rd_message, 1)
+    if err_no ~= librdkafka.RD_KAFKA_RESP_ERR_NO_ERROR then
+        return ffi.string(librdkafka.rd_kafka_err2str(err_no))
+    end
+
+    return nil
+end
+
+return {
+    ConsumerConfig = ConsumerConfig,
+    Consumer = Consumer,
+}

kafka/librdkafka.lua

@@ -24,6 +24,8 @@ ffi.cdef[[
         void                *_private;
     } rd_kafka_message_t;
 
+    void rd_kafka_message_destroy(rd_kafka_message_t *rkmessage);
+
     typedef enum rd_kafka_type_t {
         RD_KAFKA_PRODUCER,
         RD_KAFKA_CONSUMER
@@ -79,6 +81,40 @@ ffi.cdef[[
     int rd_kafka_thread_cnt (void);
 
     rd_kafka_resp_err_t rd_kafka_flush (rd_kafka_t *rk, int timeout_ms);
+
+    typedef struct rd_kafka_topic_partition_s {
+        char        *topic;             /**< Topic name */
+        int32_t      partition;         /**< Partition */
+	    int64_t      offset;            /**< Offset */
+        void        *metadata;          /**< Metadata */
+        size_t       metadata_size;     /**< Metadata size */
+        void        *opaque;            /**< Application opaque */
+        rd_kafka_resp_err_t err;        /**< Error code, depending on use. */
+        void       *_private;           /**< INTERNAL USE ONLY,
+                                         *   INITIALIZE TO ZERO, DO NOT TOUCH */
+    } rd_kafka_topic_partition_t;
+
+    typedef struct rd_kafka_topic_partition_list_s {
+        int cnt;                           /**< Current number of elements */
+        int size;                          /**< Current allocated size */
+        rd_kafka_topic_partition_t *elems; /**< Element array[] */
+    } rd_kafka_topic_partition_list_t;
+
+    rd_kafka_topic_partition_list_t *rd_kafka_topic_partition_list_new (int size);
+    void rd_kafka_topic_partition_list_destroy (rd_kafka_topic_partition_list_t *rkparlist);
+    rd_kafka_topic_partition_t *rd_kafka_topic_partition_list_add (rd_kafka_topic_partition_list_t *rktparlist, const char *topic, int32_t partition);
+
+    /**
+    * @remark Only the \c .topic field is used in the supplied \p topics list,
+    *         all other fields are ignored.
+    */
+    rd_kafka_resp_err_t rd_kafka_subscribe (rd_kafka_t *rk, const rd_kafka_topic_partition_list_t *topics);
+
+    rd_kafka_message_t *rd_kafka_consumer_poll (rd_kafka_t *rk, int timeout_ms);
+    rd_kafka_resp_err_t rd_kafka_consumer_close (rd_kafka_t *rk);
+
+    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);
 ]]
 
 local librdkafka = ffi.load("librdkafka.so.1")

kafka/producer.lua

@@ -187,6 +187,8 @@ function Producer:start()
     self._poll_fiber = fiber.create(function()
         self:_poll()
     end)
+
+    return nil
 end
 
 function Producer:stop(timeout_ms)
@@ -211,6 +213,8 @@ function Producer:stop(timeout_ms)
     librdkafka.rd_kafka_destroy(self._rd_producer)
     librdkafka.rd_kafka_wait_destroyed(timeout_ms)
     self._rd_producer = nil
+
+    return nil
 end
 
 function Producer:_get_topic_rd_config(config)

tests/consumer.lua

@@ -0,0 +1,41 @@
+local box = require('box')
+local fiber = require('fiber')
+local kafka_consumer = require('kafka.consumer')
+
+box.cfg{}
+
+local BROKERS_ADDRESS = { "kafka" }
+local TOPIC_NAME = "test_consumer"
+
+local config = kafka_consumer.ConsumerConfig.create(BROKERS_ADDRESS)
+
+local consumer = kafka_consumer.Consumer.create(config)
+
+consumer:start()
+
+consumer:subscribe({TOPIC_NAME})
+
+for i = 0, 1 do
+    fiber.create(function()
+        while true do
+            local out, err = consumer:output()
+            if err ~= nil then
+                print(string.format("got fatal error '%s'", err))
+                return
+            end
+
+            local msg = out:get()
+            if msg ~= nil then
+                print(string.format("got msg with topic='%s' partition='%s' offset='%s' value='%s'", msg:topic(), msg:partition(), msg:offset(), msg:value()))
+                local err = consumer:commit(msg)
+                if err ~= nil then
+                    print(string.format("got error '%s' while commiting msg from topic '%s'", err, msg:topic()))
+                end
+            end
+        end
+    end)
+end
+
+fiber.sleep(2)
+
+consumer:stop()