feat: Verify topic by default (#84)

Author: kutay-o

README.md

@@ -141,7 +141,6 @@ func StartAPI(cfg kafka.Config, metricCollectors ...prometheus.Collector) {
 | `consumer.groupId`               | Exception consumer group id                                                                                                                                                                                                                                                                                                                                                                                       |                             | exception-consumer-group                  |
 | `consumer.maxRetry`              | Maximum retry value for attempting to retry a message                                                                                                                                                                                                                                                                                                                                                             | 3                           |                                           |
 | `consumer.concurrency`           | Number of goroutines used at listeners                                                                                                                                                                                                                                                                                                                                                                            | 1                           |                                           |
-| `consumer.verifyTopicOnStartup`  | it checks existence of the given retry topic on the kafka cluster.                                                                                                                                                                                                                                                                                                                                                | false                       |                                           |
 | `consumer.minBytes`              | [see doc](https://pkg.go.dev/github.com/segmentio/[email protected]#ReaderConfig.MinBytes)                                                                                                                                                                                                                                                                                                                         | 1                           |                                           |
 | `consumer.maxBytes`              | [see doc](https://pkg.go.dev/github.com/segmentio/[email protected]#ReaderConfig.MaxBytes)                                                                                                                                                                                                                                                                                                                         | 1 MB                        |                                           |
 | `consumer.maxWait`               | [see doc](https://pkg.go.dev/github.com/segmentio/[email protected]#ReaderConfig.MaxWait)                                                                                                                                                                                                                                                                                                                          | 10s                         |                                           |

cronsumer.go

@@ -13,21 +13,21 @@ import (
 // ConsumeFn describes how to consume messages from specified topic.
 func New(cfg *kafka.Config, c kafka.ConsumeFn) kafka.Cronsumer {
 	cfg.Logger = logger.New(cfg.LogLevel)
+	verifyTopicOnStartup(cfg)
+	cfg.Logger.Infof("Topic [%s] verified successfully!", cfg.Consumer.Topic)
+	return internal.NewCronsumer(cfg, c)
+}
 
-	if cfg.Consumer.VerifyTopicOnStartup {
-		kclient, err := internal.NewKafkaClient(cfg)
-		if err != nil {
-			panic("panic when initializing kafka client for verify topic error: " + err.Error())
-		}
-		exist, err := internal.VerifyTopics(kclient, cfg.Consumer.Topic)
-		if err != nil {
-			panic("panic " + err.Error())
-		}
-		if !exist {
-			panic("topic: " + cfg.Consumer.Topic + " does not exist, please check cluster authority etc.")
-		}
-		cfg.Logger.Infof("Topic [%s] verified successfully!", cfg.Consumer.Topic)
+func verifyTopicOnStartup(cfg *kafka.Config) {
+	kclient, err := internal.NewKafkaClient(cfg)
+	if err != nil {
+		panic("panic when initializing kafka client for verify topic error: " + err.Error())
+	}
+	exist, err := internal.VerifyTopics(kclient, cfg.Consumer.Topic)
+	if err != nil {
+		panic("panic " + err.Error())
+	}
+	if !exist {
+		panic("topic: " + cfg.Consumer.Topic + " does not exist, please check cluster authority etc.")
 	}
-
-	return internal.NewCronsumer(cfg, c)
 }

pkg/kafka/config.go

@@ -78,7 +78,6 @@ type ConsumerConfig struct {
 	Cron                  string                   `yaml:"cron"`
 	BackOffStrategy       BackoffStrategyInterface `yaml:"backOffStrategy"`
 	SkipMessageByHeaderFn SkipMessageByHeaderFn    `yaml:"skipMessageByHeaderFn"`
-	VerifyTopicOnStartup  bool                     `yaml:"verifyTopicOnStartup"`
 	QueueCapacity         int                      `yaml:"queueCapacity"`
 }