Интеграция ADQM и Kafka

$ bin/kafka-topics.sh --create --topic topic-kafka-to-adqm --bootstrap-server <kafka_host>:9092

$ bin/kafka-console-producer.sh --topic topic-kafka-to-adqm --bootstrap-server <kafka_host>:9092

>1,"one"
>2,"two"
>3,"three"

CREATE TABLE kafka_table (id Int32, name String) ENGINE = Kafka ('<kafka_host>:9092', 'topic-kafka-to-adqm', 'clickhouse', 'CSV');

CREATE TABLE kafka_data (id Int32, name String) ENGINE = MergeTree() ORDER BY id;

CREATE MATERIALIZED VIEW kafka_data_mv TO kafka_data AS SELECT id, name FROM kafka_table;

SELECT * FROM kafka_data;

$ bin/kafka-console-producer.sh --topic topic-kafka-to-adqm --bootstrap-server <kafka_host>:9092

>4,"four"
>5,"five"

SELECT * FROM kafka_data;

INSERT INTO kafka_table VALUES (6, 'six');

$ bin/kafka-console-consumer.sh --topic topic-kafka-to-adqm --from-beginning --bootstrap-server <kafka_host>:9092

SELECT * FROM kafka_data;

DETACH TABLE kafka_table;

ALTER TABLE kafka_data
    ADD COLUMN topic String,
    ADD COLUMN offset UInt64;

DROP VIEW kafka_data_mv;

ATTACH TABLE kafka_table;

CREATE MATERIALIZED VIEW kafka_data_mv TO kafka_data AS SELECT id, name, _topic as topic, _offset as offset FROM kafka_table;

$ bin/kafka-console-producer.sh --topic topic-kafka-to-adqm --bootstrap-server <kafka_host>:9092

>7,"seven"

SELECT * FROM kafka_data;

$ sudo kadmin.local -q "add_principal -pw PASSWORD ads_user@ADS-KAFKA.LOCAL"

$ kinit -p ads_user@ADS-KAFKA.LOCAL

$ export KAFKA_OPTS="-Djava.security.auth.login.config=/tmp/client.jaas"

$ /usr/lib/kafka/bin/kafka-topics.sh \
  --create --topic topic-adqm-to-kafka-kerberos  \
  --bootstrap-server <kafka_host1>:9092,<kafka_host2>:9092,<kafka_host3>:9092  \
  --command-config /tmp/client.properties

$ /usr/lib/kafka/bin/kafka-console-producer.sh  \
  --topic topic-adqm-to-kafka-kerberos   \
  --bootstrap-server <kafka_host1>:9092,<kafka_host2>:9092,<kafka_host3>:9092  \
  --producer.config /tmp/client.properties

>1,"one"
>2,"two"
>3,"three"

$ sudo kadmin.local -q "add_principal -randkey adqm_user@ADS-KAFKA.LOCAL"

$ sudo kadmin.local -q "xst -kt /tmp/adqm_user.keytab adqm_user@ADS-KAFKA.LOCAL"

$ sudo chown clickhouse:clickhouse /tmp/adqm_user.keytab
$ sudo chmod 0600 /tmp/adqm_user.keytab

<security_protocol>SASL_PLAINTEXT</security_protocol>
<sasl_mechanism>GSSAPI</sasl_mechanism>
<sasl_kerberos_service_name>kafka</sasl_kerberos_service_name>
<sasl_kerberos_keytab>/tmp/adqm_user.keytab</sasl_kerberos_keytab>
<sasl_kerberos_principal>adqm_user@ADS-KAFKA.LOCAL</sasl_kerberos_principal>

$ sudo yum install krb5-libs krb5-workstation

# Configuration snippets may be placed in this directory as well
includedir /etc/krb5.conf.d/

[logging]
 default = FILE:/var/log/krb5libs.log
 kdc = FILE:/var/log/krb5kdc.log
 admin_server = FILE:/var/log/kadmind.log

[libdefaults]
 dns_lookup_realm = false
 ticket_lifetime = 24h
 forwardable = true
 rdns = false
 pkinit_anchors = FILE:/etc/pki/tls/certs/ca-bundle.crt
 default_realm = ADS-KAFKA.LOCAL

[realms]
 ADS-KAFKA.LOCAL = {
 admin_server = <kdc_host>
 kdc = <kdc_host>
 }

[domain_realm]
 .ads-kafka.local = ADS-KAFKA.LOCAL
 ads-kafka.local = ADS-KAFKA.LOCAL

CREATE TABLE adqm_to_ads_kerberos_kafka (id Int32, name String)
ENGINE = Kafka ('<kafka_host1>:9092,<kafka_host2>:9092,<kafka_host3>:9092', 'topic-adqm-to-kafka-kerberos', 'clickhouse_group', 'CSV');

CREATE TABLE adqm_to_ads_kerberos_data (id Int32, name String) ENGINE = MergeTree() ORDER BY id;

CREATE MATERIALIZED VIEW adqm_to_ads_kerberos_data_mv TO adqm_to_ads_kerberos_data
AS SELECT id, name FROM adqm_to_ads_kerberos_kafka;

SELECT * FROM adqm_to_ads_kerberos_data;

INSERT INTO adqm_to_ads_kerberos_kafka VALUES (4, 'four');

$ /usr/lib/kafka/bin/kafka-console-consumer.sh \
  --topic topic-adqm-to-kafka-kerberos \
  --from-beginning  \
  --bootstrap-server <kafka_host1>:9092,<kafka_host2>:9092,<kafka_host3>:9092  \
  --consumer.config /tmp/client.properties