2

В целях отладки развернул локально apache kafka с использованием этого прокта. Модифицировал кофигурационный файл следующим образом:

version: '2'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"
    networks:
      static-network:
        ipv4_address: 172.20.128.3
  kafka:
    build: .
    container_name: kafka
    restart: always
    ports:
      - "22299:9092"
    environment:
      KAFKA_CREATE_TOPICS: "kad_test_kafka_topic:1:1"
      KAFKA_ADVERTISED_HOST_NAME: 172.20.128.2
      KAFKA_ZOOKEEPER_CONNECT: 172.20.128.3:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
    networks:
      static-network:
        ipv4_address: 172.20.128.2

networks:
  static-network:
    ipam:
      config:
        - subnet: 172.20.0.0/16
          #docker-compose v3+ do not use ip_range
          ip_range: 172.20.128.0/24

Для отладки самого подключения использую следующий код на go:

package main

import (
    "fmt"
    "log"
    "os"

    "github.com/confluentinc/confluent-kafka-go/kafka"
    "github.com/google/uuid"
    "github.com/vmihailenco/msgpack"
)

const (
    kafkaServer = "172.20.128.2:22299"
    kafkaTopic  = "kad_test_kafka_topic"
    kafkaGroup  = "kad_test_group_id"
)

func errCheck(err error) {
    if err != nil {
        log.Fatalln(err)
    }
}

func printHelp() {
    fmt.Println("Three arguments avaliable:")
    fmt.Println("p - run kafka producer")
    fmt.Println("c - run kafka consumer")
    fmt.Println("cs - run kafka consumer in subscribe mode")
}

func main() {
    consoleArgs := os.Args
    if len(consoleArgs) < 2 {
        printHelp()
        log.Fatalln("No arguments")
    }

    if consoleArgs[1] == "p" {
        p, err := kafka.NewProducer(&kafka.ConfigMap{
            "bootstrap.servers": kafkaServer})
        if err != nil {
            log.Panic(err)
        }
        log.Println("Producer created successfull")
        defer p.Close()

        // Delivery report handler for produced messages
        go func() {
            log.Println("Run event listener")
            for e := range p.Events() {
                switch ev := e.(type) {
                case *kafka.Message:
                    if ev.TopicPartition.Error != nil {
                        log.Printf("Delivery failed: %v\n", ev.TopicPartition)
                    } else {
                        log.Printf("Delivered message to %v\n", ev.TopicPartition)
                    }
                }
            }
        }()

        log.Println("Start main loop")
        for i := 0; i < 3; i++ {
            newUUID, err := uuid.NewRandom()
            errCheck(err)
            value := make(map[string]string)
            value["Key1"] = "Val1"
            value["Key2"] = "Val2"
            value["Key3"] = "Val3"
            value["UUID"] = newUUID.String()

            binaryMessage, err := msgpack.Marshal(&value)
            errCheck(err)

            // Produce messages to topic (asynchronously)
            topic := kafkaTopic
            p.Produce(&kafka.Message{
                TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
                Value:          binaryMessage,
            }, nil)
            log.Printf("Sended message with UUID: %s\n", newUUID.String())
        }

        // Wait for message deliveries before shutting down
        log.Println("Wait for message delivering")
        p.Flush(15 * 1000)
        log.Println("---> Producer stop <---")
    }

    if consoleArgs[1] == "c" || consoleArgs[1] == "cs" {
        kafkaConsumer, err := kafka.NewConsumer(&kafka.ConfigMap{
            "bootstrap.servers":        kafkaServer,
            "group.id":                 kafkaGroup,
            "enable.auto.offset.store": "false",
            "auto.offset.reset":        "earliest"})
        errCheck(err)
        if kafkaConsumer == nil {
            log.Fatalln("No kafka connection")
        }
        defer kafkaConsumer.Close()
        log.Println("---> Consumer start <---")

        if consoleArgs[1] == "cs" {
            kafkaConsumer.SubscribeTopics([]string{kafkaTopic}, nil)
            log.Println("Consumer subscribed successfull")
        }

        if consoleArgs[1] == "c" {
            topicPatritions := make([]kafka.TopicPartition, 1)
            topic := kafkaTopic
            topicPatritions[0] = kafka.TopicPartition{Topic: &topic, Partition: 0}
            err = kafkaConsumer.Assign(topicPatritions)
            errCheck(err)
            log.Println("Consumer assigned successfull")
        }

        log.Println("Main loop start")
        for i := 0; i < 3; i++ {
            msg, err := kafkaConsumer.ReadMessage(-1)
            errCheck(err)
            log.Println(msg.Value)
        }
        log.Println("---> Consumer stop <---")
    }

    if consoleArgs[1] != "c" && consoleArgs[1] != "p" && consoleArgs[1] != "cs" {
        printHelp()
    }

}

