Подключение к топикам в кластере Apache Kafka®

Настройте все группы безопасности кластера так, чтобы они разрешали входящий трафик с любых IP-адресов на порт 9091. Для этого создайте следующее правило для входящего трафика:

• протокол: TCP;
• диапазон портов: 9091;
• тип источника: CIDR;
• источник: 0.0.0.0/0.

1. Запустите команду получения сообщений из топика:

   kafkacat -C \
            -b <FQDN брокера>:9092 \
            -t <имя топика> \
            -X security.protocol=SASL_PLAINTEXT \
            -X sasl.mechanisms=SCRAM-SHA-512 \
            -X sasl.username=<логин потребителя> \
            -X sasl.password=<пароль потребителя> -Z -K:
   

   Команда будет непрерывно считывать новые сообщения из топика.

2. В отдельном терминале запустите команду отправки сообщения в топик:

   echo "test message" | kafkacat -P \
          -b <FQDN брокера>:9092 \
          -t <имя топика> \
          -k key \
          -X security.protocol=SASL_PLAINTEXT \
          -X sasl.mechanisms=SCRAM-SHA-512 \
          -X sasl.username=<логин производителя> \
          -X sasl.password=<пароль производителя> -Z
   

1. Пример кода для отправки сообщений в топик:

   cs-project/producer/Program.cs

   using Confluent.Kafka;
   using System;
   using System.Collections.Generic;
   
   namespace App
   {
       class Program
       {
           public static void Main(string[] args)
           {
               int MSG_COUNT = 5;
   
               string HOST = "<FQDN хоста-брокера>:9092";
               string TOPIC = "<имя топика>";
               string USER = "<имя производителя>";
               string PASS = "<пароль производителя>";
   
               var producerConfig = new ProducerConfig(
                   new Dictionary<string,string>{
                       {"bootstrap.servers", HOST},
                       {"security.protocol", "SASL_PLAINTEXT"},
                       {"sasl.mechanisms", "SCRAM-SHA-512"},
                       {"sasl.username", USER},
                       {"sasl.password", PASS}
                   }
               );
   
               var producer = new ProducerBuilder<string, string>(producerConfig).Build();
   
               for(int i=0; i<MSG_COUNT; i++)
               {
                   producer.Produce(TOPIC, new Message<string, string> { Key = "key", Value = "test message" },
                   (deliveryReport) =>
                   {
                       if (deliveryReport.Error.Code != ErrorCode.NoError)
                       {
                           Console.WriteLine($"Failed to deliver message: {deliveryReport.Error.Reason}");
                       }
                       else
                       {
                       Console.WriteLine($"Produced message to: {deliveryReport.TopicPartitionOffset}");
                       }
                   });
               }
               producer.Flush(TimeSpan.FromSeconds(10));
           }
       }
   }
   

2. Пример кода для получения сообщений из топика:

   cs-project/consumer/Program.cs

   using Confluent.Kafka;
   using System;
   using System.Collections.Generic;
   
   namespace CCloud
   {
       class Program
       {
           public static void Main(string[] args)
           {
               string HOST = "<FQDN хоста-брокера>:9092";
               string TOPIC = "<имя топика>";
               string USER = "<имя потребителя>";
               string PASS = "<пароль потребителя>";
   
               var consumerConfig = new ConsumerConfig(
                   new Dictionary<string,string>{
                       {"bootstrap.servers", HOST},
                       {"security.protocol", "SASL_PLAINTEXT"},
                       {"sasl.mechanisms", "SCRAM-SHA-512"},
                       {"sasl.username", USER},
                       {"sasl.password", PASS},
                       {"group.id", "demo"}
                   }
               );
   
               var consumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
               consumer.Subscribe(TOPIC);
               try
               {
                   while (true)
                   {
                       var cr = consumer.Consume();
                       Console.WriteLine(cr.Message.Value);
                   }
               }
               catch (OperationCanceledException)
               {
                   // Ctrl-C was pressed.
               }
               finally
               {
                   consumer.Close();
               }
           }
       }
   }
   

3. Сборка и запуск приложений:

   cd ~/cs-project/consumer && dotnet build && \
   dotnet run bin/Debug/netcoreapp5.0/App.dll
   

   cd ~/cs-project/producer && dotnet build && \
   dotnet run bin/Debug/netcoreapp5.0/App.dll
   

