fix: set enableAutoOffsetStore for consumer

huantt · Dec 4, 2023 · 0067df4 · 0067df4
1 parent a60f330
commit 0067df4
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 17 deletions.
diff --git a/cmd/exporter.go b/cmd/exporter.go
@@ -39,7 +39,7 @@ func CreateExportCommand() (*cobra.Command, error) {
 	var sslKeyPassword string
 	var sslCertLocation string
 	var sslKeyLocation string
-	var enableAutoOffsetStore = true
+	var enableAutoOffsetStore bool
 
 	command := cobra.Command{
 		Use: "export",

diff --git a/cmd/streamer.go b/cmd/streamer.go
@@ -34,33 +34,33 @@ func CreateStreamCmd() (*cobra.Command, error) {
 		Use: "stream",
 		Run: func(cmd *cobra.Command, args []string) {
 			kafkaConsumerConfig := kafka_utils.Config{
-				BootstrapServers: fromKafkaServers,
-				SecurityProtocol: fromKafkaSecurityProtocol,
-				SASLMechanism:    fromKafkaSASKMechanism,
-				SASLUsername:     fromKafkaUsername,
-				SASLPassword:     fromKafkaPassword,
-				GroupId:          fromKafkaGroupID,
+				BootstrapServers:      fromKafkaServers,
+				SecurityProtocol:      fromKafkaSecurityProtocol,
+				SASLMechanism:         fromKafkaSASKMechanism,
+				SASLUsername:          fromKafkaUsername,
+				SASLPassword:          fromKafkaPassword,
+				GroupId:               fromKafkaGroupID,
+				EnableAutoOffsetStore: enableAutoOffsetStore,
 			}
 			consumer, err := kafka_utils.NewConsumer(kafkaConsumerConfig)
 			if err != nil {
 				panic(err)
 			}
 			kafkaProducerConfig := kafka_utils.Config{
-				BootstrapServers:      toKafkaServers,
-				SecurityProtocol:      toKafkaSecurityProtocol,
-				SASLMechanism:         toKafkaSASKMechanism,
-				SASLUsername:          toKafkaUsername,
-				SASLPassword:          toKafkaPassword,
-				EnableAutoOffsetStore: enableAutoOffsetStore,
-			}
-			producer, err := kafka_utils.NewProducer(kafkaProducerConfig)
-			if err != nil {
-				panic(err)
+				BootstrapServers: toKafkaServers,
+				SecurityProtocol: toKafkaSecurityProtocol,
+				SASLMechanism:    toKafkaSASKMechanism,
+				SASLUsername:     toKafkaUsername,
+				SASLPassword:     toKafkaPassword,
 			}
 			queueBufferingMaxMessages := kafka_utils.DefaultQueueBufferingMaxMessages
 			if kafkaProducerConfig.QueueBufferingMaxMessages > 0 {
 				queueBufferingMaxMessages = kafkaProducerConfig.QueueBufferingMaxMessages
 			}
+			producer, err := kafka_utils.NewProducer(kafkaProducerConfig)
+			if err != nil {
+				panic(err)
+			}
 			deliveryChan := make(chan kafka.Event, queueBufferingMaxMessages)
 			go func() { // Tricky: kafka require specific deliveryChan to use Flush function
 				for e := range deliveryChan {