fix bug for kafka

utils/kafka.go

@@ -4,17 +4,16 @@ import (
 	"github.com/IBM/sarama"
 	"github.com/gogo/protobuf/proto"
 	odysseus "github.com/odysseus/odysseus-protocol/gen/proto/go/base/v1"
-	"time"
+	log "github.com/sirupsen/logrus"
 )
 
 // NewKafkaProducer Create a new KafkaProducer.
 func NewKafkaProducer(brokers []string) (sarama.AsyncProducer, error) {
 	kafkaConfig := sarama.NewConfig()
-	kafkaConfig.Producer.RequiredAcks = sarama.WaitForLocal     // Only wait for the leader to ack
-	kafkaConfig.Producer.Compression = sarama.CompressionSnappy // Compress messages
-	kafkaConfig.Producer.Return.Errors = false
-	kafkaConfig.Producer.Flush.Frequency = 500 * time.Millisecond // Flush batches every 500ms
-	kafkaConfig.Net.ResolveCanonicalBootstrapServers = false
+	//kafkaConfig.Producer.RequiredAcks = sarama.WaitForLocal       // Only wait for the leader to ack
+	//kafkaConfig.Producer.Compression = sarama.CompressionSnappy   // Compress messages
+	//kafkaConfig.Producer.Flush.Frequency = 500 * time.Millisecond // Flush batches every 500ms
+	//kafkaConfig.Net.ResolveCanonicalBootstrapServers = false
 
 	producer, err := sarama.NewAsyncProducer(brokers, kafkaConfig)
 
@@ -23,8 +22,8 @@ func NewKafkaProducer(brokers []string) (sarama.AsyncProducer, error) {
 	}
 
 	go func() {
-		for _ = range producer.Errors() {
-			//log.Printf("Failed to send log entry to kafka: %v\n", err)
+		for ierr := range producer.Errors() {
+			log.WithError(ierr).Debug("Failed to send log entry to kafka")
 		}
 	}()
 
@@ -33,8 +32,8 @@ func NewKafkaProducer(brokers []string) (sarama.AsyncProducer, error) {
 
 func FireTaskReceipt(producer sarama.AsyncProducer, receipt *odysseus.TaskReceipt, topic string) error {
 	// Check time for partition key
-	var partitionKey sarama.StringEncoder
-	partitionKey = sarama.StringEncoder(time.Now().Format("2006-01-02"))
+	//var partitionKey sarama.StringEncoder
+	//partitionKey = sarama.StringEncoder(time.Now().Format("2006-01-02"))
 
 	b, err := proto.Marshal(receipt)
 	if err != nil {
@@ -43,7 +42,7 @@ func FireTaskReceipt(producer sarama.AsyncProducer, receipt *odysseus.TaskReceip
 	value := sarama.ByteEncoder(b)
 
 	producer.Input() <- &sarama.ProducerMessage{
-		Key:   partitionKey,
+		//Key:   partitionKey,
 		Topic: topic,
 		Value: value,
 	}

utils/kafka_test.go

+package utils
+
+import (
+	"fmt"
+	"github.com/google/uuid"
+	basev1 "github.com/odysseus/odysseus-protocol/gen/proto/go/base/v1"
+	"strings"
+	"testing"
+	"time"
+)
+
+func Test(t *testing.T) {
+	broker := "192.168.1.220:9092"
+	brokers := strings.Split(broker, ";")
+	producer, err := NewKafkaProducer(brokers)
+	if err != nil {
+		t.Fatalf("NewKafkaProducer failed with err:%s", err.Error())
+	}
+
+	for i := 0; i < 100; i++ {
+		taskReceipt := &basev1.TaskReceipt{
+			TaskUid: uuid.NewString(),
+		}
+		err = FireTaskReceipt(producer, taskReceipt, "taskreceipt")
+		if err != nil {
+			t.Fatalf("FireTaskReceipt failed with err:%s", err.Error())
+		}
+		fmt.Println("send task receipt success")
+		time.Sleep(time.Second)
+	}
+
+}