Always resume consumer before subscription

[abyss7]

I hereby agree to the terms of the CLA available at: https://yandex.ru/legal/cla/?lang=en

For changelog. Remove if this is non-significant change.

Category (leave one):

• Bug Fix

Short description (up to few sentences):
Fix the situation when consumer got paused before subscription and not resumed afterwards.

Detailed description (optional):
This may happen when the subscription times out for example.

[abyss7]

Should fix #5846 - or at least the new logs will shred some light on the real cause of the problem.

[alexey-milovidov]

TODO: check integration tests.

[alexey-milovidov]

Segfault in stress test (nullptr dereference):

2019.07.21 02:11:19.245828 [ 358 ] {3a0c2198-33f6-4403-a373-60895003c6b0} <Debug> executeQuery: (from [::1]:47076) SELECT * FROM merge('', defaultValueOfArgumentType('\0'))
2019.07.21 02:11:19.247688 [ 359 ] {000c3d04-b785-47ad-9bf8-4ea45d89e918} <Trace> InterpreterSelectQuery: FetchColumns -> Complete
2019.07.21 02:11:19.248413 [ 359 ] {000c3d04-b785-47ad-9bf8-4ea45d89e918} <Trace> InterpreterSelectQuery: FetchColumns -> Complete
2019.07.21 02:11:19.249687 [ 358 ] {3a0c2198-33f6-4403-a373-60895003c6b0} <Debug> test_dftcv9.tab_00386 (SelectExecutor): Key condition: unknown
2019.07.21 02:11:19.249747 [ 358 ] {3a0c2198-33f6-4403-a373-60895003c6b0} <Debug> test_dftcv9.tab_00386 (SelectExecutor): MinMax index condition: unknown
2019.07.21 02:11:19.249823 [ 358 ] {3a0c2198-33f6-4403-a373-60895003c6b0} <Debug> test_dftcv9.tab_00386 (SelectExecutor): Selected 1 parts by date, 1 parts by key, 1 marks to read from 1 ranges
2019.07.21 02:11:19.250184 [ 358 ] {3a0c2198-33f6-4403-a373-60895003c6b0} <Trace> test_dftcv9.tab_00386 (SelectExecutor): Reading approx. 8192 rows with 1 streams

