Fixed kafka client run of brokers and index out of range issues

emitter.go

@@ -61,7 +61,8 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) {
 		if nr > 0 && len(buf) > nr {
 			payload := buf[:nr]
 			meta := payloadMeta(payload)
-			requestID := string(meta[1])
+	//		requestID := string(meta[1])
+			requestID := string(meta[0])
 
 			_maxN := nr
 			if nr > 500 {

input_kafka.go

@@ -3,6 +3,7 @@ package main
 import (
 	"encoding/json"
 	"log"
+	"strings"
 
 	"github.com/Shopify/sarama"
 	"github.com/Shopify/sarama/mocks"
@@ -27,7 +28,8 @@ func NewKafkaInput(address string, config *KafkaConfig) *KafkaInput {
 		con = config.consumer
 	} else {
 		var err error
-		con, err = sarama.NewConsumer([]string{config.host}, c)
+		//con, err = sarama.NewConsumer([]string{config.host}, c)
+		con, err = sarama.NewConsumer(strings.Split(config.host,","), c)
 
 		if err != nil {
 			log.Fatalln("Failed to start Sarama(Kafka) consumer:", err)