1. Пример кода для отправки сообщения в топик:

   producer/main.go

   package main
   
   import (
         "fmt"
         "os"
         "strings"
   
         "github.com/Shopify/sarama"
   )
   
   func main() {
         brokers := "<FQDN хоста-брокера>:9092"
         splitBrokers := strings.Split(brokers, ",")
         conf := sarama.NewConfig()
         conf.Producer.RequiredAcks = sarama.WaitForAll
         conf.Producer.Return.Successes = true
         conf.Version = sarama.V0_10_0_0
         conf.ClientID = "sasl_scram_client"
         conf.Net.SASL.Enable = true
         conf.Net.SASL.Handshake = true
         conf.Net.SASL.User = "<имя производителя>"
         conf.Net.SASL.Password = "<пароль производителя>"
         conf.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA512} }
         conf.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA512)
   
         syncProducer, err := sarama.NewSyncProducer(splitBrokers, conf)
         if err != nil {
               fmt.Println("Couldn't create producer: ", err.Error())
               os.Exit(0)
         }
         publish("test message", syncProducer)
   }
   
   func publish(message string, producer sarama.SyncProducer) {
     // publish sync
     msg := &sarama.ProducerMessage {
         Topic: "<имя топика>",
         Value: sarama.StringEncoder(message),
     }
     p, o, err := producer.SendMessage(msg)
     if err != nil {
         fmt.Println("Error publish: ", err.Error())
     }
   
     fmt.Println("Partition: ", p)
     fmt.Println("Offset: ", o)
   }
   

2. Пример кода для получения сообщений из топика:

   consumer/main.go

   package main
   
   import (
         "fmt"
         "os"
         "os/signal"
         "strings"
   
         "github.com/Shopify/sarama"
   )
   
   func main() {
         brokers := "<FQDN хоста-брокера>:9092"
         splitBrokers := strings.Split(brokers, ",")
         conf := sarama.NewConfig()
         conf.Producer.RequiredAcks = sarama.WaitForAll
         conf.Version = sarama.V0_10_0_0
         conf.Consumer.Return.Errors = true
         conf.ClientID = "sasl_scram_client"
         conf.Metadata.Full = true
         conf.Net.SASL.Enable = true
         conf.Net.SASL.User =  "<имя потребителя>"
         conf.Net.SASL.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)
   
         master, err := sarama.NewConsumer(splitBrokers, conf)
         if err != nil {
                 fmt.Println("Coulnd't create consumer: ", err.Error())
                 os.Exit(1)
         }
   
         defer func() {
                 if err := master.Close(); err != nil {
                         panic(err)
                 }
         }()
   
         topic := "<имя топика>"
   
         consumer, err := master.ConsumePartition(topic, 0, sarama.OffsetOldest)
         if err != nil {
                 panic(err)
         }
   
         signals := make(chan os.Signal, 1)
         signal.Notify(signals, os.Interrupt)
   
         // Count how many message processed
         msgCount := 0
   
         // Get signal for finish
         doneCh := make(chan struct{})
         go func() {
                 for {
                         select {
                         case err := <-consumer.Errors():
                                 fmt.Println(err)
                         case msg := <-consumer.Messages():
                                 msgCount++
                                 fmt.Println("Received messages", string(msg.Key), string(msg.Value))
                         case <-signals:
                                 fmt.Println("Interrupt is detected")
                                 doneCh <- struct{}{}
                         }
                 }
         }()
   
         <-doneCh
         fmt.Println("Processed", msgCount, "messages")
   }
   

3. Сборка приложений:

   cd ~/go-project/producer && go build && \
   cd ~/go-project/consumer && go build
   

4. Запуск приложений:

   ~/go-project/consumer/consumer
   

   ~/go-project/producer/producer
   

1. Пример кода для отправки сообщений в топик:

   producer/src/java/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 {
   
     public static void main(String[] args) {
   
       int MSG_COUNT = 5;
   
       String HOST = "<FQDN брокера>:9092";
       String TOPIC = "<имя топика>";
       String USER = "<имя производителя>";
       String PASS = "<пароль производителя>";
   
       String jaasTemplate = "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"%s\" password=\"%s\";";
       String jaasCfg = String.format(jaasTemplate, USER, PASS);
       String KEY = "key";
   
       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 {
        for (int i = 1; i <= MSG_COUNT; i++){
          producer.send(new ProducerRecord<String, String>(TOPIC, KEY, "test message")).get();
          System.out.println("Test message " + i);
         }
        producer.flush();
        producer.close();
       } catch (Exception ex) {
           System.out.println(ex);
           producer.close();
       }
     }
   }
   

2. Пример кода для получения сообщений из топика:

   consumer/src/java/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 {
   
     public static void main(String[] args) {
   
       String HOST = "<FQDN брокера>:9092";
       String TOPIC = "<имя топика>";
       String USER = "<имя потребителя>";
       String PASS = "<пароль потребителя>";
   
       String jaasTemplate = "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"%s\" password=\"%s\";";
       String jaasCfg = String.format(jaasTemplate, USER, PASS);
       String GROUP = "demo";
   
       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", GROUP);
       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.key() + ":" + record.value());
         }
       }
     }
   }
   

3. Сборка приложений:

   cd ~/project/producer && mvn clean package && \
   cd ~/project/consumer && mvn clean package
   

