Connect to the Kafka cluster

The Kafka cluster can be connected to by DNS address and IP address.

We recommend connecting by DNS address because DNS addresses correspond to node roles and lead to the actual IP addresses of the master and replicas. IP addresses correspond to specific nodes. If the master is unavailable, one of the replicas will assume its role, the master's IP address will change, and the IP connection will stop working.

If the cluster is connected to a private subnet and you want to work with it via DNS, connect the cluster subnet to a cloud router with access to the external network. Use the instructions Set up Internet access through a cloud router.

A public IP address cannot be used.

Connection ports

Use ports to connect to Kafka:

9092 — port for connection without SSL certificate;
9093 — port for connection with SSL certificate.

Ways of connection

through the kcat terminal client;
from program code with and without SSL.

View the address for connection

In the Dashboard, on the top menu, click Products and select Cloud Databases.
Open the Active tab.
Open the Database Cluster page → Connection tab.
In the Addresses to connect block, look up the address.

Connect with SSL

Connecting using TLS/SSL encryption provides a secure connection between your server and the database cluster.

Download the root certificate and place it in the ~/.kafka/ folder:

mkdir -p ~/.kafka/
wget https://storage.dbaas.selcloud.ru/CA.pem -O ~/.kafka/root.crt
chmod 0600 ~/.kafka/root.crt

Use the connection example for the concumer:
```
kcat -C \
-b <host>:<port> \
-t <topic_name> \
-X sasl.username=<user_name> \
-X sasl.password=<password> \
-X security.protocol=SASL_SSL \
-X sasl.mechanisms=SCRAM-SHA-512 \
-X ssl.ca.location=$HOME/.kafka/root.crt
```
Specify:
- <host> — DNS address of the node;
- <port> — port for connection;
- <topic_name> — topic name;
- <user_name> — the name of the user with the role of concumer who has access to the topic;
- <password> — user password.
Use the connection example for the producer:
```
kcat -P \
-b <host>:<port> \
-t <topic_name> \
-X sasl.username=<user_name> \
-X sasl.password=<password> \
-X security.protocol=SASL_SSL \
-X sasl.mechanisms=SCRAM-SHA-512 \
-X ssl.ca.location=$HOME/.kafka/root.crt
```
Specify:
- <host> — DNS address of the node;
- <port> — port for connection;
- <topic_name> — topic name;
- <user_name> — name of the user with the producer role who has access to the topic;
- <password> — user password.

Download the root certificate and place it in the ~/.kafka/ folder:

mkdir -p ~/.kafka/
wget https://storage.dbaas.selcloud.ru/CA.pem -O ~/.kafka/root.crt
chmod 0600 ~/.kafka/root.crt

Install the confluent-kafka library:
```
pip install confluent-kafka
```

Use the connection example for the concumer:

import confluent_kafka

def error_callback(err):
    raise err

consumer = confluent_kafka.Consumer({
    "bootstrap.servers": "<host>:<port>",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "SCRAM-SHA-512",
    "sasl.username": "<user_name>",
    "sasl.password": "<password>",
    "group.id": "example",
    "ssl.ca.location": "<path>",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
    "error_cb": error_callback,
})

consumer.subscribe(["<topic_name>"])

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()

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<user_name> — the name of the user with the role of concumer who has access to the topic;
<password> — user password;
<path> — the full path to the root certificate;
<topic_name> — topic name.

Use the connection example for the producer:

import confluent_kafka

def error_callback(err):
    raise err

producer = confluent_kafka.Producer({
    "bootstrap.servers": "<host>:<port>",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "SCRAM-SHA-512",
    "sasl.username": "<user_name>",
    "sasl.password": "<password>",
    "ssl.ca.location": "<path>",
    "error_cb": error_callback,
})
producer.produce("<topic_name>", "message")
producer.flush(60)

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<user_name> — the name of the user with the producer role who has access to the topic;
<password> — user password;
<path> — the full path to the root certificate;
<topic_name> — topic name.

Download the root certificate and place it in the ~/.kafka/ folder:

