Merge pull request #10910 from filimonov/kafka_drop_hang_fix

alexey-milovidov · web-flow · commit 0fd0711b5145 · 2020-05-17T07:13:54.000+03:00
Fix for the hang during deletion of engine=Kafka
diff --git a/src/Storages/Kafka/ReadBufferFromKafkaConsumer.cpp b/src/Storages/Kafka/ReadBufferFromKafkaConsumer.cpp
@@ -80,22 +80,9 @@ ReadBufferFromKafkaConsumer::ReadBufferFromKafkaConsumer(
     });
 }
 
-ReadBufferFromKafkaConsumer::~ReadBufferFromKafkaConsumer()
-{
-    /// NOTE: see https://github.com/edenhill/librdkafka/issues/2077
-    try
-    {
-        if (!consumer->get_subscription().empty())
-            consumer->unsubscribe();
-        if (!assignment.empty())
-            consumer->unassign();
-        while (consumer->get_consumer_queue().next_event(100ms));
-    }
-    catch (const cppkafka::HandleException & e)
-    {
-        LOG_ERROR(log, "Exception from ReadBufferFromKafkaConsumer destructor: " << e.what());
-    }
-}
+// NOTE on removed desctuctor: There is no need to unsubscribe prior to calling rd_kafka_consumer_close().
+// check: https://github.com/edenhill/librdkafka/blob/master/INTRODUCTION.md#termination
+// manual destruction was source of weird errors (hangs during droping kafka table, etc.)
 
 void ReadBufferFromKafkaConsumer::commit()
 {
@@ -226,8 +213,13 @@ void ReadBufferFromKafkaConsumer::unsubscribe()
     // it should not raise exception as used in destructor
     try
     {
-        if (!consumer->get_subscription().empty())
-            consumer->unsubscribe();
+        // From docs: Any previous subscription will be unassigned and unsubscribed first.
+        consumer->subscribe(topics);
+
+        // I wanted to avoid explicit unsubscribe as it requires draining the messages
+        // to close the consumer safely after unsubscribe
+        // see https://github.com/edenhill/librdkafka/issues/2077
+        //     https://github.com/confluentinc/confluent-kafka-go/issues/189 etc.
     }
     catch (const cppkafka::HandleException & e)
     {
diff --git a/src/Storages/Kafka/ReadBufferFromKafkaConsumer.h b/src/Storages/Kafka/ReadBufferFromKafkaConsumer.h
@@ -28,7 +28,6 @@ class ReadBufferFromKafkaConsumer : public ReadBuffer
         const std::atomic<bool> & stopped_,
         const Names & _topics
     );
-    ~ReadBufferFromKafkaConsumer() override;
 
     void allowNext() { allowed = true; } // Allow to read next message.
     void commit(); // Commit all processed messages.
@@ -64,10 +63,13 @@ class ReadBufferFromKafkaConsumer : public ReadBuffer
 
     const std::atomic<bool> & stopped;
 
+    // order is important, need to be destructed before consumer
     Messages messages;
     Messages::const_iterator current;
 
     bool rebalance_happened = false;
+
+    // order is important, need to be destructed before consumer
     cppkafka::TopicPartitionList assignment;
     const Names topics;
 
diff --git a/src/Storages/Kafka/StorageKafka.cpp b/src/Storages/Kafka/StorageKafka.cpp
@@ -235,14 +235,19 @@ ProducerBufferPtr StorageKafka::createWriteBuffer(const Block & header)
 ConsumerBufferPtr StorageKafka::createReadBuffer()
 {
     cppkafka::Configuration conf;
+
     conf.set("metadata.broker.list", brokers);
     conf.set("group.id", group);
     conf.set("client.id", VERSION_FULL);
+
     conf.set("auto.offset.reset", "smallest");     // If no offset stored for this group, read all messages from the start
+
+    updateConfiguration(conf);
+
+    // those settings should not be changed by users.
     conf.set("enable.auto.commit", "false");       // We manually commit offsets after a stream successfully finished
     conf.set("enable.auto.offset.store", "false"); // Update offset automatically - to commit them all at once.
     conf.set("enable.partition.eof", "false");     // Ignore EOF messages
-    updateConfiguration(conf);
 
     // Create a consumer and subscribe to topics
     auto consumer = std::make_shared<cppkafka::Consumer>(conf);
diff --git a/tests/integration/test_storage_kafka/test.py b/tests/integration/test_storage_kafka/test.py
@@ -198,7 +198,6 @@ def test_kafka_settings_new_syntax(kafka_cluster):
     kafka_check_result(result, True)
 
 
-@pytest.mark.skip(reason="https://github.com/edenhill/librdkafka/issues/2077")
 @pytest.mark.timeout(180)
 def test_kafka_consumer_hang(kafka_cluster):
 
@@ -219,7 +218,7 @@ def test_kafka_consumer_hang(kafka_cluster):
         CREATE MATERIALIZED VIEW test.consumer TO test.view AS SELECT * FROM test.kafka;
         ''')
 
-    time.sleep(12)
+    time.sleep(10)
     instance.query('SELECT * FROM test.view')
 
     # This should trigger heartbeat fail,
@@ -229,19 +228,23 @@ def test_kafka_consumer_hang(kafka_cluster):
     time.sleep(0.5)
     kafka_cluster.unpause_container('kafka1')
 
+    # print("Attempt to drop")
     instance.query('DROP TABLE test.kafka')
 
+    #kafka_cluster.open_bash_shell('instance')
+
     instance.query('''
         DROP TABLE test.consumer;
         DROP TABLE test.view;
     ''')
 
-    log = '/var/log/clickhouse-server/stderr.log'
-    instance.exec_in_container(['grep', '-q', 'BROKERFAIL', log])
-    instance.exec_in_container(['grep', '-q', '|ASSIGN|', log])
-    instance.exec_in_container(['grep', '-q', 'Heartbeat failed: REBALANCE_IN_PROGRESS: group is rebalancing', log])
-    instance.exec_in_container(['grep', '-q', 'Group "consumer_hang": waiting for rebalance_cb', log])
+    # original problem appearance was a sequence of the following messages in librdkafka logs:
+    # BROKERFAIL -> |ASSIGN| -> REBALANCE_IN_PROGRESS -> "waiting for rebalance_cb" (repeated forever)
+    # so it was waiting forever while the application will execute queued rebalance callback
 
+    # from a user perspective: we expect no hanging 'drop' queries
+    # 'dr'||'op' to avoid self matching
+    assert int(instance.query("select count() from system.processes where position(lower(query),'dr'||'op')>0")) == 0
 
 @pytest.mark.timeout(180)
 def test_kafka_csv_with_delimiter(kafka_cluster):
@@ -1234,7 +1237,7 @@ def test_exception_from_destructor(kafka_cluster):
         DROP TABLE test.kafka;
     ''')
 
-    kafka_cluster.open_bash_shell('instance')
+    #kafka_cluster.open_bash_shell('instance')
     assert TSV(instance.query('SELECT 1')) == TSV('1')
 
 
