--- title: 'Подключиться к кластеру Kafka' sidebar_label: 'Подключиться к кластеру' description: 'Как подключиться к кластеру Kafka с SSL-сертификатом или без SSL-сертификата в консоли' sidebar_position: 6 --- import Formbricks from '@theme/MDXComponents/Formbricks' import Tabs from '@theme/Tabs' import TabItem from '@theme/TabItem' import {TabItemLabel} from 'docs-kit/components' # Подключиться к кластеру Kafka К кластеру Kafka можно подключиться: * через консольный клиент [kcat](https://github.com/edenhill/kcat). Чтобы избежать возможных ошибок при подключении, рекомендуем использовать библиотеку [librdkafka](https://github.com/confluentinc/librdkafka) не ниже версии 2.6.1; * программный код. Для всех способов доступно подключение [с SSL](#connect-with-ssl) и [без SSL](#connect-without-SSL). При подключении укажите [порт](#connection-ports) и [адрес](#connection-addresses). ## Порты для подключения \{#connection-ports} Для подключения к Kafka используйте порты: * 9092 — порт для подключения без SSL-сертификата; * 9093 — порт для подключения с SSL-сертификатом. ## Адреса для подключения \{#connection-addresses} Вы можете выбрать адрес для подключения в зависимости от одного из сценариев: * [подключение к кластеру в публичной подсети](#connecting-to-cluster-in-public-subnet); * [подключение из приватной подсети к кластеру в приватной подсети](#connecting-from-private-subnet-to-cluster-in-private-subnet). К кластеру в приватной подсети подключиться из интернета нельзя. Вы можете [посмотреть адрес для подключения](#view-connection-address) в панели управления. ![](https://423.selcdn.ru/kb/dbaas-kafka-connect-to-cluster-connecting-LANG-THEME.png) ### Подключение к кластеру в публичной подсети \{#connecting-to-cluster-in-public-subnet} Если кластер в публичной подсети, к нодам можно подключиться по DNS-адресу или IP-адресу из публичной подсети. Мы рекомендуем подключаться по DNS-адресу. Для DNS-адресов в кластере используется механизм master discovery — адрес привязан к роли ноды, а не к самой ноде. При недоступности мастера одна из реплик становится новым мастером и адрес переходит на новую ноду вместе с ролью. При подключении по IP-адресу из публичной подсети механизм master discovery не используется. Если одна из реплик станет новым мастером, IP-адрес мастера изменится и подключение по старому IP-адресу перестанет работать. ### Подключение из приватной подсети к кластеру в приватной подсети \{#connecting-from-private-subnet-to-cluster-in-private-subnet} Если вы подключаетесь из приватной подсети к кластеру в приватной подсети, можно использовать DNS-адрес или приватный IP-адрес. Мы рекомендуем подключаться по DNS-адресу. Для DNS-адресов в кластере используется механизм master discovery — адрес привязан к роли ноды, а не к самой ноде. При недоступности мастера одна из реплик становится новым мастером и адрес переходит на новую ноду вместе с ролью. При подключении по приватному IP-адресу механизм master discovery не используется. Если одна из реплик станет новым мастером, IP-адрес мастера изменится и подключение по старому IP-адресу перестанет работать. Чтобы подключиться из другой приватной подсети, сначала [подключите обе приватные подсети к облачному роутеру](/cloud-servers/cloud-networks/cloud-routers.mdx#connect-private-subnet-to-cloud-router). ### Посмотреть адрес для подключения \{#view-connection-address} 1. В [панели управления](https://my.selectel.ru/vpc/default/dbaas/) в верхнем меню нажмите **Продукты** и выберите **Облачные базы данных**. 2. Откройте вкладку **Активные**. 3. Откройте страницу кластера баз данных → вкладка **Подключение**. 4. В блоке **Адреса для подключения** посмотрите адрес. ## Подключиться с SSL \{#connect-with-ssl} Подключение с использованием TLS(SSL)-шифрования обеспечивает безопасное соединение между вашим сервером и кластером баз данных. Bash :::info Чтобы избежать возможных ошибок при подключении через консольный клиент [kcat](https://github.com/edenhill/kcat), рекомендуем использовать библиотеку [librdkafka](https://github.com/confluentinc/librdkafka) не ниже версии 2.6.1. ::: 1. Скачайте корневой сертификат и поместите его в папку `~/.kafka/`: ```bash mkdir -p ~/.kafka/ wget https://storage.dbaas.selcloud.ru/CA.pem -O ~/.kafka/root.crt chmod 0600 ~/.kafka/root.crt ``` 2. Используйте пример подключения для консьюмера: ```bash kcat -C \ -b : \ -t \ -X sasl.username= \ -X sasl.password= \ -X security.protocol=SASL_SSL \ -X sasl.mechanisms=SCRAM-SHA-512 \ -X ssl.ca.location=$HOME/.kafka/root.crt ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя топика; * `` — имя пользователя с ролью консьюмер, который имеет доступ к топику; * `` — пароль пользователя. 3. Используйте пример подключения для продюсера: ```bash kcat -P \ -b : \ -t \ -X sasl.username= \ -X sasl.password= \ -X security.protocol=SASL_SSL \ -X sasl.mechanisms=SCRAM-SHA-512 \ -X ssl.ca.location=$HOME/.kafka/root.crt ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя топика; * `` — имя пользователя с ролью продюсер, который имеет доступ к топику; * `` — пароль пользователя. Python (confluent-kafka) 1. Скачайте корневой сертификат и поместите его в папку `~/.kafka/`: ```bash mkdir -p ~/.kafka/ wget https://storage.dbaas.selcloud.ru/CA.pem -O ~/.kafka/root.crt chmod 0600 ~/.kafka/root.crt ``` 2. Установите библиотеку confluent-kafka: ```bash pip install confluent-kafka ``` 3. Используйте пример подключения для консьюмера: ```python import confluent_kafka def error_callback(err): raise err consumer = confluent_kafka.Consumer({ "bootstrap.servers": ":", "security.protocol": "SASL_SSL", "sasl.mechanism": "SCRAM-SHA-512", "sasl.username": "", "sasl.password": "", "group.id": "example", "ssl.ca.location": "", "auto.offset.reset": "earliest", "enable.auto.commit": False, "error_cb": error_callback, }) consumer.subscribe([""]) while True: record = consumer.poll(timeout=1.0) if record is not None: if record.error(): if record.error().code() == confluent_kafka.KafkaError._PARTITION_EOF: print(f"{record.topic()} [{record.partition()}]: reached end at offset {record.offset()}") elif record.error(): raise confluent_kafka.KafkaException(record.error()) else: print(record.value()) consumer.close() ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя пользователя с ролью консьюмер, который имеет доступ к топику; * `` — пароль пользователя; * `` — полный путь до корневого сертификата; * `` — имя топика. 4. Используйте пример подключения для продюсера: ```python import confluent_kafka def error_callback(err): raise err producer = confluent_kafka.Producer({ "bootstrap.servers": ":", "security.protocol": "SASL_SSL", "sasl.mechanism": "SCRAM-SHA-512", "sasl.username": "", "sasl.password": "", "ssl.ca.location": "", "error_cb": error_callback, }) producer.produce("", "message") producer.flush(60) ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя пользователя с ролью продюсер, который имеет доступ к топику; * `` — пароль пользователя; * `` — полный путь до корневого сертификата; * `` — имя топика. Python (kafka-python) 1. Скачайте корневой сертификат и поместите его в папку `~/.kafka/`: ```bash mkdir -p ~/.kafka/ wget https://storage.dbaas.selcloud.ru/CA.pem -O ~/.kafka/root.crt chmod 0600 ~/.kafka/root.crt ``` 2. Установите библиотеку kafka-python: ```bash pip install kafka-python ``` 3. Используйте пример подключения для консьюмера: ```python import kafka consumer = kafka.KafkaConsumer( "", bootstrap_servers=":", security_protocol="SASL_SSL", sasl_mechanism="SCRAM-SHA-512", ssl_cafile="", sasl_plain_username="", sasl_plain_password="", ) for record in consumer: print(record) ``` Укажите: * `` — имя топика * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — полный путь до корневого сертификата; * `` — имя пользователя с ролью консьюмер, который имеет доступ к топику; * `` — пароль пользователя. 4. Используйте пример подключения для продюсера: ```python import kafka producer = kafka.KafkaProducer( bootstrap_servers=":", security_protocol="SASL_SSL", sasl_mechanism="SCRAM-SHA-512", ssl_cafile="", sasl_plain_username="", sasl_plain_password="", ) future = producer.send( topic="", value=b"message", ) result = future.get(timeout=60) print(result) ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — полный путь до корневого сертификата; * `` — имя пользователя с ролью продюсер, который имеет доступ к топику; * `` — пароль пользователя; * `` — имя топика. Go 1. Скачайте корневой сертификат и поместите его в папку `~/.kafka/`: ```bash mkdir -p ~/.kafka/ wget https://storage.dbaas.selcloud.ru/CA.pem -O ~/.kafka/root.crt chmod 0600 ~/.kafka/root.crt ``` 2. Создайте файл `scram.go`:
scram.go ```go package main import ( "crypto/sha256" "crypto/sha512" "github.com/xdg-go/scram" ) var ( SHA256 scram.HashGeneratorFcn = sha256.New SHA512 scram.HashGeneratorFcn = sha512.New ) type XDGSCRAMClient struct { *scram.Client *scram.ClientConversation scram.HashGeneratorFcn } func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error) { x.Client, err = x.HashGeneratorFcn.NewClient(userName, password, authzID) if err != nil { return err } x.ClientConversation = x.Client.NewConversation() return nil } func (x *XDGSCRAMClient) Step(challenge string) (response string, err error) { response, err = x.ClientConversation.Step(challenge) return } func (x *XDGSCRAMClient) Done() bool { return x.ClientConversation.Done() } ```
3. Используйте пример подключения для консьюмера: ```go package main import ( "crypto/tls" "crypto/x509" "fmt" "io/ioutil" "os" "os/signal" "strings" "github.com/IBM/sarama" ) const BROKERS = ":" const USER = "" const PASSWORD = "" const TOPIC = "" const CERT = "" func main() { brokers := BROKERS splitBrokers := strings.Split(brokers, ",") conf := sarama.NewConfig() conf.Producer.RequiredAcks = sarama.WaitForAll conf.Version = sarama.V3_3_1_0 conf.Consumer.Return.Errors = true conf.ClientID = "go_client" conf.Metadata.Full = true conf.Net.SASL.Enable = true conf.Net.SASL.User = USER conf.Net.SASL.Password = PASSWORD conf.Net.SASL.Handshake = true conf.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA512} } conf.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA512) certs := x509.NewCertPool() pemPath := CERT pemData, err := ioutil.ReadFile(pemPath) if err != nil { fmt.Println("Couldn't load cert: ", err.Error()) panic(err) } certs.AppendCertsFromPEM(pemData) conf.Net.TLS.Enable = true conf.Net.TLS.Config = &tls.Config{ RootCAs: certs, } master, err := sarama.NewConsumer(splitBrokers, conf) if err != nil { fmt.Println("Couldn't create consumer: ", err.Error()) panic(err) } defer func() { if err := master.Close(); err != nil { panic(err) } }() consumer, err := master.ConsumePartition(TOPIC, 0, sarama.OffsetOldest) if err != nil { panic(err) } signals := make(chan os.Signal, 1) signal.Notify(signals, os.Interrupt) doneCh := make(chan struct{}) go func() { for { select { case err := <-consumer.Errors(): fmt.Println(err) case msg := <-consumer.Messages(): fmt.Println("Received messages", string(msg.Key), string(msg.Value)) case <-signals: doneCh <- struct{}{} } } }() <-doneCh } ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя пользователя с ролью консьюмер, который имеет доступ к топику; * `` — пароль пользователя; * `` — имя топика; * `` — полный путь до корневого сертификата. 4. Используйте пример подключения для продюсера: ```go package main import ( "crypto/tls" "crypto/x509" "fmt" "io/ioutil" "strings" "github.com/IBM/sarama" ) const BROKERS = ":" const USER = "" const PASSWORD = "" const TOPIC = "" const CERT = "" func main() { brokers := BROKERS splitBrokers := strings.Split(brokers, ",") conf := sarama.NewConfig() conf.Producer.RequiredAcks = sarama.WaitForAll conf.Producer.Return.Successes = true conf.Version = sarama.V3_3_1_0 conf.ClientID = "go_client" conf.Net.SASL.Enable = true conf.Net.SASL.Handshake = true conf.Net.SASL.User = USER conf.Net.SASL.Password = PASSWORD conf.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA512} } conf.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA512) certs := x509.NewCertPool() pemPath := CERT pemData, err := ioutil.ReadFile(pemPath) if err != nil { fmt.Println("Couldn't load cert: ", err.Error()) panic(err) } certs.AppendCertsFromPEM(pemData) conf.Net.TLS.Enable = true conf.Net.TLS.Config = &tls.Config{ RootCAs: certs, } syncProducer, err := sarama.NewSyncProducer(splitBrokers, conf) if err != nil { fmt.Println("Couldn't create producer: ", err.Error()) panic(err) } msg := &sarama.ProducerMessage{ Topic: TOPIC, Value: sarama.StringEncoder("message"), } _, _, err = syncProducer.SendMessage(msg) if err != nil { fmt.Println("Couldn't send message: ", err.Error()) panic(err) } } ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя пользователя с ролью консьюмер, который имеет доступ к топику; * `` — пароль пользователя; * `` — имя топика; * `` — полный путь до корневого сертификата. Node.js 1. Скачайте корневой сертификат и поместите его в папку `~/.kafka/`: ```bash mkdir -p ~/.kafka/ wget https://storage.dbaas.selcloud.ru/CA.pem -O ~/.kafka/root.crt chmod 0600 ~/.kafka/root.crt ``` 2. Установите библиотеку kafkajs: ```bash npm install kafkajs ``` 3. Используйте пример подключения для консьюмера: ```js const fs = require('fs'); const { Kafka } = require('kafkajs'); const config = { brokers: [':'], ssl: { ca: [fs.readFileSync('', 'utf-8')], }, sasl: { mechanism: 'scram-sha-512', username: '', password: '', }, }; const kafka = new Kafka(config); const consumer = kafka.consumer({ groupId: 'example' }); const run = async () => { await consumer.connect(); await consumer.subscribe({ topic: '', fromBeginning: true, }); await consumer.run({ eachMessage: async ({ message }) => { console.log({ value: message.value.toString() }); }, }); }; run(); ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя пользователя с ролью консьюмер, который имеет доступ к топику; * `` — пароль пользователя; * `` — полный путь до корневого сертификата; * `` — имя топика. 4. Используйте пример подключения для продюсера: ```js const fs = require('fs'); const { Kafka } = require('kafkajs'); const config = { brokers: [':'], ssl: { ca: [fs.readFileSync('', 'utf-8')], }, sasl: { mechanism: 'scram-sha-512', username: '', password: '', }, }; const kafka = new Kafka(config); const producer = kafka.producer(); const run = async () => { await producer.connect(); await producer.send({ topic: '', messages: [{ value: 'message' }], }); await producer.disconnect(); }; run(); ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — полный путь до корневого сертификата; * `` — имя пользователя с ролью продюсер, который имеет доступ к топику; * `` — пароль пользователя; * `` — имя топика. Java 1. Скачайте корневой сертификат и поместите его в папку `~/.kafka/`: ```bash mkdir -p ~/.kafka/ wget https://storage.dbaas.selcloud.ru/CA.pem -O ~/.kafka/root.crt chmod 0600 ~/.kafka/root.crt ``` 2. Перейдите в каталог, где будет располагаться хранилище сертификатов Java: ```bash cd /etc/security ``` 3. Добавьте SSL-сертификат в хранилище доверенных сертификатов Java (Java Key Store): ```java keytool \ -importcert \ -alias RootCA \ -file ~/.kafka/root.crt \ -keystore ssl \ -storepass \ -noprompt ``` Укажите `` — пароль хранилища для дополнительной защиты. 4. Создайте конфигурационный файл для Maven:
pom.xml ```java 4.0.0 com.example app jar 1.0 1.8 1.8 org.slf4j slf4j-simple 2.0.7 com.fasterxml.jackson.core jackson-databind 2.14.2 org.apache.kafka kafka-clients 3.4.0 ${project.artifactId}-${project.version} src src org.apache.maven.plugins maven-assembly-plugin attached package jar-with-dependencies com.example.App org.apache.maven.plugins maven-jar-plugin 3.1.0 com.example.App ```
5. Используйте пример подключения для консьюмера: ```bash src/com/example/App.java ``` ```bash package com.example; import java.util.*; import org.apache.kafka.common.*; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.clients.consumer.*; public class App { private static final String HOST = ":"; private static final String TOPIC = ""; private static final String USER = ""; private static final String PASSWORD = ""; private static final String TS_FILE = "/etc/security/ssl"; private static final String TS_PASSWORD = ""; public static void main(String[] args) { String jaasCfg = String.format("org.apache.kafka.common.security.scram.ScramLoginModule required username=\"%s\" password=\"%s\";", USER, PASSWORD); String deserializer = StringDeserializer.class.getName(); Properties props = new Properties(); props.put("bootstrap.servers", HOST); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); props.put("group.id", "example"); props.put("key.deserializer", deserializer); props.put("value.deserializer", deserializer); props.put("security.protocol", "SASL_SSL"); props.put("sasl.mechanism", "SCRAM-SHA-512"); props.put("sasl.jaas.config", jaasCfg); props.put("ssl.truststore.location", TS_FILE); props.put("ssl.truststore.password", TS_PASSWORD); Consumerconsumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList(new String[] {TOPIC})); while (true) { ConsumerRecords records = consumer.poll(100); for (ConsumerRecord record: records) { System.out.println(record.value()); } } } } ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя топика; * `` — имя пользователя с ролью консьюмер, который имеет доступ к топику; * `` — пароль пользователя; * `` — пароль хранилища. 6. Используйте пример подключения для продюсера: ```bash src/com/example/App.java ``` ```bash package com.example; import java.util.*; import org.apache.kafka.common.*; import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.clients.producer.*; public class App { private static final String HOST = ":"; private static final String TOPIC = ""; private static final String USER = ""; private static final String PASSWORD = ""; private static final String TS_FILE = "/etc/security/ssl"; private static final String TS_PASSWORD = ""; public static void main(String[] args) { String jaasCfg = String.format("org.apache.kafka.common.security.scram.ScramLoginModule required username=\"%s\" password=\"%s\";", USER, PASSWORD); String serializer = StringSerializer.class.getName(); Properties props = new Properties(); props.put("bootstrap.servers", HOST); props.put("acks", "all"); props.put("key.serializer", serializer); props.put("value.serializer", serializer); props.put("security.protocol", "SASL_SSL"); props.put("sasl.mechanism", "SCRAM-SHA-512"); props.put("sasl.jaas.config", jaasCfg); props.put("ssl.truststore.location", TS_FILE); props.put("ssl.truststore.password", TS_PASSWORD); Producer producer = new KafkaProducer<>(props); try { producer.send(new ProducerRecord(TOPIC, "key", "message")).get(); producer.flush(); producer.close(); } catch (Exception ex) { System.out.println(ex); producer.close(); } } } ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя топика; * `` — имя пользователя с ролью продюсер, который имеет доступ к топику; * `` — пароль пользователя; * `` — пароль сертификата. 7. Соберите и запустите приложение: ```java mvn clean package java -jar target/app-1.0-jar-with-dependencies.jar ``` ## Подключиться без SSL \{#connect-without-SSL} Bash :::info Чтобы избежать возможных ошибок при подключении через консольный клиент [kcat](https://github.com/edenhill/kcat), рекомендуем использовать библиотеку [librdkafka](https://github.com/confluentinc/librdkafka) не ниже версии 2.6.1. ::: 1. Откройте CLI. 2. Используйте пример подключения для консьюмера: ```bash kcat -C \ -b : \ -t \ -X sasl.username= \ -X sasl.password= \ -X security.protocol=SASL_PLAINTEXT \ -X sasl.mechanisms=SCRAM-SHA-512 ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя топика; * `` — имя пользователя с ролью консьюмер, который имеет доступ к топику; * `` — пароль пользователя. 3. Используйте пример подключения для продюсера: ```bash kcat -P \ -b : \ -t \ -X sasl.username= \ -X sasl.password= \ -X security.protocol=SASL_PLAINTEXT \ -X sasl.mechanisms=SCRAM-SHA-512 ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя топика; * `` — имя пользователя с ролью продюсер, который имеет доступ к топику; * `` — пароль пользователя. Python (confluent-kafka) 1. Установите библиотеку confluent-kafka: ```python pip install confluent-kafka ``` 2. Используйте пример подключения для консьюмера: ```python import confluent_kafka def error_callback(err): raise err consumer = confluent_kafka.Consumer({ "bootstrap.servers": ":", "security.protocol": "SASL_PLAINTEXT", "sasl.mechanism": "SCRAM-SHA-512", "sasl.username": "", "sasl.password": "", "group.id": "example", "auto.offset.reset": "earliest", "enable.auto.commit": False, "error_cb": error_callback, }) consumer.subscribe([""]) while True: record = consumer.poll(timeout=1.0) if record is not None: if record.error(): if record.error().code() == confluent_kafka.KafkaError._PARTITION_EOF: print(f"{record.topic()} [{record.partition()}]: reached end at offset {record.offset()}") elif record.error(): raise confluent_kafka.KafkaException(record.error()) else: print(record.value()) consumer.close() ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя пользователя с ролью консьюмер, который имеет доступ к топику; * `` — пароль пользователя; * `` — имя топика. 3. Используйте пример подключения для продюсера: ```python import confluent_kafka def error_callback(err): raise err producer = confluent_kafka.Producer({ "bootstrap.servers": ":", "security.protocol": "SASL_PLAINTEXT", "sasl.mechanism": "SCRAM-SHA-512", "sasl.username": "", "sasl.password": "", "error_cb": error_callback, }) producer.produce("", "message") producer.flush(60) ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя пользователя базы данных; * `` — пароль пользователя; * `` — имя топика. Python (kafka-python) 1. Установите библиотеку kafka-python: ```python pip install kafka-python ``` 2. Используйте пример подключения для консьюмера: ```python import kafka consumer = kafka.KafkaConsumer( "", bootstrap_servers=":", security_protocol="SASL_PLAINTEXT", sasl_mechanism="SCRAM-SHA-512", sasl_plain_username="", sasl_plain_password="", ) for record in consumer: print(record) ``` Укажите: * `` — имя топика * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя пользователя с ролью консьюмер, который имеет доступ к топику; * `` — пароль пользователя. 3. Используйте пример подключения для продюсера: ```python import kafka producer = kafka.KafkaProducer( bootstrap_servers=":", security_protocol="SASL_PLAINTEXT", sasl_mechanism="SCRAM-SHA-512", sasl_plain_username="", sasl_plain_password="", ) future = producer.send( topic="", value=b"message", ) result = future.get(timeout=60) print(result) ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя пользователя базы данных; * `` — пароль пользователя; * `` — имя топика. Go 1. Создайте файл `scram.go`:
scram.go ```go package main import ( "crypto/sha256" "crypto/sha512" "github.com/xdg-go/scram" ) var ( SHA256 scram.HashGeneratorFcn = sha256.New SHA512 scram.HashGeneratorFcn = sha512.New ) type XDGSCRAMClient struct { *scram.Client *scram.ClientConversation scram.HashGeneratorFcn } func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error) { x.Client, err = x.HashGeneratorFcn.NewClient(userName, password, authzID) if err != nil { return err } x.ClientConversation = x.Client.NewConversation() return nil } func (x *XDGSCRAMClient) Step(challenge string) (response string, err error) { response, err = x.ClientConversation.Step(challenge) return } func (x *XDGSCRAMClient) Done() bool { return x.ClientConversation.Done() } ```
2. Используйте пример подключения для консьюмера: ```go package main import ( "fmt" "os" "os/signal" "strings" "github.com/IBM/sarama" ) const BROKERS = ":" const USER = "" const PASSWORD = "" const TOPIC = "" func main() { brokers := BROKERS splitBrokers := strings.Split(brokers, ",") conf := sarama.NewConfig() conf.Producer.RequiredAcks = sarama.WaitForAll conf.Version = sarama.V3_3_1_0 conf.Consumer.Return.Errors = true conf.ClientID = "go_client" conf.Metadata.Full = true conf.Net.SASL.Enable = true conf.Net.SASL.User = USER conf.Net.SASL.Password = PASSWORD conf.Net.SASL.Handshake = true conf.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA512} } conf.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA512) conf.Net.TLS.Enable = false master, err := sarama.NewConsumer(splitBrokers, conf) if err != nil { fmt.Println("Couldn't create consumer: ", err.Error()) panic(err) } defer func() { if err := master.Close(); err != nil { panic(err) } }() consumer, err := master.ConsumePartition(TOPIC, 0, sarama.OffsetOldest) if err != nil { panic(err) } signals := make(chan os.Signal, 1) signal.Notify(signals, os.Interrupt) doneCh := make(chan struct{}) go func() { for { select { case err := <-consumer.Errors(): fmt.Println(err) case msg := <-consumer.Messages(): fmt.Println("Received messages", string(msg.Key), string(msg.Value)) case <-signals: doneCh <- struct{}{} } } }() <-doneCh } ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя пользователя с ролью консьюмер, который имеет доступ к топику; * `` — пароль пользователя; * `` — имя топика. 3. Используйте пример подключения для продюсера: ```go package main import ( "fmt" "strings" "github.com/IBM/sarama" ) const BROKERS = ":" const USER = "" const PASSWORD = "" const TOPIC = "" func main() { brokers := BROKERS splitBrokers := strings.Split(brokers, ",") conf := sarama.NewConfig() conf.Producer.RequiredAcks = sarama.WaitForAll conf.Producer.Return.Successes = true conf.Version = sarama.V3_3_1_0 conf.ClientID = "go_client" conf.Net.SASL.Enable = true conf.Net.SASL.Handshake = true conf.Net.SASL.User = USER conf.Net.SASL.Password = PASSWORD conf.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA512} } conf.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA512) conf.Net.TLS.Enable = false syncProducer, err := sarama.NewSyncProducer(splitBrokers, conf) if err != nil { fmt.Println("Couldn't create producer: ", err.Error()) panic(err) } msg := &sarama.ProducerMessage{ Topic: TOPIC, Value: sarama.StringEncoder("message"), } _, _, err = syncProducer.SendMessage(msg) if err != nil { fmt.Println("Couldn't send message: ", err.Error()) panic(err) } } ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя пользователя с ролью консьюмер, который имеет доступ к топику; * `` — пароль пользователя; * `` — имя топика. Node.js 1. Установите библиотеку kafkajs: ```bash npm install kafkajs ``` 2. Используйте пример подключения для консьюмера: ```js const { Kafka } = require('kafkajs'); const config = { brokers: [':'], ssl: false, sasl: { mechanism: 'scram-sha-512', username: '', password: '', }, }; const kafka = new Kafka(config); const consumer = kafka.consumer({ groupId: 'example' }); const run = async () => { await consumer.connect(); await consumer.subscribe({ topic: '', fromBeginning: true, }); await consumer.run({ eachMessage: async ({ message }) => { console.log({ value: message.value.toString() }); }, }); }; run(); ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя пользователя с ролью консьюмер, который имеет доступ к топику; * `` — пароль пользователя; * `` — имя топика. 3. Используйте пример подключения для продюсера: ```js const { Kafka } = require('kafkajs'); const config = { brokers: [':'], ssl: false, sasl: { mechanism: 'scram-sha-512', username: '', password: '', }, }; const kafka = new Kafka(config); const producer = kafka.producer(); const run = async () => { await producer.connect(); await producer.send({ topic: '', messages: [{ value: 'message' }], }); await producer.disconnect(); }; run(); ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя пользователя базы данных; * `` — пароль пользователя; * `` — имя топика. Java 1. Создайте конфигурационный файл для Maven:
pom.xml ```java 4.0.0 com.example app jar 1.0 1.8 1.8 org.slf4j slf4j-simple 2.0.7 com.fasterxml.jackson.core jackson-databind 2.14.2 org.apache.kafka kafka-clients 3.4.0 ${project.artifactId}-${project.version} src src org.apache.maven.plugins maven-assembly-plugin attached package jar-with-dependencies com.example.App org.apache.maven.plugins maven-jar-plugin 3.1.0 com.example.App ```
2. Используйте пример подключения для консьюмера: ```bash src/com/example/App.java ``` ```java package com.example; import java.util.*; import org.apache.kafka.common.*; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.clients.consumer.*; public class App { private static final String HOST = ":"; private static final String TOPIC = ""; private static final String USER = ""; private static final String PASSWORD = ""; public static void main(String[] args) { String jaasCfg = String.format("org.apache.kafka.common.security.scram.ScramLoginModule required username=\"%s\" password=\"%s\";", USER, PASSWORD); String deserializer = StringDeserializer.class.getName(); Properties props = new Properties(); props.put("bootstrap.servers", HOST); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); props.put("group.id", "example"); props.put("key.deserializer", deserializer); props.put("value.deserializer", deserializer); props.put("security.protocol", "SASL_PLAINTEXT"); props.put("sasl.mechanism", "SCRAM-SHA-512"); props.put("sasl.jaas.config", jaasCfg); Consumerconsumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList(new String[] {TOPIC})); while (true) { ConsumerRecords records = consumer.poll(100); for (ConsumerRecord record: records) { System.out.println(record.value()); } } } } ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя топика; * `` — имя пользователя с ролью консьюмер, который имеет доступ к топику; * `` — пароль пользователя. 3. Используйте пример подключения для продюсера: ```bash src/com/example/App.java ``` ```java package com.example; import java.util.*; import org.apache.kafka.common.*; import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.clients.producer.*; public class App { private static final String HOST = ":"; private static final String TOPIC = ""; private static final String USER = ""; private static final String PASSWORD = ""; public static void main(String[] args) { String jaasCfg = String.format("org.apache.kafka.common.security.scram.ScramLoginModule required username=\"%s\" password=\"%s\";", USER, PASSWORD); String serializer = StringSerializer.class.getName(); Properties props = new Properties(); props.put("bootstrap.servers", HOST); props.put("acks", "all"); props.put("key.serializer", serializer); props.put("value.serializer", serializer); props.put("security.protocol", "SASL_PLAINTEXT"); props.put("sasl.mechanism", "SCRAM-SHA-512"); props.put("sasl.jaas.config", jaasCfg); Producer producer = new KafkaProducer<>(props); try { producer.send(new ProducerRecord(TOPIC, "key", "message")).get(); producer.flush(); producer.close(); } catch (Exception ex) { System.out.println(ex); producer.close(); } } } ``` Укажите: * `` — DNS-адрес ноды; * `` — [порт для подключения](#connection-ports); * `` — имя топика; * `` — имя пользователя базы данных; * `` — пароль пользователя. 4. Соберите и запустите приложение: ```java mvn clean package java -jar target/app-1.0-jar-with-dependencies.jar ```