mkdir -p ~/.kafka/
wget https://storage.dbaas.selcloud.ru/CA.pem -O ~/.kafka/root.crt
chmod 0600 ~/.kafka/root.crt

Install the kafka-python library:
```
pip install kafka-python
```

Use the connection example for the concumer:

import kafka

consumer = kafka.KafkaConsumer(
    "<topic_name>",
    bootstrap_servers="<host>:<port>",
    security_protocol="SASL_SSL",
    sasl_mechanism="SCRAM-SHA-512",
    ssl_cafile="<path>",
    sasl_plain_username="<user_name>",
    sasl_plain_password="<password>",
)
for record in consumer:
    print(record)

Specify:

<topic_name> — topic name
<host> — DNS address of the node;
<port> — port for connection;
<path> — the full path to the root certificate;
<user_name> — the name of the user with the role of concumer who has access to the topic;
<password> — user password.

Use the connection example for the producer:

import kafka

producer = kafka.KafkaProducer(
    bootstrap_servers="<host>:<port>",
    security_protocol="SASL_SSL",
    sasl_mechanism="SCRAM-SHA-512",
    ssl_cafile="<path>",
    sasl_plain_username="<user_name>",
    sasl_plain_password="<password>",
)
future = producer.send(
    topic="<topic_name>",
    value=b"message",
)
result = future.get(timeout=60)
print(result)

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<path> — the full path to the root certificate;
<user_name> — the name of the user with the producer role who has access to the topic;
<password> — user password;
<topic_name> — topic name.

Download the root certificate and place it in the ~/.kafka/ folder:

mkdir -p ~/.kafka/
wget https://storage.dbaas.selcloud.ru/CA.pem -O ~/.kafka/root.crt
chmod 0600 ~/.kafka/root.crt

Create the file scram.go:

scram.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()
}

Use the connection example for the concumer:

package main

import (
    "crypto/tls"
    "crypto/x509"
    "fmt"
    "io/ioutil"
    "os"
    "os/signal"
    "strings"

    "github.com/IBM/sarama"
)

const BROKERS = "<host>:<port>"
const USER = "<user_name>"
const PASSWORD = "<password>"
const TOPIC = "<topic_name>"
const CERT = "<path>"

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
}

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<user_name> — the name of the user with the role of concumer who has access to the topic;
<password> — user password;
<topic_name> — topic name;
<path> — the full path to the root certificate.

Use the connection example for the producer:

package main

import (
    "crypto/tls"
    "crypto/x509"
    "fmt"
    "io/ioutil"
    "strings"

    "github.com/IBM/sarama"
)

const BROKERS = "<host>:<port>"
const USER = "<user_name>"
const PASSWORD = "<password>"
const TOPIC = "<topic_name>"
const CERT = "<path>"

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)
    }
}

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<user_name> — the name of the user with the role of concumer who has access to the topic;
<password> — user password;
<topic_name> — topic name;
<path> — the full path to the root certificate.

Download the root certificate and place it in the ~/.kafka/ folder:

mkdir -p ~/.kafka/
wget https://storage.dbaas.selcloud.ru/CA.pem -O ~/.kafka/root.crt
chmod 0600 ~/.kafka/root.crt

Install the kafkajs library:
```
npm install kafkajs
```

Use the connection example for the concumer:

const fs = require('fs');
const { Kafka } = require('kafkajs');

const config = {
  brokers: ['<host>:<port>'],
  ssl: {
    ca: [fs.readFileSync('<path>', 'utf-8')],
  },
  sasl: {
    mechanism: 'scram-sha-512',
    username: '<user_name>',
    password: '<password>',
  },
};

const kafka = new Kafka(config);
const consumer = kafka.consumer({ groupId: 'example' });

const run = async () => {
  await consumer.connect();
  await consumer.subscribe({
    topic: '<topic_name>',
    fromBeginning: true,
  });

  await consumer.run({
    eachMessage: async ({ message }) => {
     console.log({ value: message.value.toString() });
    },
  });
};

run();

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<user_name> — the name of the user with the role of concumer who has access to the topic;
<password> — user password;
<path> — the full path to the root certificate;
<topic_name> — topic name.

