fix: fix wrong mapping kafkaProducerConfig in streamer

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

Hide file tree

Showing 3 changed files with 12 additions and 7 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 bool
+	var enableAutoOffsetStore = true
 
 	command := cobra.Command{
 		Use: "export",

diff --git a/cmd/importer.go b/cmd/importer.go
@@ -23,6 +23,7 @@ func CreateImportCmd() (*cobra.Command, error) {
 	var sslCertLocation string
 	var sslKeyLocation string
 	var includePartitionAndOffset bool
+	var enableAutoOffsetStore = true
 
 	command := cobra.Command{
 		Use: "import",
@@ -47,7 +48,7 @@ func CreateImportCmd() (*cobra.Command, error) {
 				SSLKeyLocation:            sslKeyLocation,
 				SSLCertLocation:           sslCertLocation,
 				SSLKeyPassword:            sslKeyPassword,
-				EnableAutoOffsetStore:     false,
+				EnableAutoOffsetStore:     enableAutoOffsetStore,
 			}
 			producer, err := kafka_utils.NewProducer(kafkaProducerConfig)
 			if err != nil {
@@ -88,5 +89,6 @@ func CreateImportCmd() (*cobra.Command, error) {
 	command.Flags().StringVar(&sslCertLocation, "ssl-certificate-location", "", "client's certificate location")
 	command.Flags().StringVar(&sslKeyLocation, "ssl-key-location", "", "path to ssl private key")
 	command.Flags().BoolVarP(&includePartitionAndOffset, "include-partition-and-offset", "i", false, "to store partition and offset of kafka message in file")
+	command.Flags().BoolVar(&enableAutoOffsetStore, "enable-auto-offset-store", true, "to store offset in kafka broker")
 	return &command, nil
 }
diff --git a/cmd/streamer.go b/cmd/streamer.go
@@ -28,6 +28,7 @@ func CreateStreamCmd() (*cobra.Command, error) {
 	var toTopic string
 
 	var maxWaitingSecondsForNewMessage int
+	var enableAutoOffsetStore = true
 
 	command := cobra.Command{
 		Use: "stream",
@@ -45,11 +46,12 @@ func CreateStreamCmd() (*cobra.Command, error) {
 				panic(err)
 			}
 			kafkaProducerConfig := kafka_utils.Config{
-				BootstrapServers: fromKafkaServers,
-				SecurityProtocol: fromKafkaSecurityProtocol,
-				SASLMechanism:    fromKafkaSASKMechanism,
-				SASLUsername:     fromKafkaUsername,
-				SASLPassword:     fromKafkaPassword,
+				BootstrapServers:      toKafkaServers,
+				SecurityProtocol:      toKafkaSecurityProtocol,
+				SASLMechanism:         toKafkaSASKMechanism,
+				SASLUsername:          toKafkaUsername,
+				SASLPassword:          toKafkaPassword,
+				EnableAutoOffsetStore: enableAutoOffsetStore,
 			}
 			producer, err := kafka_utils.NewProducer(kafkaProducerConfig)
 			if err != nil {
@@ -105,6 +107,7 @@ func CreateStreamCmd() (*cobra.Command, error) {
 	command.Flags().StringVar(&toKafkaSASKMechanism, "to-kafka-sasl-mechanism", "", "Destination Kafka password")
 	command.Flags().StringVar(&toKafkaSecurityProtocol, "to-kafka-security-protocol", "", "Destination Kafka security protocol")
 	command.Flags().StringVar(&toTopic, "to-topic", "", "Destination topic")
+	command.Flags().BoolVar(&enableAutoOffsetStore, "enable-auto-offset-store", true, "to store offset in kafka broker")
 
 	command.Flags().IntVar(&maxWaitingSecondsForNewMessage, "max-waiting-seconds-for-new-message", 30, "Max waiting seconds for new message, then this process will be marked as finish. Set -1 to wait forever.")