Но в итоге не могу и не принять, и не отправить сообщение. Прошу помощи в указании моей ошибки.

2
  • 1) может го тоже в контейнер, в тот же yml; а папку с бинарником в вольюм; 2) попробовать утилиты в комплекте с кафкой 1 сен 2020 в 19:12
  • Насколько я понимаю разницы особой не будет, если го в контейнер упаковать. Kafka как отдельный микросервис по сути и является абстракцией. В реальном окружении действующего кода Kafka так же будет внешним компонентом. Доступ по ip есть с хостовой машины, значит у меня где-то косяк в логике подключения этих компонентов между собой. Вот не могу понять где именно. А что за утилиты?
    – anderson
    1 сен 2020 в 19:45

2 ответа 2

2

После дополнительных поисков наткнулся на ответ, в котором приводится действительно работающая конфигурация docker-compose. Так же отвечающий написал отдельную статью, поясняющую как правильно понимать те или иные параметры. После запуска конфигурации сообщения отправляются и принимаются из любого клиента (python \ go etc.).

version: '2'
services:
zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
    ZOOKEEPER_CLIENT_PORT: 2181
    ZOOKEEPER_TICK_TIME: 2000

kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
    - zookeeper
    ports:
    - 29092:29092
    environment:
    KAFKA_BROKER_ID: 1
    KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
    KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
    KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
    KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

Автор дополнительно обращает внимание, что текущая конфигурация предоставляет порт 29092 именно наружу, т.е. именно по этому порту будут обрабатываться все внешние входящие подключения. Использовать порт 9092 необходимо для внутренних подключений в рамках docker-контейнеров. Так же необходимо заменить значение localhost на адрес хоста для docker, если необходимо обрабатывать внешние (относительно машины с docker) подключения.

2
  • хорошо; дальше еще квест будет, если кластер настраивать и ssl 3 сен 2020 в 9:09
  • В моём случае этого не требуется. С моей стороны необходимо подготовить программу, которая будет взаимодействовать с kafka и для меня kafka лишь абстракция. Настройкой кластера занимается отдельно специально-обученный человек. Моя задача была лишь в развёртывании отладочной среды.
    – anderson
    3 сен 2020 в 9:23
1

Написано хитро, настараживает, например - "22299:9092". Можно попробовать одинаковые порты.

Если не вышло:

Чтобы понять, где проблема (в go, или в yml), я бы начал с простого конфига:

  • zookeeper
  • kafka
  • тестовый контейнер с утилитами (если кафку скачать, там с ней простые клиенты прилагаются)

Проверил, что работает -- docker exec ... в тестовый контейнер.

Потом можно:

  • попробовать в тестовом контейнере go -- volume в папку с результатом компиляции
  • или сделать хитрую сеть, и проверить утилитами.
2
  • От хитрой сети как раз таки хочу уйти и сделать максимально просто. 22299:9092 - это обычный проброс порта. Конфиг сделан с зафиксированными ip-адресами и портами для упрощения: я гарантировано знаю где находится kafka.
    – anderson
    2 сен 2020 в 5:23
  • вероятно kafka считает, что она на 172.20.128.2:9092; парного ADVERTISED параметрa для порта я не вижу 2 сен 2020 в 6:59

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge you have read our privacy policy.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.