Added ReqID as Kafka's record key

Author: prdpx7

kafka.go

@@ -31,7 +31,7 @@ type InputKafkaConfig struct {
 	Host       string `json:"input-kafka-host"`
 	Topic      string `json:"input-kafka-topic"`
 	UseJSON    bool   `json:"input-kafka-json-format"`
-	Offset     string  `json:"input-kafka-offset"`
+	Offset     string `json:"input-kafka-offset"`
 	SASLConfig SASLKafkaConfig
 }
 

output_kafka.go

@@ -65,6 +65,8 @@ func (o *KafkaOutput) ErrorHandler() {
 // PluginWrite writes a message to this plugin
 func (o *KafkaOutput) PluginWrite(msg *Message) (n int, err error) {
 	var message sarama.StringEncoder
+	meta := payloadMeta(msg.Meta)
+	recordKey := byteutils.SliceToString(meta[1])
 
 	if !o.config.UseJSON {
 		message = sarama.StringEncoder(byteutils.SliceToString(msg.Meta) + byteutils.SliceToString(msg.Data))
@@ -74,10 +76,7 @@ func (o *KafkaOutput) PluginWrite(msg *Message) (n int, err error) {
 		for k, v := range mimeHeader {
 			header[k] = strings.Join(v, ", ")
 		}
-
-		meta := payloadMeta(msg.Meta)
 		req := msg.Data
-
 		kafkaMessage := KafkaMessage{
 			ReqURL:     byteutils.SliceToString(proto.Path(req)),
 			ReqType:    byteutils.SliceToString(meta[0]),
@@ -95,6 +94,9 @@ func (o *KafkaOutput) PluginWrite(msg *Message) (n int, err error) {
 		Topic:     o.config.Topic,
 		Value:     message,
 		Timestamp: time.Now(),
+		//
+		// Making ReqID as Kafka record's key
+		Key: sarama.StringEncoder(recordKey),
 	}
 
 	return len(message), nil

output_kafka_test.go

@@ -23,6 +23,12 @@ func TestOutputKafkaRAW(t *testing.T) {
 
 	resp := <-producer.Successes()
 
+	key, _ := resp.Key.Encode()
+
+	if string(key) != "2" {
+		t.Errorf("Key not properly encoded: %q", key)
+	}
+
 	data, _ := resp.Value.Encode()
 
 	if string(data) != "1 2 3\nGET / HTTP1.1\r\nHeader: 1\r\n\r\n" {
@@ -46,6 +52,12 @@ func TestOutputKafkaJSON(t *testing.T) {
 
 	resp := <-producer.Successes()
 
+	key, _ := resp.Key.Encode()
+
+	if string(key) != "2" {
+		t.Errorf("Key not properly encoded: %q", key)
+	}
+
 	data, _ := resp.Value.Encode()
 
 	if string(data) != `{"Req_URL":"","Req_Type":"1","Req_ID":"2","Req_Ts":"3","Req_Method":"GET"}` {