feat: added topic exists check (#59)

consumer.go

@@ -1,6 +1,7 @@
 package kafka
 
 import (
+	"github.com/segmentio/kafka-go/topics"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -55,6 +56,8 @@ func (c *consumer) GetMetricCollectors() []prometheus.Collector {
 func (c *consumer) Consume() {
 	go c.subprocesses.Start()
 
+	c.topicExists()
+
 	c.wg.Add(1)
 	go c.startConsume()
 
@@ -157,3 +160,26 @@ func (c *consumer) process(message *Message) {
 		c.metric.TotalProcessedMessagesCounter++
 	}
 }
+
+func (c *consumer) topicExists() {
+	list, err := topics.List(c.context, &kafka.Client{
+		Addr:    kafka.TCP(c.brokers...),
+		Timeout: 3 * time.Second,
+	})
+
+	if err != nil {
+		c.logger.Errorf("Topic Exists Function Err %s", err.Error())
+	}
+
+	var exist bool
+	for i := range list {
+		if list[i].Name == c.topic {
+			exist = true
+			break
+		}
+	}
+
+	if !exist {
+		c.logger.Errorf("Topic doesn't exist")
+	}
+}

consumer_base.go

@@ -67,6 +67,8 @@ type base struct {
 	incomingMessageStream     chan *IncomingMessage
 	singleConsumingStream     chan *Message
 	batchConsumingStream      chan []*Message
+	brokers                   []string
+	topic                     string
 	retryTopic                string
 	subprocesses              subprocesses
 	wg                        sync.WaitGroup
@@ -116,6 +118,8 @@ func newBase(cfg *ConsumerConfig, messageChSize int) (*base, error) {
 		consumerState:             stateRunning,
 		skipMessageByHeaderFn:     cfg.SkipMessageByHeaderFn,
 		metricPrefix:              cfg.MetricPrefix,
+		brokers:                   cfg.Reader.Brokers,
+		topic:                     cfg.Reader.Topic,
 	}
 
 	if cfg.DistributedTracingEnabled {