**Describe the bug**
Reading from Kafka table leads to segfault, if broker is unavailable.

**How to reproduce**
repro: #12487

**Error message and/or stacktrace**
```
2020.07.14 12:08:10.498536 [ 73 ] {} <Debug> StorageKafka (kafka): [rdk:CGRPQUERY] [thrd:main]: Group "test_bad_reschedule": no broker available for coordinator query: intervaled in state query-coord
2020.07.14 12:08:10.498648 [ 75 ] {} <Debug> StorageKafka (kafka): [rdk:BROKERFAIL] [thrd:kafka1:19091/bootstrap]: kafka1:19091/bootstrap: failed: err: Local: Broker transport failure: (errno: Connection refused)
2020.07.14 12:08:11.497986 [ 73 ] {} <Debug> StorageKafka (kafka): [rdk:CGRPQUERY] [thrd:main]: Group "test_bad_reschedule": no broker available for coordinator query: intervaled in state query-coord
2020.07.14 12:08:11.498442 [ 75 ] {} <Debug> StorageKafka (kafka): [rdk:BROKERFAIL] [thrd:kafka1:19091/bootstrap]: kafka1:19091/bootstrap: failed: err: Local: Broker transport failure: (errno: Connection refused)
2020.07.14 12:08:12.498978 [ 73 ] {} <Debug> StorageKafka (kafka): [rdk:CGRPQUERY] [thrd:main]: Group "test_bad_reschedule": no broker available for coordinator query: intervaled in state query-coord
2020.07.14 12:08:12.499324 [ 75 ] {} <Debug> StorageKafka (kafka): [rdk:BROKERFAIL] [thrd:kafka1:19091/bootstrap]: kafka1:19091/bootstrap: failed: err: Local: Broker transport failure: (errno: Connection refused)
2020.07.14 12:08:13.498106 [ 73 ] {} <Debug> StorageKafka (kafka): [rdk:CGRPQUERY] [thrd:main]: Group "test_bad_reschedule": no broker available for coordinator query: intervaled in state query-coord
2020.07.14 12:08:13.498343 [ 75 ] {} <Debug> StorageKafka (kafka): [rdk:BROKERFAIL] [thrd:kafka1:19091/bootstrap]: kafka1:19091/bootstrap: failed: err: Local: Broker transport failure: (errno: Connection refused)
2020.07.14 12:08:14.498127 [ 73 ] {} <Debug> StorageKafka (kafka): [rdk:CGRPQUERY] [thrd:main]: Group "test_bad_reschedule": no broker available for coordinator query: intervaled in state query-coord
2020.07.14 12:08:15.000876 [ 8 ] {} <Trace> BaseDaemon: Received signal 11
2020.07.14 12:08:15.001081 [ 76 ] {} <Fatal> BaseDaemon: ########################################
2020.07.14 12:08:15.001108 [ 76 ] {} <Fatal> BaseDaemon: (version 20.7.1.1, build id: A9EF261825E641A5) (from thread 19) (no query) Received signal Segmentation fault (11)
2020.07.14 12:08:15.001124 [ 76 ] {} <Fatal> BaseDaemon: Address: 0x78 Access: read. Address not mapped to object.
2020.07.14 12:08:15.001133 [ 76 ] {} <Fatal> BaseDaemon: Stack trace: 0xe45020b 0xe44f540 0xddb61d7 0xddb78f2 0xddb80ce 0x9b00916 0x9b0299b 0x7f9e83ad3669 0x7f9e839ea323
2020.07.14 12:08:15.003837 [ 76 ] {} <Fatal> BaseDaemon: 3. /home/ap11/ClickHouse/build_clang/../src/Storages/Kafka/ReadBufferFromKafkaConsumer.h:48: DB::StorageKafka::streamToViews() @ 0xe45020b in /usr/bin/clickhouse
2020.07.14 12:08:15.004697 [ 76 ] {} <Fatal> BaseDaemon: 4. /home/ap11/ClickHouse/build_clang/../src/Storages/Kafka/StorageKafka.cpp:489: DB::StorageKafka::threadFunc() @ 0xe44f540 in /usr/bin/clickhouse
2020.07.14 12:08:15.005741 [ 76 ] {} <Fatal> BaseDaemon: 5. /home/ap11/ClickHouse/build_clang/../contrib/libcxx/include/functional:0: DB::BackgroundSchedulePoolTaskInfo::execute() @ 0xddb61d7 in /usr/bin/clickhouse
2020.07.14 12:08:15.006654 [ 76 ] {} <Fatal> BaseDaemon: 6. /home/ap11/ClickHouse/build_clang/../contrib/poco/Foundation/include/Poco/AtomicCounter.h:314: DB::BackgroundSchedulePool::threadFunction() @ 0xddb78f2 in /usr/bin/clickhouse
2020.07.14 12:08:15.007399 [ 76 ] {} <Fatal> BaseDaemon: 7. /home/ap11/ClickHouse/build_clang/../src/Core/BackgroundSchedulePool.cpp:0: std::__1::__function::__func<ThreadFromGlobalPool::ThreadFromGlobalPool<DB::BackgroundSchedulePool::BackgroundSchedulePool(unsigned long, unsigned long, unsigned long, char const*)::$_1>(DB::BackgroundSchedulePool::BackgroundSchedulePool(unsigned long, unsigned long, unsigned long, char const*)::$_1&&)::'lambda'(), std::__1::allocator<ThreadFromGlobalPool::ThreadFromGlobalPool<DB::BackgroundSchedulePool::BackgroundSchedulePool(unsigned long, unsigned long, unsigned long, char const*)::$_1>(DB::BackgroundSchedulePool::BackgroundSchedulePool(unsigned long, unsigned long, unsigned long, char const*)::$_1&&)::'lambda'()>, void ()>::operator()() @ 0xddb80ce in /usr/bin/clickhouse
2020.07.14 12:08:15.008125 [ 76 ] {} <Fatal> BaseDaemon: 8. /home/ap11/ClickHouse/build_clang/../contrib/libcxx/include/atomic:1036: ThreadPoolImpl<std::__1::thread>::worker(std::__1::__list_iterator<std::__1::thread, void*>) @ 0x9b00916 in /usr/bin/clickhouse
2020.07.14 12:08:15.008244 [ 76 ] {} <Fatal> BaseDaemon: 9. /home/ap11/ClickHouse/build_clang/../contrib/libcxx/include/memory:2615: void* std::__1::__thread_proxy<std::__1::tuple<std::__1::unique_ptr<std::__1::__thread_struct, std::__1::default_delete<std::__1::__thread_struct> >, void ThreadPoolImpl<std::__1::thread>::scheduleImpl<void>(std::__1::function<void ()>, int, std::__1::optional<unsigned long>)::'lambda1'()> >(void*) @ 0x9b0299b in /usr/bin/clickhouse
2020.07.14 12:08:15.008261 [ 76 ] {} <Fatal> BaseDaemon: 10. start_thread @ 0x9669 in /usr/lib/x86_64-linux-gnu/libpthread-2.30.so
2020.07.14 12:08:15.008274 [ 76 ] {} <Fatal> BaseDaemon: 11. __clone @ 0x122323 in /usr/lib/x86_64-linux-gnu/libc-2.30.so
```

Block getHeader() const override;
    void readPrefixImpl() override;
    Block readImpl() override;
    void readSuffixImpl() override;
    void commit();
    bool isStalled() const { return buffer->isStalled(); }
private:
    StorageKafka & storage;
    StorageMetadataPtr metadata_snapshot;
    const std::shared_ptr<Context> context;
    Names column_names;
    Poco::Logger * log;
    UInt64 max_block_size;
    ConsumerBufferPtr buffer;
    bool broken = true;
    bool finished = false;
    bool commit_in_suffix;
    const Block non_virtual_header;
    const Block virtual_header;
};
}