kafka-tls.go

package main

import (
        "fmt"
        "io/ioutil"
        "log"
        "os"

        "crypto/tls"
        "crypto/x509"

        "github.com/Shopify/sarama"
)

func main() {
        sarama.Logger = log.New(os.Stdout, "[Sarama] ", log.LstdFlags)

        config := sarama.NewConfig()
        config.Producer.RequiredAcks = sarama.WaitForAll
        config.Producer.Retry.Max = 1

        pem, err := ioutil.ReadFile("./ca.pem")
        if err != nil {
                log.Fatal("Could not load certificate!")
        }

        certPool := x509.NewCertPool()
        certPool.AppendCertsFromPEM(pem)

        config.Net.TLS.Enable = true
        config.Net.TLS.Config = &tls.Config{
                RootCAs: certPool,
                //MinVersion:         tls.VersionTLS10,
                //InsecureSkipVerify: true,
        }

        var broker string
        if os.Getenv("BROKER") == "" {
                broker = "...:9093"
        } else {
                broker = os.Getenv("BROKER")
        }

        producer, err := sarama.NewSyncProducer([]string{broker}, config)
        if err != nil {
                // Should not reach here
                log.Fatalf("Could not create producer. %v", err)
        }

        defer func() {
                if err := producer.Close(); err != nil {
                        // Should not reach here
                        panic(err)
                }
        }()

        topic := "hello-logs"
        msg := &sarama.ProducerMessage{
                Topic: topic,
                Value: sarama.StringEncoder(fmt.Sprintf("hello %s", broker)),
        }

        partition, offset, err := producer.SendMessage(msg)
        if err != nil {
                log.Fatalf("Could not send message. %v", err)
        }

        fmt.Printf("Message is stored in topic(%s)/partition(%d)/offset(%d)\n", topic, partition, offset)
}