Handle errors for Kafka engine by fastio · Pull Request #21850 · ClickHouse/ClickHouse

fastio · 2021-03-18T05:44:49Z

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

Changelog category (leave one):

Improvement

Changelog entry (a user-readable short description of the changes that goes to CHANGELOG.md):

Allow publishing Kafka errors to a virtual column of Kafka engine, controlled by the kafka_handle_error_mode setting.

Detailed description / Documentation draft:

When the parameter kafka_handle_error_mode is set to stream, the errors will be pushed in to virtual column of Kafka engine.
If the kafka_handle_error_mode is set to default, silently ignore errors with some threshold, similar to what we have now.

Use case:

 CREATE TABLE default.tt
(
    `i` Int64,
    `s` String
)
ENGINE = Kafka
SETTINGS kafka_broker_list = '172.19.0.32:9092', kafka_topic_list = 't2', kafka_group_name = 'g1', kafka_format = 'JSONEachRow', kafka_handle_error_mode='stream';

The table data is used to save data from kafka.

CREATE MATERIALIZED VIEW default.data
(
    `i` Int64,
    `s` String
)
ENGINE = MergeTree
ORDER BY i
SETTINGS index_granularity = 8192 AS
SELECT
    i,
    s
FROM default.tt
WHERE length(_error) = 0

The table error is used to save the exception row information.

CREATE MATERIALIZED VIEW default.kafka_errors
(
    `topic` String,
    `partition` Int64,
    `offset` Int64,
    `raw` String,
    `error` String
)
ENGINE = MergeTree
ORDER BY topic
SETTINGS index_granularity = 8192 AS
SELECT
    _topic AS topic,
    _partition AS partition,
    _offset AS offset,
    _raw_message AS raw,
    _error AS error
FROM default.tt
WHERE length(_error) > 0

src/Storages/Kafka/StorageKafka.h

azat

@fastio looks useful, can you add a test (tests/integration/test_sotrage_kafka)?

fastio · 2021-03-18T07:29:28Z

@fastio looks useful, can you add a test (tests/integration/test_sotrage_kafka)?

Sure.

src/Core/Settings.h

src/Processors/Formats/IRowInputFormat.h

src/Formats/FormatFactory.h

src/Storages/Kafka/KafkaBlockInputStream.cpp

src/Common/ErrorCodes.cpp

src/Storages/Kafka/KafkaBlockInputStream.cpp

tests/integration/test_storage_kafka/test.py

tests/queries/0_stateless/01781_general_format_for_parser.sh

filimonov

Lgtm

akuzm · 2021-04-09T19:58:39Z

fuzzer #22943
functional stateless tests #22944

filimonov · 2021-04-11T06:04:09Z

@fastio please check https://clickhouse-test-reports.s3.yandex.net/21639/03a2ba315b6b75c9264a89f712adeb61db19ce3d/integration_tests_(release).html#fail1

filimonov · 2021-05-15T15:21:21Z

@fastio please check https://clickhouse-test-reports.s3.yandex.net/21639/03a2ba315b6b75c9264a89f712adeb61db19ce3d/integration_tests_(release).html#fail1

Fixed here: dd619e1

alexcd90 · 2021-08-19T10:20:03Z

i want to know could it to adapted to the kafka procuder ? in other verseion ck, i found the producer occur loss data issue.

alexey-milovidov · 2022-04-07T22:21:12Z

@akuzm it is strange to have it for Kafka, but not for RabbitMQ.

akuzm · 2022-04-08T05:59:08Z

So what, is it the only strange thing in ClickHouse? As I said, stop nagging me, I'm not your employee for a long time already.

alexey-milovidov · 2022-04-08T13:30:05Z

@akuzm We need to prevent inconsistencies in our system.

robot-clickhouse added the pr-improvement Pull request with some product improvements label Mar 18, 2021

fastio mentioned this pull request Mar 18, 2021

Add _error and _raw_message virtual columns for Kafka engine #20249

Closed

sundy-li reviewed Mar 18, 2021

View reviewed changes

src/Storages/Kafka/StorageKafka.h Show resolved Hide resolved

azat reviewed Mar 18, 2021

View reviewed changes