consumer.go

package main

import (
	"fmt"
	"github.com/confluentinc/confluent-kafka-go/kafka"
)

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

	if err != nil {
		panic(err)
	}

	c.SubscribeTopics([]string{"users.notification"}, nil)

	for {
		fmt.Println("Waiting for messages...")

		msg, err := c.ReadMessage(-1)
		if err == nil {
			fmt.Println("Topic : ", *msg.TopicPartition.Topic)
			fmt.Println("Partition : ", msg.TopicPartition.Partition)
			fmt.Println("Offset : ", msg.TopicPartition.Offset)
			fmt.Println("Value : ", string(msg.Value))
			c.CommitMessage(msg)
		} else {
			// The client will automatically try to recover from all errors.
			fmt.Printf("Consumer error: %v (%v)\n", err, msg)
		}
	}

	c.Close()
}