mkdir kafka-golang-demo
cd kafka-golang-demo
go mod init kafka-demo
go get github.com/Shopify/sarama

package main
 
import (
    "fmt"
    "log"
    "time"
 
    "github.com/Shopify/sarama"
)
 
func main() {
    // Konfigurasi Kafka
    config := sarama.NewConfig()
    config.Producer.Return.Successes = true
    config.Producer.RequiredAcks = sarama.WaitForAll
    config.Producer.Retry.Max = 5
 
    // Koneksi ke Kafka broker
    brokers := []string{"localhost:9092"}
    producer, err := sarama.NewSyncProducer(brokers, config)
    if err != nil {
        log.Fatalf("Error creating producer: %v", err)
    }
    defer producer.Close()
 
    // Kirim pesan
    topic := "user-events"
    message := &sarama.ProducerMessage{
        Topic: topic,
        Value: sarama.StringEncoder("User John Doe registered at " + time.Now().String()),
    }
 
    partition, offset, err := producer.SendMessage(message)
    if err != nil {
        log.Printf("Error sending message: %v", err)
        return
    }
 
    fmt.Printf("Message sent to partition %d at offset %d\n", partition, offset)
}

package main
 
import (
    "fmt"
    "log"
    "sync"
 
    "github.com/Shopify/sarama"
)
 
func main() {
    // Konfigurasi consumer
    config := sarama.NewConfig()
    config.Consumer.Return.Errors = true
 
    // Koneksi ke Kafka
    brokers := []string{"localhost:9092"}
    consumer, err := sarama.NewConsumer(brokers, config)
    if err != nil {
        log.Fatalf("Error creating consumer: %v", err)
    }
    defer consumer.Close()
 
    // Subscribe ke topic
    topic := "user-events"
    partitionConsumer, err := consumer.ConsumePartition(topic, 0, sarama.OffsetNewest)
    if err != nil {
        log.Fatalf("Error creating partition consumer: %v", err)
    }
    defer partitionConsumer.Close()
 
    // Loop untuk membaca pesan
    for {
        select {
        case message := <-partitionConsumer.Messages():
            fmt.Printf("Received message: %s\n", string(message.Value))
        case error := <-partitionConsumer.Errors():
            fmt.Printf("Error: %v\n", error)
        }
    }
}

┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│  Producer   │───▶│   Broker    │───▶│  Consumer   │
│             │    │             │    │             │
└─────────────┘    └─────────────┘    └─────────────┘
                          │
                    ┌─────────────┐
                    │    Topic    │
                    │ (Partition) │
                    └─────────────┘

// Contoh konfigurasi partition
topic := "user-events"
partitions := []int{0, 1, 2} // 3 partitions
 
for _, partition := range partitions {
    go func(p int) {
        partitionConsumer, _ := consumer.ConsumePartition(topic, p, sarama.OffsetNewest)
        for message := range partitionConsumer.Messages() {
            processMessage(message)
        }
    }(partition)
}

func sendMessageWithRetry(producer sarama.SyncProducer, message *sarama.ProducerMessage) error {
    maxRetries := 3
    for i := 0; i < maxRetries; i++ {
        _, _, err := producer.SendMessage(message)
        if err == nil {
            return nil
        }
        log.Printf("Retry %d: %v", i+1, err)
        time.Sleep(time.Second * time.Duration(i+1))
    }
    return fmt.Errorf("failed after %d retries", maxRetries)
}

func createProducerPool(brokers []string, poolSize int) ([]sarama.SyncProducer, error) {
    var producers []sarama.SyncProducer
    
    for i := 0; i < poolSize; i++ {
        config := sarama.NewConfig()
        config.Producer.Return.Successes = true
        
        producer, err := sarama.NewSyncProducer(brokers, config)
        if err != nil {
            return nil, err
        }
        producers = append(producers, producer)
    }
    
    return producers, nil
}

func gracefulShutdown(consumer sarama.Consumer, producer sarama.SyncProducer) {
    c := make(chan os.Signal, 1)
    signal.Notify(c, os.Interrupt, syscall.SIGTERM)
    
    <-c
    fmt.Println("Shutting down gracefully...")
    
    consumer.Close()
    producer.Close()
    os.Exit(0)
}

import "github.com/prometheus/client_golang/prometheus"
 
var (
    messagesSent = prometheus.NewCounter(prometheus.CounterOpts{
        Name: "kafka_messages_sent_total",
        Help: "Total number of messages sent to Kafka",
    })
    
    messagesReceived = prometheus.NewCounter(prometheus.CounterOpts{
        Name: "kafka_messages_received_total",
        Help: "Total number of messages received from Kafka",
    })
)

import "go.uber.org/zap"
 
logger, _ := zap.NewProduction()
defer logger.Sync()
 
logger.Info("Message sent to Kafka",
    zap.String("topic", topic),
    zap.Int32("partition", partition),
    zap.Int64("offset", offset),
)

// Order Service
func (s *OrderService) CreateOrder(order Order) error {
    // Simpan order ke database
    err := s.repo.Save(order)
    if err != nil {
        return err
    }
    
    // Kirim event ke Kafka
    event := OrderCreatedEvent{
        OrderID: order.ID,
        UserID:  order.UserID,
        Amount:  order.TotalAmount,
        Time:    time.Now(),
    }
    
    return s.kafkaProducer.Send("order-created", event)
}
 
// Payment Service
func (s *PaymentService) ProcessPayment(event OrderCreatedEvent) error {
    // Proses pembayaran
    payment := Payment{
        OrderID: event.OrderID,
        Amount:  event.Amount,
        Status:  "pending",
    }
    
    return s.repo.Save(payment)
}

func (s *UserService) TrackUserActivity(userID string, action string) error {
    activity := UserActivity{
        UserID:    userID,
        Action:    action,
        Timestamp: time.Now(),
        IP:        getClientIP(),
    }
    
    return s.kafkaProducer.Send("user-activities", activity)
}

func checkKafkaConnection(brokers []string) error {
    config := sarama.NewConfig()
    config.Net.DialTimeout = 5 * time.Second
    
    client, err := sarama.NewClient(brokers, config)
    if err != nil {
        return fmt.Errorf("cannot connect to Kafka: %v", err)
    }
    defer client.Close()
    
    return nil
}

// Untuk memastikan message ordering, gunakan partition key
message := &sarama.ProducerMessage{
    Topic: topic,
    Key:   sarama.StringEncoder(userID), // Same userID = same partition
    Value: sarama.StringEncoder(payload),
}