update code

server/node.go

@@ -83,7 +83,7 @@ func (n *Node) Loop(idx int) {
 	postReceipt := func(task *odysseus.TaskContent, result *odysseus.TaskResponse, err error) error {
 		receipt := new(odysseus.TaskReceipt)
 		receipt.TaskUuid = task.TaskUuid
-		receipt.TaskFinishTime = uint64(time.Now().Unix())
+		receipt.TaskFinishTime = uint64(time.Now().UnixNano())
 		receipt.TaskId = task.TaskId
 		receipt.TaskUid = task.TaskUid
 		receipt.TaskWorkload = task.TaskWorkload

utils/kafka.go

 package utils
 
 import (
-	"fmt"
 	"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.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)
 
@@ -24,7 +23,7 @@ func NewKafkaProducer(brokers []string) (sarama.AsyncProducer, error) {
 
 	go func() {
 		for ierr := range producer.Errors() {
-			fmt.Printf("Failed to send log entry to kafka: %v\n", ierr)
+			log.WithError(ierr).Debug("Failed to send log entry to kafka")
 		}
 	}()
 

utils/kafka_test.go

 package utils
 
 import (
+	"fmt"
 	"github.com/google/uuid"
 	basev1 "github.com/odysseus/odysseus-protocol/gen/proto/go/base/v1"
 	"strings"
@@ -24,6 +25,7 @@ func Test(t *testing.T) {
 		if err != nil {
 			t.Fatalf("FireTaskReceipt failed with err:%s", err.Error())
 		}
+		fmt.Println("send task receipt success")
 		time.Sleep(time.Second)
 	}