main.go

package main

import (
	"fmt"
	"log"
	"sarama"
)

const (
	topic = "aruptest"
)

func main() {
	brokers := []string{"localhost:9092"}

	clientId := "client"
	client, err := sarama.NewClient(clientId, brokers, nil)

	if err != nil {
		panic("Unable to create kafka client " + err.Error())
	}

	producer, err := sarama.NewProducer(client, topic,
		&sarama.ProducerConfig{
			Partitioner: sarama.NewHashPartitioner(),
			RequiredAcks: sarama.NoResponse,
			Compression: sarama.CompressionGZIP})	// Snappy/Gzip compression is not working at the moment

	if err != nil {
		panic(fmt.Sprintf("Unable to create kafka producer: %v", err))
	} else {
		log.Printf("Successfully created kafka producer: %v. Topic: %v", brokers,
			topic)
	}

	var msg string
	for ;; {
		fmt.Scanf("%s", &msg)
		err = producer.SendMessage(nil, sarama.StringEncoder(msg))

		if err == nil {
			log.Printf("Request successfully sent: [%v]", msg)
		} else {
			log.Println("Request couldn't be sent: [%v]", msg)
		}
	}
}