chore:

consumer.go

@@ -1,7 +1,6 @@
 package kafka
 
 import (
-	"github.com/segmentio/kafka-go/topics"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -56,8 +55,6 @@ func (c *consumer) GetMetricCollectors() []prometheus.Collector {
 func (c *consumer) Consume() {
 	go c.subprocesses.Start()
 
-	c.topicExists()
-
 	c.wg.Add(1)
 	go c.startConsume()
 
@@ -160,26 +157,3 @@ 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")
-	}
-}

test/integration/go.mod

@@ -10,7 +10,7 @@ require (
 )
 
 require (
-	github.com/Trendyol/kafka-cronsumer v1.5.1 // indirect
+	github.com/Trendyol/kafka-cronsumer v1.5.2 // indirect
 	github.com/Trendyol/otel-kafka-konsumer v0.0.7 // indirect
 	github.com/andybalholm/brotli v1.0.5 // indirect
 	github.com/ansrivas/fiberprometheus/v2 v2.6.1 // indirect

test/integration/go.sum

@@ -2,6 +2,7 @@ github.com/Trendyol/kafka-cronsumer v1.5.0 h1:MI0/ncHrlCvOV0Ro4h9avm2izsNprBw4Qf
 github.com/Trendyol/kafka-cronsumer v1.5.0/go.mod h1:VpweJmKY+6dppFhzWOZDbZfxBNuJkSxB12CcuZWBNFU=
 github.com/Trendyol/kafka-cronsumer v1.5.1 h1:L8RLxo8zSGOfVpjtXLUqL3PsJLZdeoFcOvN1yCY/GyQ=
 github.com/Trendyol/kafka-cronsumer v1.5.1/go.mod h1:VpweJmKY+6dppFhzWOZDbZfxBNuJkSxB12CcuZWBNFU=
+github.com/Trendyol/kafka-cronsumer v1.5.2/go.mod h1:VpweJmKY+6dppFhzWOZDbZfxBNuJkSxB12CcuZWBNFU=
 github.com/Trendyol/otel-kafka-konsumer v0.0.7 h1:sT1TE2rgfsdrJWrXKz5j6dPkKJsvP+Tv0Dea4ORqJ+4=
 github.com/Trendyol/otel-kafka-konsumer v0.0.7/go.mod h1:zdCaFclzRCO9fzcjxkHrWOB3I2+uTPrmkq4zczkD1F0=
 github.com/andybalholm/brotli v1.0.5 h1:8uQZIdzKmjc/iuPu7O2ioW48L81FgatrcpfFmiq/cCs=