如何正确配置ClickHouse以无报错连接不同认证方式的Kafka集群？

问题根源 #

你在config.xml中配置的全局<kafka>参数会被所有Kafka引擎表默认继承，导致连接无认证的Kafka-B时，ClickHouse依然使用SASL认证逻辑，和Kafka-B的无认证配置冲突，触发握手失败报错。

两种可行解决方案 #

方案1：使用Kafka配置Profile（推荐） #

在config.xml中为每个Kafka集群单独配置带<profile>标识的<kafka>块，实现集群级配置隔离：

<!-- Kafka-A的专属配置Profile -->
<kafka>
    <profile>kafka_a</profile>
    <security_protocol>sasl_plaintext</security_protocol>
    <sasl_mechanism>SCRAM-SHA-256</sasl_mechanism>
    <sasl_username>xxx</sasl_username>
    <sasl_password>xxx</sasl_password>
    <debug>all</debug>
    <auto_offset_reset>latest</auto_offset_reset>
    <compression_type>snappy</compression_type> 
</kafka>

<!-- Kafka-B的专属配置Profile（无认证） -->
<kafka>
    <profile>kafka_b</profile>
    <security_protocol>plaintext</security_protocol>
    <auto_offset_reset>latest</auto_offset_reset>
    <compression_type>snappy</compression_type> 
</kafka>

创建Kafka引擎表时，通过kafka_profile参数指定对应集群的配置：

• 连接Kafka-A的表：

CREATE TABLE kafka_a_table (
    -- 定义表结构
    id Int64,
    content String
) ENGINE = Kafka
SETTINGS 
    kafka_broker_list = 'kafka_a_brokers:9092',
    kafka_topic_list = 'your_topic_a',
    kafka_group_name = 'clickhouse_group_a',
    kafka_format = 'JSONEachRow',
    kafka_profile = 'kafka_a';

• 连接Kafka-B的表：

CREATE TABLE kafka_b_table (
    -- 定义表结构
    id Int64,
    content String
) ENGINE = Kafka
SETTINGS 
    kafka_broker_list = 'kafka_b_brokers:9092',
    kafka_topic_list = 'your_topic_b',
    kafka_group_name = 'clickhouse_group_b',
    kafka_format = 'JSONEachRow',
    kafka_profile = 'kafka_b';

方案2：表级直接指定参数（无需全局配置） #

如果不想维护全局Profile，也可以在创建每个Kafka表时，直接在SETTINGS中写入对应集群的所有参数，覆盖默认配置：

• 连接Kafka-A的表：

CREATE TABLE kafka_a_table (
    id Int64,
    content String
) ENGINE = Kafka
SETTINGS 
    kafka_broker_list = 'kafka_a_brokers:9092',
    kafka_topic_list = 'your_topic_a',
    kafka_group_name = 'clickhouse_group_a',
    kafka_format = 'JSONEachRow',
    kafka_security_protocol = 'sasl_plaintext',
    kafka_sasl_mechanism = 'SCRAM-SHA-256',
    kafka_sasl_username = 'xxx',
    kafka_sasl_password = 'xxx',
    kafka_auto_offset_reset = 'latest',
    kafka_compression_type = 'snappy';

• 连接Kafka-B的表：

CREATE TABLE kafka_b_table (
    id Int64,
    content String
) ENGINE = Kafka
SETTINGS 
    kafka_broker_list = 'kafka_b_brokers:9092',
    kafka_topic_list = 'your_topic_b',
    kafka_group_name = 'clickhouse_group_b',
    kafka_format = 'JSONEachRow',
    kafka_security_protocol = 'plaintext',
    kafka_auto_offset_reset = 'latest',
    kafka_compression_type = 'snappy';

内容的提问来源于stack exchange，提问作者姜冠宇