DOCSEARCHbyAlgolia
Submit Docs

kafka-go

repository·main·Indexed 27 days ago

https://github.com/segmentio/kafka-go

A Go client library for Apache Kafka providing low-level connection management via the Conn API and high-level abstractions via the Reader and Writer APIs. It supports Go contexts, consumer groups, and is compatible with Kafka versions 0.10.1.0 to 2.7.1. Requires Go 1.15 or later.

Tokens
49.6K
Snippets
111
Records
304
Agent score
94%

What's inside kafka-go

  1. Project Overview and Compatibility

    main

    Overview

    kafka-go provides both low-level and high-level APIs for interacting with Kafka, designed to mirror Go standard library interfaces and support Go context for asynchronous cancellations and timeouts.

    Compatibility

    • Kafka Versions: Tested with versions 0.10.1.0 to 2.7.1. Newer features in later Kafka versions may not be implemented.
    • Go Versions: Requires Go 1.15 or later.

  2. Use Xerial-compatible Snappy framing in Go

    main
    Standard Go snappy implementations are often incompatible with the framing format used by Xerial-based packages. Use go-xerial-snappy to provide Xerial-compatible Snappy framing support in your Go applications. This is specifically required for compatibility with applications like Apache Kafka that utilize the Xerial framing format.

  6. Write to multiple topics with a single Writer

    main

    To write to multiple topics using one Writer, do not define the Topic field in the kafka.Writer configuration. Instead, specify the Topic on each individual kafka.Message.

    Warning: These patterns are mutually exclusive. If you set Writer.Topic, you must not define Message.Topic, and vice versa. Doing so will cause the Writer to return an error due to ambiguity.

    w := &kafka.Writer{
	Addr:     kafka.TCP("localhost:9092", "localhost:9093", "localhost:9094"),
    // NOTE: When Topic is not defined here, each Message must define it instead.
	Balancer: &kafka.LeastBytes{},
}

err := w.WriteMessages(context.Background(),
    // NOTE: Each Message has Topic defined, otherwise an error is returned.
	kafka.Message{
        Topic: "topic-A",
		Key:   []byte("Key-A"),
		Value: []byte("Hello World!"),
	},
	kafka.Message{
        Topic: "topic-B",
		Key:   []byte("Key-B"),
		Value: []byte("One!"),
	},
)

  7. Configure TLS for Connection, Reader, or Writer

    main

    To connect to a Kafka cluster with TLS enabled, you must provide a tls.Config via a Dialer (for Conn and Reader) or via the Transport field (for Writer). If TLS is required by the cluster but not configured in the client, you may encounter io.ErrUnexpectedEOF errors.

    // For Reader
dialer := &kafka.Dialer{
    Timeout: 10 * time.Second,
    TLS:     &tls.Config{...},
}

r := kafka.NewReader(kafka.ReaderConfig{
    Brokers: []string{"localhost:9092"},
    Dialer:  dialer,
})

// For Writer
w := kafka.Writer{
    Addr: kafka.TCP("localhost:9092"),
    Transport: &kafka.Transport{
        TLS: &tls.Config{},
    },
}

  8. Configure SASL Authentication

    main

    Use the SASLMechanism field on a kafka.Dialer or the SASL field on a kafka.Transport to authenticate.

    Supported mechanisms include:

    • Plain: Use plain.Mechanism with Username and Password.
    • SCRAM: Use scram.Mechanism(scram.SHA512, "username", "password").

    It is recommended to create and share kafka.Transport instances across your application to manage connection pools efficiently.

    // Example using SCRAM with a shared Transport for a Writer
mechanism, err := scram.Mechanism(scram.SHA512, "username", "password")
if err != nil {
    panic(err)
}

sharedTransport := &kafka.Transport{
    SASL: mechanism,
}

w := kafka.Writer{
	Addr:      kafka.TCP("localhost:9092"),
	Topic:    "topic-A",
	Balancer: &kafka.Hash{},
	Transport: sharedTransport,
}

  9. Perform explicit offset commits

    main

    If you need manual control over when offsets are committed, use FetchMessage instead of ReadMessage, followed by CommitMessages.

    Note: In consumer groups, committing a message with offset $N$ will also commit all previous offsets for that partition.

    ctx := context.Background()
for {
    m, err := r.FetchMessage(ctx)
    if err != nil {
        break
    }
    fmt.Printf("message at topic/partition/offset %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))
    if err := r.CommitMessages(ctx, m); err != nil {
        log.Fatal("failed to commit messages:", err)
    }
}

  10. Run Bitnami Kafka docker-compose and test cases

    main

    To run the Bitnami Kafka environment and execute the kafka-go test suite, use the following commands:

    Start the Kafka cluster:

    # docker-compose -f ./docker_compose_versions/docker-compose-<kafka_version>.yml up -d

    Run the test cases:

    # go clean -cache; KAFKA_SKIP_NETTEST=1 KAFKA_VERSION=<a.b.c> go test -race -cover ./...
    # docker-compose -f ./docker_compose_versions/docker-compose-<kafka_version>.yml up -d

# go clean -cache; KAFKA_SKIP_NETTEST=1 KAFKA_VERSION=<a.b.c> go test -race -cover ./...;

  11. Create Kafka topics

    main

    Topic creation depends on your Kafka server configuration:

    1. If auto.create.topics.enable='true': Topics are created automatically as a side effect of calling kafka.DialLeader.
    2. If auto.create.topics.enable='false': You must create topics explicitly by finding the controller via conn.Controller() and calling CreateTopics on a connection to that controller.
    // to create topics when auto.create.topics.enable='false'
topic := "my-topic"

conn, err := kafka.Dial("tcp", "localhost:9092")
if err != nil {
    panic(err.Error())
}
defer conn.Close()

controller, err := conn.Controller()
if err != nil {
    panic(err.Error())
}
var controllerConn *kafka.Conn
controllerConn, err = kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
if err != nil {
    panic(err.Error())
}
defer controllerConn.Close()

topicConfigs := []kafka.TopicConfig{
    {
        Topic:             topic,
        NumPartitions:     1,
        ReplicationFactor: 1,
    },
}

err = controllerConn.CreateTopics(topicConfigs...)
if err != nil {
    panic(err.Error())
}

  12. Produce messages using the Writer type

    main

    The Writer type is a high-level API for producing messages to Kafka. It provides automatic retries, reconnections, configurable message distribution (balancers), synchronous or asynchronous writes, and graceful shutdowns via Close().

    To use it, define a Writer with an address (using kafka.TCP), a target topic, and a Balancer to determine how messages are distributed across partitions.

    // make a writer that produces to topic-A, using the least-bytes distribution
w := &kafka.Writer{
	Addr:     kafka.TCP("localhost:9092", "localhost:9093", "localhost:9094"),
	Topic:   "topic-A",
	Balancer: &kafka.LeastBytes{},
}

err := w.WriteMessages(context.Background(),
	kafka.Message{
		Key:   []byte("Key-A"),
		Value: []byte("Hello World!"),
	},
	kafka.Message{
		Key:   []byte("Key-B"),
		Value: []byte("One!"),
	},
	kafka.Message{
		Key:   []byte("Key-C"),
		Value: []byte("Two!"),
	},
)
if err != nil {
    log.Fatal("failed to write messages:", err)
}

if err := w.Close(); err != nil {
    log.Fatal("failed to close writer:", err)
}