golang confluent kafka consumer doesn't receive messages

I was trying to create a simple producer / consumer kafka duo. Since producer was successfully working, according to the examples in confluent's github page, I had trouble while implementing consumer. I use cloud kafka broker, which is Cloudkarafka. The consumer.go code is below:

func main() {
config := &kafka.ConfigMap{
	"metadata.broker.list":            "XXXXXXX", // 3 hosts Cloudkarafka provides to me
	"security.protocol":               "SASL_SSL",
	"sasl.mechanisms":                 "SCRAM-SHA-256",
	"sasl.username":                   "XXXXXXXX", // My username provided by Cloudkarafka
	"sasl.password":                   "XXXXXXXX", // My password provided by 
	"group.id":                        "cloudkarafka-example",
	"go.events.channel.enable":        true,
	"go.application.rebalance.enable": true,
	"default.topic.config":            kafka.ConfigMap{"auto.offset.reset": "earliest"},
	//"debug":                           "generic,broker,security",
}
topic := "XXXXXX" + "A" // username + "A"

consumer, err := kafka.NewConsumer(config)

if err != nil {
	panic(fmt.Sprintf("Failed to create consumer: %s", err))
}

topics := []string{topic}
//consumer.SubscribeTopics(topics, nil)
err = consumer.SubscribeTopics(topics, nil)
run := true
for run == true {
	ev := consumer.Poll(0)
	switch e := ev.(type) {
	case *kafka.Message:
		fmt.Printf("%% Message on %s:\n%s\n",
			e.TopicPartition, string(e.Value))
	case kafka.PartitionEOF:
		fmt.Printf("%% Reached %v\n", e)
	case kafka.Error:
		fmt.Fprintf(os.Stderr, "%% Error: %v\n", e)
		run = false
	default:
		fmt.Printf("Ignored %v\n", e)
	}
}

consumer.Close()
}

The problem here I get is, even though I produce messages to the same topic, consumer always stays in the default case, and constantly gives the output "Ignored <nil> ". Since I feel beginner to these topics, any help & suggestion would be appreciated.

ps: I use Windows 11, in the details it says "confluent-kafka-go is not supported on Windows" but the code works just stays in default state, also the producer part just works fine.

producer.go:

config := &kafka.ConfigMap{
	"metadata.broker.list": "XXXXXXXXXX",
	"security.protocol":    "SASL_SSL",
	"sasl.mechanisms":      "SCRAM-SHA-256",
	"sasl.username":        "XXXXXXXXX",
	"sasl.password":        "XXXXXXXXX",
	"group.id":             "cloudkarafka-example",
	"default.topic.config": kafka.ConfigMap{"auto.offset.reset": "earliest"},
	//"debug":                           "generic,broker,security",
}
topic := "XXXXX-" + "A"
p, err := kafka.NewProducer(config)
if err != nil {
	fmt.Printf("Failed to create producer: %s\n", err)
	os.Exit(1)
}
fmt.Printf("Created Producer %v\n", p)
deliveryChan := make(chan kafka.Event)

for i := 0; i < 10; i++ {
	value := fmt.Sprintf("[%d] Hello Go!", i+1)
	err = p.Produce(&kafka.Message{TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny}, Value: []byte(value)}, deliveryChan)
	e := <-deliveryChan
	m := e.(*kafka.Message)
	if m.TopicPartition.Error != nil {
		fmt.Printf("Delivery failed: %v\n", m.TopicPartition.Error)
	} else {
		fmt.Printf("Delivered message to topic %s [%d] at offset %v\n",
			*m.TopicPartition.Topic, m.TopicPartition.Partition, m.TopicPartition.Offset)
	}
}
close(deliveryChan)

Poll() will return nil on timeout. Since you are specifying a timeout of 0ms, I suspect that what you are seeing is the behaviour of a consumer with no messages to consume.

i.e. you are asking it to wait 0ms for new messages, there are never any new messages so the Poll() call is immediately returning nil every time, all the time. Without a specific nil case, these are handled by your default case.

1. Are you SURE you are producing messages to the same topic your consumer is subscribed to?<br/><br/>As Andrey pointed out in his comment, either your topic ids are different or you have obfuscated them differently in your consumer vs producer example code. It may be more helpful to attempt first reproducing your problem with a configuration that does not require obfuscation, to avoid uncertainty on such points.

2. Are you getting any error from Subscribe() (why aren't you checking this)?

3. How long have you waited to see messages consumed?<br/><br/>It can take a few seconds for a broker to accept a new consumer into a group; with a 0ms timeout, you may see lots of "no message" events before you eventually start receiving any waiting messages.

For a minimal, working example, I'd suggest keeping things as simple as possible:

• You don't need to configure go.events.channel.enable if you are using Poll() to read messages

• You don't need to configure go.application.rebalance.enable if you aren't interested in, and don't need to modify, initial offsets.

• If you aren't interested in events such as PartitionEOF etc (and you likely aren't) then you might want to consider using the higher-level consumer.ReadMessage() function rather than Poll() (ReadMessage returns only messages or errors and ignores all other events).

I have faced this exact issue the problem is documented in the library

//   go.events.channel.enable (bool, false) - [deprecated] Enable the Events() channel. Messages and events will be pushed on the Events() channel and the Poll() interface will be disabled.
//   go.events.channel.size (int, 1000) - Events() channel size
//   go.logs.channel.enable (bool, false) - Forward log to Logs() channel.
//   go.logs.channel (chan kafka.LogEvent, nil) - Forward logs to application-provided channel instead of Logs(). Requires go.logs.channel.enable=true.
//
// WARNING: Due to the buffering nature of channels (and queues in general) the
// use of the events channel risks receiving outdated events and
// messages. Minimizing go.events.channel.size reduces the risk
// and number of outdated events and messages but does not eliminate
// the factor completely. With a channel size of 1 at most one
// event or message may be outdated.

Dont set go.events.channel.enable to true