Updated on 2024-05-29 GMT+08:00

Interconnecting with Kafka Through Kerberos Authentication

This section applies only to MRS 3.3.0-LTS or later.

Scenario

The following content describes how to connect ClickHouse to Kafka using Kerberos authentication to consume Kafka data.

Prerequisites

  • A Kafka cluster has been created and is in security mode.

Procedure

  1. Log in to the FusionInsight Manager of the cluster where Kafka is deployed, choose System > Permission > User > Create User, and create a human-machine user with the Kafka permission. For example, create a human-machine user ck_user1. For details about Kafka user permissions, see Managing Kafka User Permissions.
  2. Choose System > Permission > User. On the displayed page, locate the ck_user1 user, and click More > Download Authentication Credential in the Operation column of the user. Save the file and decompress it to obtain the user.keytab and krb5.conf files. Rename the user.keytab file clickhouse_to_kafka.keytab.
  3. Log in to the FusionInsight Manager of the cluster where ClickHouse is deployed, and choose Cluster > Services > ClickHouse. Click Configurations then All Configurations, and click ClickHouseServer(Role) > Engine. The following table shows the parameters need to be configured.

    Parameter

    Description

    kafka.sasl_kerberos_principal

    Principal for connecting to Kafka. Enter the username created in 1.

    kafka.sasl_kerberos_realm

    Domain name of the Kafka cluster

    kafka.sasl_mechanisms

    SASL authentication for connecting to Kafka. The parameter value is GSSAPI.

    kafka_auth_mode

    Authentication mode for the ClickHouse to connect to the Kafka. Set this parameter to Kerberos.

    sasl_kerberos_keytab

    Authentication file for connecting to Kafka, which is the clickhouse_to_kafka.keytab file uploaded in 2.

  4. Click Save. In the displayed dialog box, click OK to save the configuration. Choose Instances, select ClickHouseServer, and click More > Instance Rolling Restart.
  5. Switch to the Kafka client installation directory. For details, see Using the Kafka Client.

    1. Log in to the node where the Kafka client is installed as the Kafka client installation user.
    2. Run the following command to go to the client installation directory:

      cd /opt/client

    3. Run the following command to configure environment variables:

      source bigdata_env

    4. Run the following command to authenticate the current user.

      kinit Component service user

  6. Run the following command to create a Kafka topic. For details, see Managing Kafka Topics.

    kafka-topics.sh --topic topic1 --create --zookeeper IP address of the Zookeeper role instance:Port used by ZooKeeper to listen to client/kafka --partitions 2 --replication-factor 1

    • --topic is the name of the topic to be created, for example, topic1.
    • --zookeeper is the IP address of the node where the ZooKeeper role instances are located, which can be the IP address of any of the three role instances. You can obtain the IP address of the node by performing the following steps:

      Log in to FusionInsight Manager and choose Cluster > Services > ZooKeeper > Instance. View the IP addresses of the ZooKeeper role instances.

    • --partitions and --replication-factor are the topic partitions and topic backup replicas, respectively. The number of the two parameters cannot exceed the number of Kafka role instances.
    • To obtain the Port used by ZooKeeper to listen to client, log in to FusionInsight Manager, click Cluster, choose Services > ZooKeeper, and view the value of clientPort on the Configuration tab page. The default value is 24002.

  7. Log in to the ClickHouse client node and connect it to the ClickHouse server. For details, see Using ClickHouse from Scratch.
  8. Create a Kafka table engine. The following is an example:

    CREATE TABLE queue1 (
    key String,
    value String,
    event_date DateTime
    ) ENGINE = Kafka()
    SETTINGS kafka_broker_list = 'kafka_ip1:21007,kafka_ip2:21007,kafka_ip3:21007',
    kafka_topic_list = 'topic1',
    kafka_group_name = 'group2',
    kafka_format = 'CSV',
    kafka_row_delimiter = '\n',
    kafka_handle_error_mode='stream';

    The required parameters are as follows.

    Parameter

    Description

    kafka_broker_list

    A list of IP addresses and port numbers of Kafka broker instances. For example, :IP address 1 of Kafka broker instance:9092,IP address 2 of Kafka broker instance:9092,IP address 3 of Kafka broker instance:9092

    To obtain the IP address of a Kafka broker instance, perform the following operations:

    Log in to FusionInsight Manager and choose Cluster > Services > Kafka. Click the Instances tab to query the IP addresses of the Kafka instances.

    kafka_topic_list

    Topic where Kafka data is consumed

    kafka_group_name

    Kafka consumer group

    kafka_format

    Formatting type of consumed data. JSONEachRow indicates the JSON format (a piece of data in each line). CSV indicates the data is in a line but separated by commas (,).

    kafka_row_delimiter

    Delimiter character, which ends a message.

    kafka_handle_error_mode

    If this parameter is set to stream, each message processing exception is printed. You need to create a view and query the specific exception of abnormal data through the view.

    The following example shows you how to create a view:

    CREATE MATERIALIZED VIEW default.kafka_errors2
    (
    `topic` String,
    `key` String,
    `partition` Int64,
    `offset` Int64,
    `timestamp` Date,
    `timestamp_ms` Int64,
    `raw` String,
    `error` String
    )
    ENGINE = MergeTree
    ORDER BY (topic, partition, offset)
    SETTINGS index_granularity = 8192 AS
    SELECT
    _topic AS topic,
    _key  AS key,
    _partition AS partition,
    _offset AS offset,
    _timestamp AS timestamp,
    _timestamp_ms AS timestamp_ms,
    _raw_message AS raw,
    _error AS error
    FROM default.queue1;

    Query the view. The following is an example:

    host1 :) select * from kafka_errors2;  SELECT * FROM kafka_errors2  Query id: bf4d788f-bcb9-44f5-95d0-a6c83c591ddb  ┌─topic──┬─key─┬─partition─┬─offset─┬──timestamp─┬─timestamp_ms─┬─raw─┬─error────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐ │ topic1 │     │         1 │      8 │ 2023-06-20 │   1687252213 │ 456 │ Cannot parse date: value is too short: (at row 1) Buffer has gone, cannot extract information about what has been parsed. │ └────────┴─────┴───────────┴────────┴────────────┴──────────────┴─────┴──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘  1 rows in set. Elapsed: 0.003 sec.   host1 :)

    kafka_skip_broken_messages

    (Optional) Number of Kafka data records where parsing exceptions are ignored. If N exceptions occur and the background thread ends, the materialized view is re-arranged to monitor the data.

    kafka_num_consumers

    (Optional) Number of consumers of a single Kafka engine. You can set this parameter to a larger value to improve the consumption data throughput. But the maximum value of this parameter cannot exceed the total number of partitions of the corresponding topic.

    For details about other configurations, see https://clickhouse.com/docs/en/engines/table-engines/integrations/kafka.

  9. Connect the client to ClickHouse to create a local table. The following is an example:

    CREATE TABLE daily1(
    key String,
    value String,
    event_date DateTime
    )ENGINE = MergeTree()
    ORDER BY key;

  10. Connect the client to ClickHouse to create a materialized view. The following is an example:

    CREATE MATERIALIZED VIEW default.consumer1 TO default.daily1 (
    `event_date` DateTime,
    `key` String,
    `value` String
    ) AS
    SELECT
    event_date,
    key,
    value
    FROM default.queue1;

  11. Perform 5 again to go to the Kafka client installation directory.
  12. Run the following command to send a message to the topic created in 6:

    kafka-console-producer.sh --broker-list IP address 1 of the kafka broker instance:9092,IP address 2 of the kafka broker instance:9092,IP address 3 of the kafka broker instance:9092 --topic topic1
    >a1,b1,'2020-08-01 10:00:00'
    >a2,b2,'2020-08-02 10:00:00'
    >a3,b3,'2020-08-02 10:00:00'
    >a4,b4,'2023-09-02 10:00:00'

  13. Query the consumed Kafka data and the preceding materialized view. The following is an example:

    select * from daily;