Example for integration with confluentinc/confluent-kafka-go

[bitdean]

Describe the feature

Since confluentinc/confluent-kafka-go is most generic and it uses underlaying librdkafka natively. Would it be possible that someone would provide example how to integrate with it?

Use Case

• application using confluentinc/confluent-kafka-go would like to use AWS MSK IAM SASL signer for connecting to AWS MSK cluster via IAM role
• provide example how to connect when using confluentinc/confluent-kafka-go

Proposed Solution

No response

Other Information

No response

Acknowledgements

• [X] I may be able to implement this feature request
• [ ] This feature might incur a breaking change

aws-msk-iam-sasl-signer-go Module Versions Used

1.0.0

Go version used

1.21

[nboergerotto]

I got it running with the following code:

package main

import (
    "context"
    "fmt"
    "github.com/aws/aws-msk-iam-sasl-signer-go/signer"
    "github.com/confluentinc/confluent-kafka-go/kafka"
    "time"
)

func main() {
    bearerToken := createToken()

    producer := createProducerWithToken(bearerToken)
    defer producer.Close()

    produceMessagesToTopic(producer)

    flush(producer)
}

func createToken() kafka.OAuthBearerToken {
    token, tokenExpirationTime, err := signer.GenerateAuthToken(context.TODO(), "YOUR-AWS-REGION")
    if err != nil {
        panic(err)
    }
    seconds := tokenExpirationTime / 1000
    nanoseconds := (tokenExpirationTime % 1000) * 1000000
    bearerToken := kafka.OAuthBearerToken{
        TokenValue: token,
        Expiration: time.Unix(seconds, nanoseconds),
    }

    return bearerToken
}

func createProducerWithToken(bearerToken kafka.OAuthBearerToken) *kafka.Producer {
    kafkaConfig := &kafka.ConfigMap{
        "bootstrap.servers": "YOUR-BROKER-URL",
        "client.id":         "golang-test-producer",
        "security.protocol": "sasl_ssl",
        "sasl.mechanism":    "OAUTHBEARER",
    }
    producer, err := kafka.NewProducer(kafkaConfig)
    if err != nil {
        panic(err)
    }

    err = producer.SetOAuthBearerToken(bearerToken)
    if err != nil {
        panic(err)
    }
    return producer
}

func produceMessagesToTopic(producer *kafka.Producer) {
    topic := "YOUR-TOPIC-NAME"
    for _, word := range []string{"Welcome", "to", "the", "Confluent", "Kafka", "Golang", "client"} {
        err := producer.Produce(&kafka.Message{
            TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
            Value:          []byte(word),
        }, nil)
        fmt.Printf("Successfully produced word %s\n", word)
        if err != nil {
            fmt.Printf("Failed to produce: %v\n", err)
        }
    }
}

func flush(producer *kafka.Producer) {
    producer.Flush(15 * 1000)
}

You have to replace the following strings in my code: YOUR-BROKER-URL, YOUR-AWS-REGION, YOUR-TOPIC-NAME

For running the code in an ECS-Task, I'm using the following Dockerfile:

# syntax=docker/dockerfile:1

FROM --platform=linux/amd64 golang:1.22 as builder
WORKDIR /app
COPY . .
RUN go mod download
RUN GOOS=linux go build -o /kafka-producer-golang-confluent

FROM --platform=linux/amd64 debian:bookworm-slim as runner
COPY --from=builder /kafka-producer-golang-confluent /kafka-producer-golang-confluent
RUN apt-get update && apt-get install -y ca-certificates && update-ca-certificates
CMD ["/kafka-producer-golang-confluent"]

For me, in order to get it running, it was very important to install ca-certificates ("RUN apt-get update && apt-get install -y ca-certificates && update-ca-certificates"), since debian:bookworm-slim doesn't have them by default. Depending on your docker base image, you may not need to do this though.

This exact code is successfully producing events on my kafka-topic, once the client-policies are correctly set as stated in the AWS documentation.

[dixon14]

Hi, I encountered a problem when connecting consumer to the brokers using the IAM signer and the confluent kafka go client.

Setup the code similar as the above example. When I tried to ReadMessage from a topic, it return KafkaErrTimeOut. Looked at the debug logs - it shows no brokers is available for coordination but my brokers are healthy. If anyone can help on this?

type MessageProcessor interface {
    Process(msg *confluentkafka.Message) error
}

type Consumer struct {
    config       config.Config
    consumer     *confluentkafka.Consumer
    processorMap map[string]MessageProcessor
}

func New(
    ctx context.Context,
    config config.Config,
    processorMap map[string]MessageProcessor,
) *Consumer {
    configMap := &confluentkafka.ConfigMap{
        "bootstrap.servers":  config.KafkaBootstrapServers,
        "group.id":           config.KafkaGroupID,
        "security.protocol":  "SASL_SSL",
        "sasl.mechanism":     "OAUTHBEARER",
        "auto.offset.reset":  "latest",
        "enable.auto.commit": true,
        "session.timeout.ms": 60000,
        // "heartbeat.interval.ms": 60000,
        // "max.poll.interval.ms":  300000,
        // "debug":                 "all",
    }

    consumer, err := confluentkafka.NewConsumer(configMap)
    if err != nil {
        return nil
    }

    token, err := createToken(ctx)
    if err != nil {
        return nil
    }

    if err := consumer.SetOAuthBearerToken(token); err != nil {
        return nil
    }

    var topics []string
    for topic := range processorMap {
        topics = append(topics, topic)
    }

    // Subscribe to multiple topics
    if err = consumer.SubscribeTopics(topics, nil); err != nil {
        log.Err(err).Msg("Unable to subscribe to topics.")
        return nil
    }

    return &Consumer{
        config:       config,
        consumer:     consumer,
        processorMap: processorMap,
    }
}

// Start indicates that Kafka consumer will start polling messages
// from the subscribed topics in the kafka bootstrap servers
func (c *Consumer) Start() error {
    for {
        msg, err := c.consumer.ReadMessage(100 * time.Millisecond)
        if err != nil {
            return err
        }
        // fmt.Printf("Consumed event from topic %s: key = %-10s value = %s\n",
        //  *msg.TopicPartition.Topic, string(msg.Key), string(msg.Value))
        topic := *msg.TopicPartition.Topic
        processor := c.processorMap[topic]

        if err := processor.Process(msg); err != nil {
            log.Err(err).Msgf("Error in processing message from topic %s", topic)
            return err
        }
    }
}

func createToken(ctx context.Context) (confluentkafka.OAuthBearerToken, error) {
    token, tokenExpirationTimeMs, err := signer.GenerateAuthToken(ctx, "us-east-1")
    if err != nil {
        return confluentkafka.OAuthBearerToken{}, err
    }

    seconds := tokenExpirationTimeMs / 1000
    nanoseconds := (tokenExpirationTimeMs % 1000) * 1000000
    bearerToken := confluentkafka.OAuthBearerToken{
        TokenValue: token,
        Expiration: time.Unix(seconds, nanoseconds),
    }

    return bearerToken, nil
}