DOCSEARCHbyAlgolia
Submit Docs

confluent-kafka-go

repository·master·Indexed 26 days ago

https://github.com/confluentinc/confluent-kafka-go

A high-performance Golang client for Apache Kafka that serves as a wrapper around the C library librdkafka. It provides high-level Producer and Consumer APIs supporting Kafka consumer groups and includes prebuilt statically linked librdkafka bundles for MacOSX, Windows, and Linux (glibc and musl).

Tokens
4.8K
Snippets
17
Records
26
Agent score
90%

What's inside confluent-kafka-go

  2. Generate HTML documentation

    master

    To generate one-page HTML documentation, use the mk/doc-gen.py script. This requires the beautifulsoup4 Python package.

    1. Activate your virtual environment.
    2. Install dependencies: pip install beautifulsoup4.
    3. Run the documentation makefile command.
    pip install beautifulsoup4
make -f mk/Makefile docs

  3. Install confluent-kafka-go using Go Modules

    master

    To use confluent-kafka-go in your project, import the kafka package from GitHub. A dependency to the latest stable version will be automatically added to your go.mod file when you build your project.

    Requirements:

    • Go 1.25+
    • librdkafka 2.15.0+
    • CGO_ENABLED must NOT be set to 0.

    Building for Alpine Linux (musl): If you are building for Alpine Linux, you must specify the -tags musl flag.

    import "github.com/confluentinc/confluent-kafka-go/v2/kafka"
    go build ./...
    go build -tags musl ./...

  5. Build kafkatest compatible clients with static librdkafka

    master

    To use the Go client with official Apache Kafka client tests, you must build the verifiable clients using statically linked librdkafka. This ensures the clients are portable and contain all necessary dependencies for the test environment.

    $ mkdir ~/src/kafka/tests/go

$ cd go_verifiable_producer
$ go build -tags static
$ cp go_verifiable_producer ~/src/kafka/tests/go

$ cd go_verifiable_consumer
$ go build -tags static
$ cp go_verifiable_consumer ~/src/kafka/tests/go

  6. Create truly static builds for Linux using musl

    master

    To avoid glibc version errors when compiling for a different target system, use musl to create truly static builds. This requires having musl-gcc installed on your system.

    CC=/path/to/musl-gcc go build --ldflags '-linkmode external -extldflags "-static"' -tags musl

  7. Build confluent-kafka-go with specific librdkafka linking

    master

    You can control how librdkafka is linked during your application build using Go build tags. These tags must be passed to your go build, go get, or go install commands.

    • Default (No tags): Uses the bundled platform-specific static build of librdkafka. This works on macOS and glibc-based Linux distributions (e.g., Ubuntu, CentOS).
    • -tags musl: Required when building for or on musl-based Linux distributions, such as Alpine. This uses the bundled static musl build of librdkafka.
    • -tags dynamic: Links librdkafka dynamically. You must ensure a shared librdkafka library is already installed on the system (via apt-get, yum, or source build).

  8. Configure librdkafka dependencies

    master

    The Go client includes prebuilt librdkafka binaries for several platforms (Mac OSX, glibc-based Linux, musl-based Linux, and Windows amd64). However, you must install librdkafka manually on your system if:

    1. Your platform is not supported by the prebuilt binaries.
    2. You require GSSAPI/Kerberos authentication support.

    If installing manually:

    • Debian/Ubuntu: Install librdkafka-dev.
    • Redhat: Install librdkafka-devel.
    • MacOS: brew install librdkafka pkg-config.
    • Alpine: apk add librdkafka-dev pkgconf.
    • Windows: Static builds are included; manual installation is only needed for GSSAPI/Kerberos.

    Important: After installing librdkafka manually, you must build your Go application using the -tags dynamic flag.

  9. Import a static librdkafka bundle

    master

    To import a new version of the static librdkafka bundle into the repository, follow these steps:

    1. Create the static librdkafka bundle following the instructions in librdkafka's packaging/nuget/README.md.
    2. Run the ./import.sh script provided in this directory, passing the path to your generated bundle.

    The script automates the following:

    • Creates a new branch.
    • Imports the bundle (copies the static library and rdkafka.h header file).
    • Generates a new ../build_..go file for the specific platform and variant.
    • Creates a commit and pushes the branch to GitHub for PR review.

    Note: When merging the resulting Pull Request, do NOT squash or rebase; merge it as is.

    $ ./import.sh ~/path/to/librdkafka-static-bundle-v1.4.0.tgz

  10. Run tests for confluent-kafka-go

    master

    To run the test suites, ensure you have a Kafka cluster available if running integration or benchmark tests. You may need to configure kafka/testconf.json with your bootstrap brokers and topic names.

    Configuration for kafka/testconf.json:

    {
  "Brokers": "<bootstrap-brokers>",
  "Topic": "<test-topic-name>"
}

    Test Commands:

    • Unit tests: go test
    • Benchmark tests: go test -bench .
    • Full test suite (integration): go test ./... (requires a configured cluster)
    • Code coverage:
      go test -coverprofile=coverage.out -bench=.
go tool cover -func=coverage.out

  11. Enable the KIP-848 Next-Gen Consumer Protocol

    master

    To use the next-generation consumer group rebalance protocol (KIP-848), you must use confluent-kafka-go 2.12.0+ and Kafka brokers 4.0.0+. The new protocol is not enabled by default and is controlled by the group.protocol configuration property.

    To enable it, set:

    • group.protocol=consumer
    • (Optional) group.remote.assignor=<assignor> (e.g., uniform or range). If unset, the broker defaults to uniform.

    Note that the Group Leader role is removed in this protocol; assignments are now calculated by the Group Coordinator (broker) and distributed via heartbeats.

  12. Update Rebalance Callbacks for KIP-848 Incremental Protocol

    master

    The KIP-848 protocol is fully incremental. If you use rebalance callbacks, you must switch from full assignment methods to incremental ones.

    Important Changes:

    • Use consumer.IncrementalAssign(e.Partitions) to assign new partitions.
    • Use consumer.IncrementalUnassign(e.Partitions) to revoke partitions.
    • Do not use consumer.Assign() or consumer.Unassign() after subscribing with group.protocol='consumer'.
    • The e.Partitions list in KIP-848 contains only the incremental changes (the specific partitions being added or revoked), unlike the classic protocol which provided the full assignment.

    If you do not call these methods manually, the client will handle the incremental assignment internally.

    // Incremental assignor for KIP-848
func onRebalanceCooperative(consumer *kafka.Consumer, ev kafka.Event) {
    switch e := ev.(type) {
    case kafka.AssignedPartitions:
        fmt.Printf("[KIP-848] Incrementally assigning: %v\n", e.Partitions)
        // Optional: client handles if omitted
        consumer.IncrementalAssign(e.Partitions)

    case kafka.RevokedPartitions:
        fmt.Printf("[KIP-848] Incrementally revoking: %v\n", e.Partitions)
        // Optional: client handles if omitted
        consumer.IncrementalUnassign(e.Partitions)
    }
}