Disable singleton kafka client (#148)

Co-authored-by: burak.gundogdu <[email protected]>
Trendyol · Oct 31, 2024 · 0c34509 · 0c34509
1 parent 541cca7
commit 0c34509
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 26 deletions.
diff --git a/producer_test.go b/producer_test.go
@@ -84,7 +84,7 @@ func Test_producer_Close_Successfully(t *testing.T) {
 
 type mockWriter struct{}
 
-func (m *mockWriter) WriteMessages(_ context.Context, msg ...kafka.Message) error {
+func (m *mockWriter) WriteMessages(_ context.Context, _ ...kafka.Message) error {
 	return nil
 }
 

diff --git a/verify_topic.go b/verify_topic.go
@@ -3,7 +3,6 @@ package kafka
 import (
 	"context"
 	"fmt"
-	"sync"
 
 	"github.com/segmentio/kafka-go"
 )
@@ -17,39 +16,31 @@ type client struct {
 	*kafka.Client
 }
 
-var (
-	kafkaClientInstance *client
-	kafkaClientOnce     sync.Once
-)
-
 func newKafkaClient(cfg *ConsumerConfig) (kafkaClient, error) {
 	var err error
-	kafkaClientOnce.Do(func() {
-		kc := &client{
-			Client: &kafka.Client{
-				Addr: kafka.TCP(cfg.Reader.Brokers...),
-			},
-		}
+	kc := &client{
+		Client: &kafka.Client{
+			Addr: kafka.TCP(cfg.Reader.Brokers...),
+		},
+	}
 
-		transport := &Transport{
-			Transport: &kafka.Transport{
-				MetadataTopics: cfg.getTopics(),
-			},
-		}
-		if err = fillLayer(transport, cfg.SASL, cfg.TLS); err != nil {
-			err = fmt.Errorf("error when initializing kafka client for verify topic purpose %w", err)
-			return
-		}
+	transport := &Transport{
+		Transport: &kafka.Transport{
+			MetadataTopics: cfg.getTopics(),
+		},
+	}
+	if err = fillLayer(transport, cfg.SASL, cfg.TLS); err != nil {
+		err = fmt.Errorf("error when initializing kafka client for verify topic purpose %w", err)
+		return nil, err
+	}
 
-		kc.Transport = transport
-		kafkaClientInstance = kc
-	})
+	kc.Transport = transport
 
 	if err != nil {
 		return nil, err
 	}
 
-	return kafkaClientInstance, nil
+	return kc, nil
 }
 
 func (k *client) GetClient() *kafka.Client {