Dev

CMakeLists.txt

@@ -17,7 +17,7 @@ if(${tntver} VERSION_LESS 1.7.4.291)
     message(FATAL_ERROR "Tarantool >= 1.7.4-291 is required")
 endif()
 
-if(STATIC_BUILD)
+if(STATIC_BUILD OR $ENV{STATIC_BUILD})
     include(buildRdKafka)
     buildrdkafka()
 else()

Makefile

@@ -97,11 +97,11 @@ tests-dep:
 tests-run:
 	cd ./tests && \
 		. venv/bin/activate && \
-		pytest -vv && \
+		pytest -W ignore -vv && \
 		deactivate
 
 test-run-with-docker: tests-dep docker-run-all
-	sleep 5
+	sleep 10
 
 	docker run \
     		--net=${NETWORK} \
@@ -145,6 +145,12 @@ test-run-with-docker: tests-dep docker-run-all
 		kafka-topics --create --topic test_multi_consume_2 --partitions 1 --replication-factor 1 \
 		--if-not-exists --zookeeper zookeeper:2181
 
+	docker run \
+		--net=${NETWORK} \
+		--rm confluentinc/cp-kafka:5.0.0 \
+		kafka-topics --create --topic test_consuming_from_last_committed_offset --partitions 1 --replication-factor 1 \
+		--if-not-exists --zookeeper zookeeper:2181
+
 	cd ./tests && \
 		python3 -m venv venv && \
 		. venv/bin/activate && \
@@ -153,7 +159,7 @@ test-run-with-docker: tests-dep docker-run-all
 
 	cd ./tests && \
 		. venv/bin/activate && \
-		pytest -vv && \
+		pytest -W ignore -vv && \
 		deactivate
 
 #######################################################################

README.md

@@ -50,6 +50,9 @@ tarantoolctl rocks STATIC_BUILD=ON install kafka
     local log_callback = function(fac, str, level)
         log.info("got log: %d - %s - %s", level, fac, str)
     end
+    local rebalance_callback = function(msg)
+        log.info("got rebalance msg: %s", json.encode(msg))
+    end
 
     local consumer, err = tnt_kafka.Consumer.create({
         brokers = "localhost:9092", -- brokers for bootstrap
@@ -61,6 +64,7 @@ tarantoolctl rocks STATIC_BUILD=ON install kafka
         }, -- options for librdkafka
         error_callback = error_callback, -- optional callback for errors
         log_callback = log_callback, -- optional callback for logs and debug messages
+        rebalance_callback = rebalance_callback,  -- optional callback for rebalance messages
         default_topic_options = {
             ["auto.offset.reset"] = "earliest",
         }, -- optional default topic options
@@ -126,6 +130,9 @@ tarantoolctl rocks STATIC_BUILD=ON install kafka
     local log_callback = function(fac, str, level)
         log.info("got log: %d - %s - %s", level, fac, str)
     end
+    local rebalance_callback = function(msg)
+        log.info("got rebalance msg: %s", json.encode(msg))
+    end
 
     local consumer, err = tnt_kafka.Consumer.create({
         brokers = "localhost:9092", -- brokers for bootstrap
@@ -137,6 +144,7 @@ tarantoolctl rocks STATIC_BUILD=ON install kafka
         }, -- options for librdkafka
         error_callback = error_callback, -- optional callback for errors
         log_callback = log_callback, -- optional callback for logs and debug messages
+        rebalance_callback = rebalance_callback,  -- optional callback for rebalance messages
         default_topic_options = {
             ["auto.offset.reset"] = "earliest",
         }, -- optional default topic options
@@ -316,7 +324,7 @@ because `rd_kafka_destroy` sometimes hangs forever.
 
 ### Async
 
-Result: over 150000 produced messages per second on macbook pro 2016
+Result: over 160000 produced messages per second on macbook pro 2016
 
 Local run in docker:
 ```bash
@@ -340,7 +348,7 @@ Local run in docker:
 
 ### Auto offset store enabled
 
-Result: over 140000 consumed messages per second on macbook pro 2016
+Result: over 190000 consumed messages per second on macbook pro 2016
 
 Local run in docker:
 ```bash
@@ -351,7 +359,7 @@ Local run in docker:
 
 ### Manual offset store
 
-Result: over 140000 consumed messages per second on macbook pro 2016
+Result: over 190000 consumed messages per second on macbook pro 2016
 
 Local run in docker:
 ```bash

benchmarks/sync_producer.lua

@@ -17,7 +17,7 @@ local function produce()
     local producer, err = tnt_kafka.Producer.create({
         brokers = "kafka:9092",
         options = {
-            ["queue.buffering.max.ms"] = "100",
+            ["queue.buffering.max.ms"] = "50",
         }
     })
     if err ~= nil then
@@ -27,7 +27,7 @@ local function produce()
 
     local before = clock.monotonic64()
     local input_ch = fiber.channel();
-    for i = 1, 12000 do
+    for i = 1, 10000 do
         fiber.create(function()
             while true do
                 if input_ch:is_closed() then
@@ -57,6 +57,9 @@ local function produce()
 
     for i = 1, 10000000 do
         input_ch:put(i)
+        if i % 10000 == 0 then
+            fiber.yield()
+        end
     end
 
     input_ch:close()

docker/Dockerfile

@@ -2,12 +2,19 @@ FROM tarantool/tarantool:1.x-centos7
 
 RUN yum update -y
 
-RUN yum install -y librdkafka librdkafka-devel cmake gcc tarantool-devel
+RUN yum install -y cmake \
+                   gcc \
+                   gcc-c++ \
+                   cyrus-sasl-lib \
+                   openssl-libs \
+                   tarantool-devel
 
 ADD . /opt/tarantool
 
 WORKDIR /opt/tarantool
 
+ENV STATIC_BUILD ON
+
 RUN cmake .
 
 RUN make

kafka/CMakeLists.txt

@@ -9,7 +9,7 @@ endif(APPLE)
 
 target_link_libraries(tntkafka pthread)
 
-if (STATIC_BUILD)
+if (STATIC_BUILD OR $ENV{STATIC_BUILD})
 	add_dependencies(tntkafka rdkafka)
 	target_link_libraries(tntkafka ${CMAKE_SOURCE_DIR}/librdkafka/src/librdkafka.a)
 else()