Use the connection example for the producer:

const fs = require('fs');
const { Kafka } = require('kafkajs');

const config = {
  brokers: ['<host>:<port>'],
  ssl: {
    ca: [fs.readFileSync('<path>', 'utf-8')],
  },
  sasl: {
    mechanism: 'scram-sha-512',
    username: '<user_name>',
    password: '<password>',
  },
};

const kafka = new Kafka(config);
const producer = kafka.producer();

const run = async () => {
  await producer.connect();
  await producer.send({
    topic: '<topic_name>',
    messages: [{ value: 'message' }],
  });
  await producer.disconnect();
};

run();

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<path> — the full path to the root certificate;
<user_name> — the name of the user with the producer role who has access to the topic;
<password> — user password;
<topic_name> — topic name.

Download the root certificate and place it in the ~/.kafka/ folder:

mkdir -p ~/.kafka/
wget https://storage.dbaas.selcloud.ru/CA.pem -O ~/.kafka/root.crt
chmod 0600 ~/.kafka/root.crt

Navigate to the directory where the Java certificate store will be located:
```
cd /etc/security
```
Add the SSL certificate to the Java trusted certificate store (Java Key Store):
```
keytool \
-importcert \
-alias RootCA \
-file ~/.kafka/root.crt \
-keystore ssl \
-storepass <keystore password> \
-noprompt
```
Specify <keystore_password> -keystore password for additional security.

Create a configuration file for Maven:

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.example</groupId>
    <artifactId>app</artifactId>
    <packaging>jar</packaging>
    <version>1.0</version>
    <properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>2.0.7</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.14.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.4.0</version>
        </dependency>
    </dependencies>
    <build>
        <finalName>${project.artifactId}-${project.version}</finalName>
        <sourceDirectory>src</sourceDirectory>
        <resources>
            <resource>
                <directory>src</directory>
            </resource>
        </resources>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <executions>
                    <execution>
                        <goals>
                            <goal>attached</goal>
                        </goals>
                        <phase>package</phase>
                        <configuration>
                            <descriptorRefs>
                                <descriptorRef>jar-with-dependencies</descriptorRef>
                            </descriptorRefs>
                            <archive>
                                <manifest>
                                    <mainClass>com.example.App</mainClass>
                                </manifest>
                            </archive>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-jar-plugin</artifactId>
                <version>3.1.0</version>
                <configuration>
                    <archive>
                        <manifest>
                            <mainClass>com.example.App</mainClass>
                        </manifest>
                    </archive>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

Use the connection example for the concumer:

src/com/example/App.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 = "<host>:<port>";
    private static final String TOPIC = "<topic_name>";
    private static final String USER = "<user_name>";
    private static final String PASSWORD = "<password>";
    private static final String TS_FILE = "/etc/security/ssl";
    private static final String TS_PASSWORD = "<keystore_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);

        Consumer<String, String>consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(new String[] {TOPIC}));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record: records) {
                System.out.println(record.value());
            }
        }
    }
}

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<topic_name> — topic name;
<user_name> — the name of the user with the role of concumer who has access to the topic;
<password> — user password;
<keystore_password> — keystore password.

Use the connection example for the producer:

src/com/example/App.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 = "<host>:<port>";
    private static final String TOPIC = "<topic_name>";
    private static final String USER = "<user_name>";
    private static final String PASSWORD = "<password>";
    private static final String TS_FILE = "/etc/security/ssl";
    private static final String TS_PASSWORD = "<keystore_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<String, String> producer = new KafkaProducer<>(props);
        try {
            producer.send(new ProducerRecord<String, String>(TOPIC, "key", "message")).get();
            producer.flush();
            producer.close();
        } catch (Exception ex) {
            System.out.println(ex);
            producer.close();
        }
    }
}

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<topic_name> — topic name;
<user_name> — name of the user with the producer role who has access to the topic;
<password> — user password;
<keystore_password> — certificate password.

Build and run the application:

mvn clean package
java -jar target/app-1.0-jar-with-dependencies.jar

