Write Data to a Kafka Topic in Go programming language Example
In this tutorial, you will learn how to write data to a Kafka topic in Go programming language.
To send data to a Kafka topic, we will use Confluent's Kafka Golang library. It provides a high level Producer and Consumer with support for balanced consumer groups of Apache Kafka 0.9 and above.
An application that writes data to a Kafka topic is called a Kafka Producer Application.
You can also learn how to read data from a Kafka topic in Go here.
Follow the steps below to install Confluent Kafka library and create a producer application to send data to a Kafka topic in Go:
Installing Confluent Kafka Golang Library
Install the latest version of Confluent Kafka Go Library:
go get -u gopkg.in/confluentinc/confluent-kafka-go.v1/kafka
If you get this error C compiler "gcc" not found: exec: "gcc": executable file not found in %PATH% while installing confluent-kafka-go library on Windows, then install TDM-GCC. gcc provides a C compiler.
Creating the Kafka Producer
Producer code to send data to a Kafka topic:
package main
import (
"fmt"
"encoding/json"
"github.com/confluentinc/confluent-kafka-go/kafka"
)
func produce(messages string, topicName string) {
fmt.Printf("Starting producer...")
p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "localhost:9092"})
if err != nil {
panic(err)
}
defer p.Close()
// Delivery report handler for produced messages
go func() {
for e := range p.Events() {
switch ev := e.(type) {
case *kafka.Message:
if ev.TopicPartition.Error != nil {
fmt.Printf("Delivery failed: %v\n", ev.TopicPartition)
} else {
fmt.Printf("Delivered message to %v\n", ev.TopicPartition)
}
}
}
}()
// Produce messages to topic (asynchronously)
topic := topicName
data := []byte(messages)
p.Produce(&kafka.Message{
TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
Value: data,
}, nil)
// Wait for message deliveries before shutting down
p.Flush(15 * 1000)
}
//Send data to Kafka topic
type Data struct {
Id int
Message string
Source string
Destination string
}
func main() {
topicName := "myTopic"
data := []Data{
{Id: 2, Message: "World", Source: "1", Destination: "A"},
{Id: 2, Message: "Earth", Source: "1", Destination: "B"},
{Id: 2, Message: "Planets", Source: "2", Destination: "C"},
}
stringJson, _ := json.Marshal(data)
fmt.Println(string(stringJson))
produce(string(stringJson), topicName)
}
Kafka Producer application is ready. Next, run the Kafka Services and create a topic myTopic on your local machine.
If you do not have Apache Kafka server already setup on your local machine then learn how to do it here for Windows and here for Mac/Ubuntu/Linux.