Remove default kafka bootstrap consumer behavior

only create a kafka consumer if explicitly set
banzaicloud · Oct 6, 2022 · 7fa55b8 · 7fa55b8
1 parent ed7192b
commit 7fa55b8
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 34 deletions.
diff --git a/cmd/allspark/main.go b/cmd/allspark/main.go
@@ -184,28 +184,30 @@ func main() {
 		srv.Run()
 	}()
 
-	// Kafka server
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		consumer := kafka.NewConsumer(configuration.KafkaServer.BootstrapServer, configuration.KafkaServer.Topic, configuration.KafkaServer.ConsumerGroup, logger)
-		srv := server.New(consumer, logger, errorHandler)
-		if wl != nil {
-			srv.SetWorkload(wl)
-		}
-
-		kafkaRequests, err := request.CreateRequestsFromStringSlice(viper.GetStringSlice("kafkaRequests"), logger.WithField("server", "kafka"))
-		if err != nil {
-			panic(err)
-		}
-		if len(kafkaRequests) == 0 {
-			kafkaRequests = requests
-		}
-
-		srv.SetRequests(kafkaRequests)
-		srv.SetSQLClient(sqlClient)
-		srv.Run()
-	}()
+	if configuration.KafkaServer.BootstrapServer != "" && configuration.KafkaServer.Topic != "" && configuration.KafkaServer.ConsumerGroup != "" {
+		// Kafka server
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			consumer := kafka.NewConsumer(configuration.KafkaServer.BootstrapServer, configuration.KafkaServer.Topic, configuration.KafkaServer.ConsumerGroup, logger)
+			srv := server.New(consumer, logger, errorHandler)
+			if wl != nil {
+				srv.SetWorkload(wl)
+			}
+
+			kafkaRequests, err := request.CreateRequestsFromStringSlice(viper.GetStringSlice("kafkaRequests"), logger.WithField("server", "kafka"))
+			if err != nil {
+				panic(err)
+			}
+			if len(kafkaRequests) == 0 {
+				kafkaRequests = requests
+			}
+
+			srv.SetRequests(kafkaRequests)
+			srv.SetSQLClient(sqlClient)
+			srv.Run()
+		}()
+	}
 
 	wg.Wait()
 }
diff --git a/internal/kafka/consumer.go b/internal/kafka/consumer.go
@@ -74,17 +74,5 @@ func (c *Consumer) Close() error {
 }
 
 func (c *Consumer) Validate() (*Consumer, error) {
-	if c.BootstrapServer == "" {
-		c.BootstrapServer = "kafka-all-broker.kafka:29092"
-	}
-
-	if c.ConsumerGroup == "" {
-		c.ConsumerGroup = "allspark-consumer-group"
-	}
-
-	if c.Topic == "" {
-		c.Topic = "example-topic"
-	}
-
 	return c, nil
 }