Connect without SSL

Open the CLI.
Use the connection example for the concumer:
```
kcat -C \
-b <host>:<port> \
-t <topic_name> \
-X sasl.username=<user_name> \
-X sasl.password=<password> \
-X security.protocol=SASL_PLAINTEXT \
-X sasl.mechanisms=SCRAM-SHA-512
```
Specify:
- <host> — DNS address of the node;
- <port> — port for connection;
- <topic_name> — topic name;
- <user_name> — the name of the user with the role of concumer who has access to the topic;
- <password> — user password.
Use the connection example for the producer:
```
kcat -P \
-b <host>:<port> \
-t <topic_name> \
-X sasl.username=<user_name> \
-X sasl.password=<password> \
-X security.protocol=SASL_PLAINTEXT \
-X sasl.mechanisms=SCRAM-SHA-512
```
Specify:
- <host> — DNS address of the node;
- <port> — port for connection;
- <topic_name> — topic name;
- <user_name> — the name of the user with the producer role who has access to the topic;
- <password> — user password.

Install the confluent-kafka library:
```
pip install confluent-kafka
```

Use the connection example for the concumer:

import confluent_kafka

def error_callback(err):
    raise err

consumer = confluent_kafka.Consumer({
    "bootstrap.servers": "<host>:<port>",
    "security.protocol": "SASL_PLAINTEXT",
    "sasl.mechanism": "SCRAM-SHA-512",
    "sasl.username": "<user_name>",
    "sasl.password": "<password>",
    "group.id": "example",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
    "error_cb": error_callback,
})

consumer.subscribe(["<topic_name>"])

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()

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<user_name> — the name of the user with the role of concumer who has access to the topic;
<password> — user password;
<topic_name> — topic name.

Use the connection example for the producer:

import confluent_kafka

def error_callback(err):
    raise err

producer = confluent_kafka.Producer({
    "bootstrap.servers": "<host>:<port>",
    "security.protocol": "SASL_PLAINTEXT",
    "sasl.mechanism": "SCRAM-SHA-512",
    "sasl.username": "<user_name>",
    "sasl.password": "<password>",
    "error_cb": error_callback,
})
producer.produce("<topic_name>", "message")
producer.flush(60)

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<user_name> — database user name;
<password> — user password;
<topic_name> — topic name.

Install the kafka-python library:
```
pip install kafka-python
```

Use the connection example for the concumer:

import kafka

consumer = kafka.KafkaConsumer(
    "<topic_name>",
    bootstrap_servers="<host>:<port>",
    security_protocol="SASL_PLAINTEXT",
    sasl_mechanism="SCRAM-SHA-512",
    sasl_plain_username="<user_name>",
    sasl_plain_password="<password>",
)
for record in consumer:
    print(record)

Specify:

<topic_name> — topic name
<host> — DNS address of the node;
<port> — port for connection;
<user_name> — the name of the user with the role of concumer who has access to the topic;
<password> — user password.

Use the connection example for the producer:

import kafka

producer = kafka.KafkaProducer(
    bootstrap_servers="<host>:<port>",
    security_protocol="SASL_PLAINTEXT",
    sasl_mechanism="SCRAM-SHA-512",
    sasl_plain_username="<user_name>",
    sasl_plain_password="<password>",
)
future = producer.send(
    topic="<topic_name>",
    value=b"message",
)
result = future.get(timeout=60)
print(result)

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<user_name> — database user name;
<password> — user password;
<topic_name> — topic name.

Create the file scram.go:

scram.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()
}

Use the connection example for the concumer:

package main

import (
    "fmt"
    "os"
    "os/signal"
    "strings"

    "github.com/IBM/sarama"
)

const BROKERS = "<host>:<port>"
const USER = "<user_name>"
const PASSWORD = "<password>"
const TOPIC = "<topic_name>"

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
}

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<user_name> — the name of the user with the role of concumer who has access to the topic;
<password> — user password;
<topic_name> — topic name.

Use the connection example for the producer:

package main

import (
    "fmt"
    "strings"

    "github.com/IBM/sarama"
)

