Подключение go к apache kafka

В целях отладки развернул локально 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()
    }

}

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