4. Запуск приложений:

   java -jar ~/project/producer/target/app-0.1.0-jar-with-dependencies.jar
   

   java -jar ~/project/consumer/target/app-0.1.0-jar-with-dependencies.jar
   

1. Пример кода для отправки сообщений в топик:

   producer.js

   "use strict"
   const Kafka = require('node-rdkafka');
   
   const MSG_COUNT = 5;
   
   const HOST = "<FQDN брокера>:9092";
   const TOPIC = "<имя топика>";
   const USER = "<имя производителя>";
   const PASS = "<пароль производителя>";
   
   const producer = new Kafka.Producer({
     'bootstrap.servers': HOST,
     'sasl.username': USER,
     'sasl.password': PASS,
     'security.protocol': "SASL_PLAINTEXT",
     'sasl.mechanisms': "SCRAM-SHA-512"
   });
   
   producer.connect();
   
   producer.on('ready', function() {
     try {
       for (let i = 0; i < MSG_COUNT; ++i) {
         producer.produce(TOPIC, -1, Buffer.from("test message"), "key");
         console.log("Produced: test message");
       }
   
       producer.flush(10000, () => {
           producer.disconnect();
         });
     } catch (err) {
       console.error('Error');
       console.error(err);
     }
   });
   

2. Пример кода для получения сообщений из топика:

   consumer.js

   "use strict"
   const Kafka = require('node-rdkafka');
   
   const MSG_COUNT = 5;
   
   const HOST = "<FQDN брокера>:9092";
   const TOPIC = "<имя топика>";
   const USER = "<имя потребителя>";
   const PASS = "<пароль потребителя>";
   
   const consumer = new Kafka.Consumer({
     'bootstrap.servers': HOST,
     'sasl.username': USER,
     'sasl.password': PASS,
     'security.protocol': "SASL_PLAINTEXT",
     'sasl.mechanisms': "SCRAM-SHA-512",
     'group.id': "demo"
   });
   
   consumer.connect();
   
   consumer
     .on('ready', function() {
       consumer.subscribe([TOPIC]);
       consumer.consume();
     })
     .on('data', function(data) {
       console.log(data.value.toString());
     });
   
   process.on('SIGINT', () => {
     console.log('\nDisconnecting consumer ...');
     consumer.disconnect();
   });
   

3. Запуск приложений:

   node consumer.js
   

   node producer.js
   

1. Запустите команду получения сообщений из топика:

   <путь к директории с файлами Kafka>\bin\windows\kafka-console-consumer.bat `
       --bootstrap-server <FQDN брокера>:9092 `
       --topic <имя топика> `
       --property print.key=true `
       --property key.separator=":" `
       --consumer-property security.protocol=SASL_PLAINTEXT `
       --consumer-property sasl.mechanism=SCRAM-SHA-512 `
       --consumer-property sasl.jaas.config="org.apache.kafka.common.security.scram.ScramLoginModule required username='<логин потребителя>' password='<пароль потребителя>';" 
   

   Команда будет непрерывно считывать новые сообщения из топика.

2. В отдельном терминале запустите команду отправки сообщения в топик:

   echo "key:test message" | <путь к директории с файлами Kafka>\bin\windows\kafka-console-producer.bat `
       --bootstrap-server <FQDN брокера>:9092 `
       --topic <имя топика> `
       --property parse.key=true `
       --property key.separator=":" `
       --producer-property acks=all `
       --producer-property security.protocol=SASL_PLAINTEXT `
       --producer-property sasl.mechanism=SCRAM-SHA-512 `
       --producer-property sasl.jaas.config="org.apache.kafka.common.security.scram.ScramLoginModule required username='<логин производителя>' password='<пароль производителя>';"
   

1. Пример кода для отправки сообщения в топик:

   producer.py

   from kafka import KafkaProducer
   
   producer = KafkaProducer(
       bootstrap_servers='<FQDN хоста-брокера>:9092',
       security_protocol="SASL_PLAINTEXT",
       sasl_mechanism="SCRAM-SHA-512",
       sasl_plain_password='<пароль производителя>',
       sasl_plain_username='<имя производителя>')
   
   producer.send('<имя топика>', b'test message', b'key')
   producer.flush()
   producer.close()
   

2. Пример кода для получения сообщений из топика:

   consumer.py

   from kafka import KafkaConsumer
   
   consumer = KafkaConsumer(
       '<имя топика>',
       bootstrap_servers='<FQDN брокера>:9092',
       security_protocol="SASL_PLAINTEXT",
       sasl_mechanism="SCRAM-SHA-512",
       sasl_plain_password='<пароль потребителя>',
       sasl_plain_username='<имя потребителя>')
   
   print("ready")
   
   for msg in consumer:
       print(msg.key.decode("utf-8") + ":" + msg.value.decode("utf-8"))
   

3. Запуск приложений:

   python3 producer.py
   

   python3 consumer.py