const BROKERS = "<host>:<port>"
const USER = "<user_name>"
const PASSWORD = "<password>"
const TOPIC = "<topic_name>"

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)
    }
}

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<user_name> — the name of the user with the role of concumer who has access to the topic;
<password> — user password;
<topic_name> — topic name.

Install the kafkajs library:
```
npm install kafkajs
```

Use the connection example for the concumer:

const { Kafka } = require('kafkajs');

const config = {
  brokers: ['<host>:<port>'],
  ssl: false,
  sasl: {
    mechanism: 'scram-sha-512',
    username: '<user_name>',
    password: '<password>',
  },
};

const kafka = new Kafka(config);
const consumer = kafka.consumer({ groupId: 'example' });

const run = async () => {
  await consumer.connect();
  await consumer.subscribe({
    topic: '<topic_name>',
    fromBeginning: true,
  });

  await consumer.run({
    eachMessage: async ({ message }) => {
      console.log({ value: message.value.toString() });
    },
  });
};

run();

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<user_name> — the name of the user with the role of concumer who has access to the topic;
<password> — user password;
<topic_name> — topic name.

Use the connection example for the producer:

const { Kafka } = require('kafkajs');

const config = {
  brokers: ['<host>:<port>'],
  ssl: false,
  sasl: {
    mechanism: 'scram-sha-512',
    username: '<user_name>',
    password: '<password>',
  },
};

const kafka = new Kafka(config);
const producer = kafka.producer();

const run = async () => {
  await producer.connect();
  await producer.send({
    topic: '<topic_name>',
    messages: [{ value: 'message' }],
  });
  await producer.disconnect();
};

run();

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<user_name> — database user name;
<password> — user password;
<topic_name> — topic name.

Create a configuration file for Maven:

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.example</groupId>
    <artifactId>app</artifactId>
    <packaging>jar</packaging>
    <version>1.0</version>
    <properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>2.0.7</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.14.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.4.0</version>
        </dependency>
    </dependencies>
    <build>
        <finalName>${project.artifactId}-${project.version}</finalName>
        <sourceDirectory>src</sourceDirectory>
        <resources>
            <resource>
                <directory>src</directory>
            </resource>
        </resources>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <executions>
                    <execution>
                        <goals>
                            <goal>attached</goal>
                        </goals>
                        <phase>package</phase>
                        <configuration>
                            <descriptorRefs>
                                <descriptorRef>jar-with-dependencies</descriptorRef>
                            </descriptorRefs>
                            <archive>
                                <manifest>
                                    <mainClass>com.example.App</mainClass>
                                </manifest>
                            </archive>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-jar-plugin</artifactId>
                <version>3.1.0</version>
                <configuration>
                    <archive>
                        <manifest>
                            <mainClass>com.example.App</mainClass>
                        </manifest>
                    </archive>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

Use the connection example for the concumer:

src/com/example/App.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 = "<host>:<port>";
    private static final String TOPIC = "<topic_name>";
    private static final String USER = "<user_name>";
    private static final String PASSWORD = "<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);

        Consumer<String, String>consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(new String[] {TOPIC}));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record: records) {
                System.out.println(record.value());
            }
        }
    }
}

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<topic_name> — topic name;
<user_name> — the name of the user with the role of concumer who has access to the topic;
<password> — user password.

Use the connection example for the producer:

src/com/example/App.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 = "<host>:<port>";
    private static final String TOPIC = "<topic_name>";
    private static final String USER = "<user_name>";
    private static final String PASSWORD = "<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<String, String> producer = new KafkaProducer<>(props);
        try {
            producer.send(new ProducerRecord<String, String>(TOPIC, "key", "message")).get();
            producer.flush();
            producer.close();
        } catch (Exception ex) {
            System.out.println(ex);
            producer.close();
        }
    }
}

Specify:

<host> — DNS address of the node;
<port> — port for connection;
<topic_name> — topic name;
<user_name> — database user name;
<password> — user password.

Build and run the application:

mvn clean package
java -jar target/app-1.0-jar-with-dependencies.jar