Freeze the Kafka buffer after first empty response (#5283)

* Check inside inferior streams for cancellation while reading.
* Stop reading from Kafka buffer after first empty read.

dbms/src/Formats/BlockInputStreamFromRowInputStream.cpp

@@ -63,7 +63,7 @@ Block BlockInputStreamFromRowInputStream::readImpl()
             if (rows_portion_size && batch == rows_portion_size)
             {
                 batch = 0;
-                if (!checkTimeLimit())
+                if (!checkTimeLimit() || isCancelled())
                     break;
             }
 

dbms/src/Storages/Kafka/ReadBufferFromKafkaConsumer.cpp

@@ -27,6 +27,8 @@ void ReadBufferFromKafkaConsumer::subscribe(const Names & topics)
         consumer->poll(5s);
         consumer->resume();
     }
+
+    stalled = false;
 }
 
 void ReadBufferFromKafkaConsumer::unsubscribe()
@@ -38,6 +40,12 @@ void ReadBufferFromKafkaConsumer::unsubscribe()
 /// Do commit messages implicitly after we processed the previous batch.
 bool ReadBufferFromKafkaConsumer::nextImpl()
 {
+    /// NOTE: ReadBuffer was implemented with a immutable buffer contents in mind.
+    ///       If we failed to poll any message once - don't try again.
+    ///       Otherwise, the |poll_timeout| expectations get flawn.
+    if (stalled)
+        return false;
+
     if (current == messages.end())
     {
         commit();
@@ -48,7 +56,10 @@ bool ReadBufferFromKafkaConsumer::nextImpl()
     }
 
     if (messages.empty() || current == messages.end())
+    {
+        stalled = true;
         return false;
+    }
 
     if (auto err = current->get_error())
     {

dbms/src/Storages/Kafka/ReadBufferFromKafkaConsumer.h

@@ -38,6 +38,7 @@ private:
     Poco::Logger * log;
     const size_t batch_size = 1;
     const size_t poll_timeout = 0;
+    bool stalled = false;
 
     Messages messages;
     Messages::const_iterator current;