Connect to your cluster using the Kafka Go client

​

Before you begin

​

Steps

1. Install the Kafka Go client.
   go get -u github.com/confluentinc/confluent-kafka-go/v2/kafka
   
2. Build a Go application to produce and consume messages.
   package main
   
   import (
     "fmt"
     "github.com/confluentinc/confluent-kafka-go/v2/kafka"
     "time"
   )
   
   func main() {
     // Step 1: replace with your configurations
     serverUrl := "SERVER-URL"
     jwtToken := "API-KEY"
     topicName := "test-go-topic"
     namespace := "public/default"
     password := "token:" + jwtToken
   
     // Step 2: create a producer to send messages
     producer, err := kafka.NewProducer(&kafka.ConfigMap{
       "bootstrap.servers": serverUrl,
       "security.protocol": "SASL_SSL",
       "sasl.mechanism":    "PLAIN",
       "sasl.username":     namespace,
       "sasl.password":     password,
     })
     if err != nil {
       panic(err)
     }
     defer producer.Close()
   
     err = producer.Produce(&kafka.Message{
       TopicPartition: kafka.TopicPartition{Topic: &topicName, Partition: kafka.PartitionAny},
       Value:          []byte("hello world"),
     }, nil)
   
     if err != nil {
       panic(err)
     }
     producer.Flush(1000)
   
     // wait for delivery report
     e := <-producer.Events()
   
     message := e.(*kafka.Message)
     if message.TopicPartition.Error != nil {
       fmt.Printf("failed to deliver message: %v\n",
         message.TopicPartition)
     } else {
       fmt.Printf("delivered to topic %s [%d] at offset %v\n",
         *message.TopicPartition.Topic,
         message.TopicPartition.Partition,
         message.TopicPartition.Offset)
     }
   
     // Step 3: create a consumer to read messages
     consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
       "bootstrap.servers":  serverUrl,
       "security.protocol":  "SASL_SSL",
       "sasl.mechanisms":    "PLAIN",
       "sasl.username":      namespace,
       "sasl.password":      password,
       "session.timeout.ms": 6000,
       "group.id":           "my-group",
       "auto.offset.reset":  "earliest",
       "isolation.level":    "read_uncommitted", // Note: Ursa does not support read_committed for now
     })
   
     if err != nil {
       panic(fmt.Sprintf("Failed to create consumer: %s", err))
     }
     defer consumer.Close()
   
     topics := []string{topicName}
     err = consumer.SubscribeTopics(topics, nil)
     if err != nil {
       panic(fmt.Sprintf("Failed to subscribe topics: %s", err))
     }
   
     // read one message then exit
     for {
       fmt.Println("polling...")
       message, err = consumer.ReadMessage(1 * time.Second)
       if err == nil {
         fmt.Printf("consumed from topic %s [%d] at offset %v: %+v",
           *message.TopicPartition.Topic,
           message.TopicPartition.Partition, message.TopicPartition.Offset, string(message.Value))
         break
       }
     }
   }
   
   • SERVER-URL: the Kafka service URL of your StreamNative cluster.
   • API-KEY: an API key of your service account.
3. Run the Go application and you should see the following output:
   delivered to topic test-go-topic [0] at offset 29
   polling...
   polling...
   polling...
   polling...
   polling...
   polling...
   polling...
   consumed from topic test-go-topic [0] at offset 15: hello world