rdkafka + Go + (macOS?) receive failed

[aviramha]

Bug Description

When connecting to kafka in Go using confluent kafka we're seeing

%3|1692687932.300|FAIL|rdkafka#consumer-1| [thrd:kafka-service:9092/bootstrap]: kafka-service:9092/bootstrap: Receive failed: Undefined error: 0 (after 0ms in state UP, 5 identical error(s) suppressed)
Consumer error: kafka-service:9092/bootstrap: Receive failed: Undefined error: 0 (after 0ms in state UP, 5 identical error(s) suppressed) (<nil>)
Consumer error: 1/1 brokers are down (<nil>)

Steps to Reproduce

Clone https://github.com/metalbear-co/playground
Change the following files to use confluent kafka:

ip-visit-consumer/main.go

package main

import (
	"context"
	"fmt"
	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
	"github.com/gin-gonic/gin"
	"github.com/spf13/viper"
	"net/http"
	"time"
)

var ctx = context.Background()

// Config
// Struct that holds local service port, remote redis host and port
type Config struct {
	Port               int16
	KafkaAddress       string
	KafkaTopic         string
	KafkaConsumerGroup string
}

type IpMessage struct {
	Ip string `json:"ip"`
}

func loadConfig() Config {
	viper.BindEnv("port")
	viper.BindEnv("kafkaaddress")
	viper.BindEnv("kafkatopic")
	viper.BindEnv("kafkaconsumergroup")

	config := Config{}
	config.Port = int16(viper.GetInt("port"))
	config.KafkaAddress = viper.GetString("kafkaaddress")
	config.KafkaTopic = viper.GetString("kafkatopic")
	config.KafkaConsumerGroup = viper.GetString("kafkaconsumergroup")

	return config
}

func StartKafkaReader(address, topic, group string) {

	c, err := kafka.NewConsumer(&kafka.ConfigMap{
		"bootstrap.servers": address,
		"group.id":          group,
		"auto.offset.reset": "earliest",
	})

	if err != nil {
		panic(err)
	}

	c.SubscribeTopics([]string{topic}, nil)

	// A signal handler or similar could be used to set this to false to break the loop.
	run := true

	for run {
		msg, err := c.ReadMessage(time.Second)
		if err == nil {
			fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value))
		} else if !err.(kafka.Error).IsTimeout() {
			// The client will automatically try to recover from all errors.
			// Timeout is not considered an error because it is raised by
			// ReadMessage in absence of messages.
			fmt.Printf("Consumer error: %v (%v)\n", err, msg)
		}
	}

	c.Close()

}

func main() {
	config := loadConfig()

	go StartKafkaReader(config.KafkaAddress, config.KafkaTopic, config.KafkaConsumerGroup)

	router := gin.Default()
	router.GET("/health", func(ctx *gin.Context) { ctx.Status(http.StatusOK) })
	fmt.Print("loaded")
	router.Run("0.0.0.0:" + fmt.Sprint(config.Port))
}

add go go.mod

	github.com/confluentinc/confluent-kafka-go/v2 v2.2.0

then run go mod tidy
debug the app using GoLand

Backtrace

No response

Relevant Logs

No response

Your operating system and version

macOS m1

Local process

go binary

Local process version

No response

Additional Info

No response