feat: Add KEY support

.gitignore

@@ -14,3 +14,4 @@ gopath/src/golang.org/x/tools
 
 #Final binaries
 /service
+/kafka-console-producer

README.md

@@ -29,6 +29,8 @@ Use environment variables
 - KAFKA_PORT, "9092", Port to connect to input Kafka peers
 - TOPIC, "mytopic", The topic to consume
 - VERBOSE, "false, Set to `true` if you want verbose output
+- KEY, "", The key of produced messages. Default (empty) will produce in every topic
+- FINISH_TIMEOUT", "1", Number of seconds to wait for exit after end of line is received
 
 ## Download
 

gopath/src/service/main.go

@@ -10,16 +10,18 @@ import (
 
 func main() {
 	log.SetOutput(os.Stderr)
-	log.Print("kafka-console-producer v0.5.1")
+	log.Print("kafka-console-producer v0.6.0")
 	var (
 		kafkaService  = getConfig("KAFKA_SERVICE", "kafka") // The DNS name for input Kafka broker service
 		kafkaPort     = getConfig("KAFKA_PORT", "9092")     // Port to connect to input Kafka peers
-		topic         = getConfig("TOPIC", "mytopic")       // The topic to consume
+		topic         = getConfig("TOPIC", "mytopic")       // The topic to produce to
 		verbose       = getConfig("VERBOSE", "false")       // Set to `true` if you want to turn on sarama logging
 		finishTimeout = getConfig("FINISH_TIMEOUT", "1")    // Number of seconds to wait for exit after end of line is received
+		key           = getConfig("KEY", "")                // The key of produced messages. Default (empty) will produce in every topic
 	)
 	messages := make(chan string)
-	saramaProducer := producer(kafkaService, kafkaPort, topic, messages, verbose == "true")
+
+	saramaProducer := producer(kafkaService, kafkaPort, topic, messages, key, verbose == "true")
 	scanner := bufio.NewScanner(os.Stdin)
 	for scanner.Scan() {
 		event := scanner.Text()

gopath/src/service/producer.go

@@ -14,6 +14,7 @@ func producer(kafkaService string,
 	kafkaPort string,
 	topic string,
 	messages chan string,
+	key string,
 	verbose bool) sarama.AsyncProducer {
 
 	var (
@@ -71,11 +72,19 @@ func producer(kafkaService string,
 	}()
 
 	// Read stdin for ever and publish messages
+	var producerKey sarama.Encoder
+	if key == "" {
+		producerKey = nil
+	} else {
+		producerKey = sarama.StringEncoder(key)
+	}
+
 	go func() {
 		for {
 			msg := <-messages
 			producerMessage := &sarama.ProducerMessage{
 				Topic: topic,
+				Key:   producerKey,
 				Value: sarama.StringEncoder(msg),
 			}
 			producer.Input() <- producerMessage

version

@@ -1 +1 @@
-0.5.1
+0.6.0