update kafka go version

Author: wuriyanto48

cmd/kafka-cli/main.go

@@ -30,34 +30,34 @@ func main() {
 	}
 
 	ctx := context.Background()
-	// publisher, err := kafka.NewKafkaGoPublisher(args.Topic, args.Brokers...)
-	// if err != nil {
-	// 	fmt.Println("error : ", err)
-
-	// 	os.Exit(1)
-	// }
-
-	// subsriber, err := kafka.NewKafkaGoSubscriber(args.Topic, args.Brokers...)
-	// if err != nil {
-	// 	fmt.Println("error : ", err)
-
-	// 	os.Exit(1)
-	// }
-
-	publisher, err := kafka.NewSaramaPublisher(args.Brokers...)
+	publisher, err := kafka.NewKafkaGoPublisher(args.Topic, args.Brokers...)
 	if err != nil {
 		fmt.Println("error : ", err)
 
 		os.Exit(1)
 	}
 
-	subsriber, err := kafka.NewSaramaSubscriber(args.Brokers...)
+	subsriber, err := kafka.NewKafkaGoSubscriber(args.Topic, args.Brokers...)
 	if err != nil {
 		fmt.Println("error : ", err)
 
 		os.Exit(1)
 	}
 
+	// publisher, err := kafka.NewSaramaPublisher(args.Brokers...)
+	// if err != nil {
+	// 	fmt.Println("error : ", err)
+
+	// 	os.Exit(1)
+	// }
+
+	// subsriber, err := kafka.NewSaramaSubscriber(args.Brokers...)
+	// if err != nil {
+	// 	fmt.Println("error : ", err)
+
+	// 	os.Exit(1)
+	// }
+
 	runner := kafka.Runner{
 		Publisher:  publisher,
 		Subscriber: subsriber,

go.mod

@@ -1,9 +1,12 @@
 module github.com/musobarlab/kafka-cli
 
+go 1.15
+
 require (
+	github.com/DataDog/zstd v1.4.0 // indirect
 	github.com/Shopify/sarama v1.22.0
 	github.com/pkg/profile v1.3.0 // indirect
-	github.com/segmentio/kafka-go v0.3.4
+	github.com/segmentio/kafka-go v0.4.16
 	github.com/stretchr/objx v0.2.0 // indirect
 	golang.org/x/net v0.0.0-20190424112056-4829fb13d2c6 // indirect
 	golang.org/x/sys v0.0.0-20190428183149-804c0c7841b5 // indirect

go.sum

@@ -14,22 +14,34 @@ github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 h1:YEetp8
 github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21/go.mod h1:+020luEh2TKB4/GOp8oxxtq0Daoen/Cii55CzbTV6DU=
 github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc=
 github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
+github.com/frankban/quicktest v1.11.3/go.mod h1:wRf/ReqHper53s+kmmSZizM8NamnL3IM0I9ntUbOk+k=
 github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/klauspost/compress v1.9.8 h1:VMAMUUOh+gaxKTMk+zqbjsSjsIcUcL/LF4o63i82QyA=
+github.com/klauspost/compress v1.9.8/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/pierrec/lz4 v0.0.0-20190327172049-315a67e90e41 h1:GeinFsrjWz97fAxVUEd748aV0cYL+I6k44gFJTCVvpU=
 github.com/pierrec/lz4 v0.0.0-20190327172049-315a67e90e41/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc=
 github.com/pierrec/lz4 v2.0.5+incompatible h1:2xWsjqPFWcplujydGg4WmhC/6fZqK42wMM8aXeqhl0I=
 github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
+github.com/pierrec/lz4 v2.6.0+incompatible h1:Ix9yFKn1nSPBLFl/yZknTp8TU5G4Ps0JDmguYK6iH1A=
+github.com/pierrec/lz4 v2.6.0+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
 github.com/pkg/profile v1.2.1/go.mod h1:hJw3o1OdXxsrSjjVksARp5W95eeEaEfptyVZyv6JUPA=
 github.com/pkg/profile v1.3.0/go.mod h1:hJw3o1OdXxsrSjjVksARp5W95eeEaEfptyVZyv6JUPA=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a h1:9ZKAASQSHhDYGoxY8uLVpewe1GDZ2vu2Tr/vTdVAkFQ=
 github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/segmentio/kafka-go v0.3.4 h1:Mv9AcnCgU14/cU6Vd0wuRdG1FBO0HzXQLnjBduDLy70=
 github.com/segmentio/kafka-go v0.3.4/go.mod h1:OT5KXBPbaJJTcvokhWR2KFmm0niEx3mnccTwjmLvSi4=
+github.com/segmentio/kafka-go v0.4.16 h1:9dt78ehM9qzAkekA60D6A96RlqDzC3hnYYa8y5Szd+U=
+github.com/segmentio/kafka-go v0.4.16/go.mod h1:19+Eg7KwrNKy/PFhiIthEPkO8k+ac7/ZYXwYM9Df10w=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoHMkEqE=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
 github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
@@ -50,3 +62,6 @@ golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190428024724-550556f78a90/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

kafka_go_publisher.go

@@ -17,7 +17,6 @@ type KafkaGoPublisherImpl struct {
 func NewKafkaGoPublisher(topic string, addresses ...string) (*KafkaGoPublisherImpl, error) {
 	config := ka.WriterConfig{
 		Brokers:          addresses,
-		Topic:            topic,
 		Balancer:         &ka.LeastBytes{},
 		CompressionCodec: snappy.NewCompressionCodec(),
 		BatchTimeout:     5 * time.Millisecond,

kafka_go_subscriber.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	"github.com/segmentio/kafka-go"
 	ka "github.com/segmentio/kafka-go"
@@ -21,12 +22,12 @@ type KafkaGoSubscriberImpl struct {
 //NewKafkaGoSubscriber constructor of KafkaGoSubscriberImpl
 func NewKafkaGoSubscriber(topic string, addresses ...string) (*KafkaGoSubscriberImpl, error) {
 	config := kafka.ReaderConfig{
-		Brokers:  addresses,
-		Topic:    topic,
-		GroupID:  "kafka-cli-group",
-		MinBytes: 10e3, // 10KB
-		MaxBytes: 10e6, // 10MB
-		// CommitInterval: time.Second, // flushes commits to Kafka every second
+		Brokers:        addresses,
+		Topic:          topic,
+		GroupID:        "kafka-cli-group",
+		MinBytes:       10e3,        // 10KB
+		MaxBytes:       10e6,        // 10MB
+		CommitInterval: time.Second, // flushes commits to Kafka every second
 	}
 
 	reader := ka.NewReader(config)
@@ -45,7 +46,7 @@ func (s *KafkaGoSubscriberImpl) Subscribe(ctx context.Context, topics ...string)
 
 	go func() {
 		for {
-			message, err := s.reader.ReadMessage(ctx)
+			message, err := s.reader.FetchMessage(ctx)
 			if err != nil {
 				if err == io.EOF {
 					fmt.Println("read message end ")
@@ -57,7 +58,7 @@ func (s *KafkaGoSubscriberImpl) Subscribe(ctx context.Context, topics ...string)
 
 			fmt.Println()
 			fmt.Printf("\nMessage = %s\nTopic = %s", string(message.Value), message.Topic)
-			//s.reader.CommitMessages(ctx, message)
+			s.reader.CommitMessages(ctx, message)
 		}
 	}()