thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-09-21 09:10:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

Drop claimed from the KafkaBlockInputStream

Browse Source
This commit is contained in:
Azat Khuzhin 2020-05-03 16:04:04 +03:00
parent a10cfc3ae5
commit 61c5df5662
2 changed files with 6 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/Storages/Kafka/KafkaBlockInputStream.cpp
View File

@ -33,7 +33,7 @@ KafkaBlockInputStream::KafkaBlockInputStream(
KafkaBlockInputStream::~KafkaBlockInputStream()
{
if (!claimed)
if (!buffer)
return;
if (broken)
@ -51,7 +51,6 @@ void KafkaBlockInputStream::readPrefixImpl()
{
auto timeout = std::chrono::milliseconds(context.getSettingsRef().kafka_max_wait_ms.totalMilliseconds());
buffer = storage.popReadBuffer(timeout);
claimed = !!buffer;
if (!buffer)
return;

7
src/Storages/Kafka/KafkaBlockInputStream.h
View File

@ -33,9 +33,12 @@ private:
UInt64 max_block_size;
ConsumerBufferPtr buffer;
bool broken = true, finished = false, claimed = false, commit_in_suffix;
bool broken = true;
bool finished = false;
bool commit_in_suffix;
const Block non_virtual_header, virtual_header;
const Block non_virtual_header;
const Block virtual_header;
};
}