2019.07.21 02:11:19.317824 [ 992 ] {} <Error> BaseDaemon: 0. clickhouse-server(StackTrace::StackTrace(ucontext_t const&)+0x35) [0x14fe1535]
1. clickhouse-server() [0x7812866]
2. /lib/x86_64-linux-gnu/libpthread.so.0(+0x12890) [0x7f0f5b7f6890]
3. clickhouse-server(DB::wrapInNullable(COW<DB::IColumn>::immutable_ptr<DB::IColumn> const&, DB::Block const&, std::__1::vector<unsigned long, std::__1::allocator<unsigned long> > const&, unsigned long, unsigned long)+0x191) [0x13673871]
4. clickhouse-server(DB::FunctionCast::prepareRemoveNullable(std::__1::shared_ptr<DB::IDataType const> const&, std::__1::shared_ptr<DB::IDataType const> const&) const::'lambda'(DB::Block&, std::__1::vector<unsigned long, std::__1::allocator<unsigned long> > const&, unsigned long, unsigned long)::operator()(DB::Block&, std::__1::vector<unsigned long, std::__1::allocator<unsigned long> > const&, unsigned long, unsigned long) const+0xe0e) [0x8083e4e]
5. clickhouse-server(DB::PreparedFunctionCast::executeImpl(DB::Block&, std::__1::vector<unsigned long, std::__1::allocator<unsigned long> > const&, unsigned long, unsigned long)+0x343) [0x7fcce53]
6. clickhouse-server(DB::PreparedFunctionImpl::execute(DB::Block&, std::__1::vector<unsigned long, std::__1::allocator<unsigned long> > const&, unsigned long, unsigned long, bool)+0x1eba) [0x1367afda]
7. clickhouse-server(DB::IFunctionBase::execute(DB::Block&, std::__1::vector<unsigned long, std::__1::allocator<unsigned long> > const&, unsigned long, unsigned long, bool)+0x148) [0x784a3c8]
8. clickhouse-server(DB::castColumn(DB::ColumnWithTypeAndName const&, std::__1::shared_ptr<DB::IDataType const> const&, DB::Context const&)+0x1795) [0x126dead5]
9. clickhouse-server() [0x11eead57]
10. clickhouse-server(DB::ConvertingBlockInputStream::ConvertingBlockInputStream(DB::Context const&, std::__1::shared_ptr<DB::IBlockInputStream> const&, DB::Block const&, DB::ConvertingBlockInputStream::MatchColumnsMode)+0x78f) [0x11ee960f]
11. clickhouse-server(DB::StorageMerge::convertingSourceStream(DB::Block const&, DB::Context const&, std::__1::shared_ptr<DB::IAST>&, std::__1::shared_ptr<DB::IBlockInputStream>&, DB::QueryProcessingStage::Enum)+0x2fd) [0x12a65c3d]
12. clickhouse-server(DB::StorageMerge::createSourceStreams(DB::SelectQueryInfo const&, DB::QueryProcessingStage::Enum const&, unsigned long, DB::Block const&, std::__1::shared_ptr<DB::IStorage> const&, DB::TableStructureReadLockHolder const&, std::__1::vector<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> >, std::__1::allocator<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > > >&, DB::Context&, unsigned long, bool, bool)+0x236c) [0x12a6362c]
13. clickhouse-server(DB::StorageMerge::read(std::__1::vector<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> >, std::__1::allocator<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > > > const&, DB::SelectQueryInfo const&, DB::Context const&, DB::QueryProcessingStage::Enum, unsigned long, unsigned int)+0xe1c) [0x12a59a3c]
14. clickhouse-server(void DB::InterpreterSelectQuery::executeFetchColumns<DB::InterpreterSelectQuery::Pipeline>(DB::QueryProcessingStage::Enum, DB::InterpreterSelectQuery::Pipeline&, std::__1::shared_ptr<DB::PrewhereInfo> const&, std::__1::vector<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> >, std::__1::allocator<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > > > const&)+0x98eb) [0x11f7982b]
15. clickhouse-server(void DB::InterpreterSelectQuery::executeImpl<DB::InterpreterSelectQuery::Pipeline>(DB::InterpreterSelectQuery::Pipeline&, std::__1::shared_ptr<DB::IBlockInputStream> const&, bool)+0x194d) [0x11f57fcd]
16. clickhouse-server(DB::InterpreterSelectQuery::executeWithMultipleStreams()+0xef) [0x11f239ef]
17. clickhouse-server(DB::InterpreterSelectWithUnionQuery::executeWithMultipleStreams()+0x1a2) [0x1231f872]
18. clickhouse-server(DB::InterpreterSelectWithUnionQuery::execute()+0xdd) [0x1232035d]
19. clickhouse-server() [0x12704088]
20. clickhouse-server(DB::executeQuery(std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&, DB::Context&, bool, DB::QueryProcessingStage::Enum, bool)+0x29a) [0x127022aa]
21. clickhouse-server(DB::TCPHandler::runImpl()+0x245e) [0x74b8f6e]
22. clickhouse-server(DB::TCPHandler::run()+0x19c) [0x74d765c]
23. clickhouse-server(Poco::Net::TCPServerConnection::start()+0x2f) [0x13f65dcf]
24. clickhouse-server(Poco::Net::TCPServerDispatcher::run()+0x240) [0x13f66a70]
25. clickhouse-server(Poco::PooledThread::run()+0xe5) [0x151cb355]
26. clickhouse-server(Poco::ThreadImpl::runnableEntry(void*)+0xbf) [0x151c59ff]
27. clickhouse-server(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* (*)(void*), Poco::ThreadImpl*> >(void*)+0x8e) [0x151c9a8e]
28. /lib/x86_64-linux-gnu/libpthread.so.0(+0x76db) [0x7f0f5b7eb6db]
29. /lib/x86_64-linux-gnu/libc.so.6(clone+0x3f) [0x7f0f5b17688f]

[alexey-milovidov]

Segfault in stress test (nullptr dereference):

Tech debt, will be addressed asap.