Repository: gofr-dev/gofr

Issue:
How to configure Gofr for Kafka with multi brokers cluster
Hello!

In production, several times is necessary to use kafka cluster, [IP1:port, IP2:port, IP3:port, ...]. In this moment, the Gofr version only receive one broker (en several examples localhost:9092). Please, increment this characteristic in new version

for example, configs/.env:

PUBSUB_BROKERS=10.100.10.11:9092,10.100.10.13:9092, 10.100.10.15:9092

brokers := strings.Split(PUBSUB_BROKERS, ",")
kafka.NewReader(kafka.ReaderConfig{
     brokers:  brokers,
     ...
}

Thanks! 



Discussion:

Solution patch:
diff --git a/docs/advanced-guide/using-publisher-subscriber/page.md b/docs/advanced-guide/using-publisher-subscriber/page.md
index 7db50e5c..d70ccc2e 100644
--- a/docs/advanced-guide/using-publisher-subscriber/page.md
+++ b/docs/advanced-guide/using-publisher-subscriber/page.md
@@ -47,10 +47,10 @@ that are specific for the type of message broker user wants to use.
 ---
 
 - `PUBSUB_BROKER`
-- Address to connect to kafka broker.
+- Address to connect to kafka broker. Multiple brokers can be added as comma seperated values.
 - `+`
 -
-- `localhost:9092`
+- `localhost:9092` or `localhost:8087,localhost:8088,localhost:8089`
 - Not empty string
 
 ---
diff --git a/pkg/gofr/container/container.go b/pkg/gofr/container/container.go
index fb3cc5f7..5077aeb2 100644
--- a/pkg/gofr/container/container.go
+++ b/pkg/gofr/container/container.go
@@ -138,8 +138,10 @@ func (c *Container) Create(conf config.Config) {
 				InsecureSkipVerify: conf.Get("KAFKA_TLS_INSECURE_SKIP_VERIFY") == "true",
 			}
 
+			pubsubBrokers := strings.Split(conf.Get("PUBSUB_BROKER"), ",")
+
 			c.PubSub = kafka.New(&kafka.Config{
-				Broker:           conf.Get("PUBSUB_BROKER"),
+				Brokers:          pubsubBrokers,
 				Partition:        partition,
 				ConsumerGroupID:  conf.Get("CONSUMER_ID"),
 				OffSet:           offSet,
diff --git a/pkg/gofr/container/mock_container.go b/pkg/gofr/container/mock_container.go
index 27e7de87..4fbbf5f2 100644
--- a/pkg/gofr/container/mock_container.go
+++ b/pkg/gofr/container/mock_container.go
@@ -33,8 +33,8 @@ type Mocks struct {
 
 type options func(c *Container, ctrl *gomock.Controller) any
 
-//nolint:revive // WithMockHTTPService returns an exported type intentionally; options are internal and subject to change.
-func WithMockHTTPService(httpServiceNames ...string) options {
+func WithMockHTTPService(httpServiceNames ...string) options { //nolint:revive // WithMockHTTPService returns an
+	// exported type intentionally; options are internal and subject to change.
 	return func(c *Container, ctrl *gomock.Controller) any {
 		mockservice := service.NewMockHTTP(ctrl)
 		for _, s := range httpServiceNames {
diff --git a/pkg/gofr/datasource/pubsub/kafka/conn.go b/pkg/gofr/datasource/pubsub/kafka/conn.go
new file mode 100644
index 00000000..46100e5c
--- /dev/null
+++ b/pkg/gofr/datasource/pubsub/kafka/conn.go
@@ -0,0 +1,188 @@
+package kafka
+
+import (
+	"context"
+	"errors"
+	"net"
+	"strconv"
+	"sync"
+
+	"github.com/segmentio/kafka-go"
+)
+
+//nolint:unused // We need this wrap around for testing purposes.
+type Conn struct {
+	conns []*kafka.Conn
+}
+
+// initialize creates and configures all Kafka client components.
+func (k *kafkaClient) initialize(ctx context.Context) error {
+	dialer, err := setupDialer(&k.config)
+	if err != nil {
+		return err
+	}
+
+	conns, err := connectToBrokers(ctx, k.config.Brokers, dialer, k.logger)
+	if err != nil {
+		return err
+	}
+
+	multi := &multiConn{
+		conns:  conns,
+		dialer: dialer,
+	}
+
+	writer := createKafkaWriter(&k.config, dialer, k.logger)
+	reader := make(map[string]Reader)
+
+	k.logger.Logf("connected to %d Kafka brokers", len(conns))
+
+	k.dialer = dialer
+	k.conn = multi
+	k.writer = writer
+	k.reader = reader
+
+	return nil
+}
+
+func (k *kafkaClient) getNewReader(topic string) Reader {
+	reader := kafka.NewReader(kafka.ReaderConfig{
+		GroupID:     k.config.ConsumerGroupID,
+		Brokers:     k.config.Brokers,
+		Topic:       topic,
+		MinBytes:    10e3,
+		MaxBytes:    10e6,
+		Dialer:      k.dialer,
+		StartOffset: int64(k.config.OffSet),
+	})
+
+	return reader
+}
+
+func (k *kafkaClient) DeleteTopic(_ context.Context, name string) error {
+	return k.conn.DeleteTopics(name)
+}
+
+func (k *kafkaClient) Controller() (broker kafka.Broker, err error) {
+	return k.conn.Controller()
+}
+
+func (k *kafkaClient) CreateTopic(_ context.Context, name string) error {
+	topics := kafka.TopicConfig{Topic: name, NumPartitions: 1, ReplicationFactor: 1}
+
+	return k.conn.CreateTopics(topics)
+}
+
+type multiConn struct {
+	conns  []Connection
+	dialer *kafka.Dialer
+	mu     sync.RWMutex
+}
+
+func (m *multiConn) Controller() (kafka.Broker, error) {
+	if len(m.conns) == 0 {
+		return kafka.Broker{}, errNoActiveConnections
+	}
+
+	// Try all connections until we find one that works
+	for _, conn := range m.conns {
+		if conn == nil {
+			continue
+		}
+
+		controller, err := conn.Controller()
+		if err == nil {
+			return controller, nil
+		}
+	}
+
+	return kafka.Broker{}, errNoActiveConnections
+}
+
+func (m *multiConn) CreateTopics(topics ...kafka.TopicConfig) error {
+	controller, err := m.Controller()
+	if err != nil {
+		return err
+	}
+
+	controllerAddr := net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port))
+
+	controllerResolvedAddr, err := net.ResolveTCPAddr("tcp", controllerAddr)
+	if err != nil {
+		return err
+	}
+
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	for _, conn := range m.conns {
+		if conn == nil {
+			continue
+		}
+
+		if tcpAddr, ok := conn.RemoteAddr().(*net.TCPAddr); ok {
+			if tcpAddr.IP.Equal(controllerResolvedAddr.IP) && tcpAddr.Port == controllerResolvedAddr.Port {
+				return conn.CreateTopics(topics...)
+			}
+		}
+	}
+
+	// If not found, create a new connection
+	conn, err := m.dialer.DialContext(context.Background(), "tcp", controllerAddr)
+	if err != nil {
+		return err
+	}
+
+	m.conns = append(m.conns, conn)
+
+	return conn.CreateTopics(topics...)
+}
+
+func (m *multiConn) DeleteTopics(topics ...string) error {
+	controller, err := m.Controller()
+	if err != nil {
+		return err
+	}
+
+	controllerAddr := net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port))
+
+	controllerResolvedAddr, err := net.ResolveTCPAddr("tcp", controllerAddr)
+	if err != nil {
+		return err
+	}
+
+	for _, conn := range m.conns {
+		if conn == nil {
+			continue
+		}
+
+		if tcpAddr, ok := conn.RemoteAddr().(*net.TCPAddr); ok {
+			// Match IP (after resolution) and Port
+			if tcpAddr.IP.Equal(controllerResolvedAddr.IP) && tcpAddr.Port == controllerResolvedAddr.Port {
+				return conn.DeleteTopics(topics...)
+			}
+		}
+	}
+
+	// If not found, create a new connection
+	conn, err := m.dialer.DialContext(context.Background(), "tcp", controllerAddr)
+	if err != nil {
+		return err
+	}
+
+	m.conns = append(m.conns, conn)
+
+	return conn.DeleteTopics(topics...)
+}
+
+func (m *multiConn) Close() error {
+	var err error
+
+	for _, conn := range m.conns {
+		if conn != nil {
+			err = errors.Join(err, conn.Close())
+		}
+	}
+
+	return err
+}
diff --git a/pkg/gofr/datasource/pubsub/kafka/errors.go b/pkg/gofr/datasource/pubsub/kafka/errors.go
new file mode 100644
index 00000000..188a0d53
--- /dev/null
+++ b/pkg/gofr/datasource/pubsub/kafka/errors.go
@@ -0,0 +1,22 @@
+package kafka
+
+import "errors"
+
+var (
+	ErrConsumerGroupNotProvided    = errors.New("consumer group id not provided")
+	errFailedToConnectBrokers      = errors.New("failed to connect to any kafka brokers")
+	errBrokerNotProvided           = errors.New("kafka broker address not provided")
+	errPublisherNotConfigured      = errors.New("can't publish message. Publisher not configured or topic is empty")
+	errBatchSize                   = errors.New("KAFKA_BATCH_SIZE must be greater than 0")
+	errBatchBytes                  = errors.New("KAFKA_BATCH_BYTES must be greater than 0")
+	errBatchTimeout                = errors.New("KAFKA_BATCH_TIMEOUT must be greater than 0")
+	errClientNotConnected          = errors.New("kafka client not connected")
+	errUnsupportedSASLMechanism    = errors.New("unsupported SASL mechanism")
+	errSASLCredentialsMissing      = errors.New("SASL credentials missing")
+	errUnsupportedSecurityProtocol = errors.New("unsupported security protocol")
+	errNoActiveConnections         = errors.New("no active connections to brokers")
+	errCACertFileRead              = errors.New("failed to read CA certificate file")
+	errClientCertLoad              = errors.New("failed to load client certificate")
+	errNotController               = errors.New("not a controller")
+	errUnreachable                 = errors.New("unreachable")
+)
diff --git a/pkg/gofr/datasource/pubsub/kafka/health.go b/pkg/gofr/datasource/pubsub/kafka/health.go
index 393a03da..e7af1af9 100644
--- a/pkg/gofr/datasource/pubsub/kafka/health.go
+++ b/pkg/gofr/datasource/pubsub/kafka/health.go
@@ -2,28 +2,96 @@ package kafka
 
 import (
 	"encoding/json"
+	"fmt"
+	"net"
+	"strconv"
 
 	"gofr.dev/pkg/gofr/datasource"
 )
 
-func (k *kafkaClient) Health() (health datasource.Health) {
-	health = datasource.Health{Details: make(map[string]any)}
+func (k *kafkaClient) Health() datasource.Health {
+	health := datasource.Health{
+		Status:  datasource.StatusDown,
+		Details: make(map[string]any),
+	}
+
+	if k.conn == nil {
+		health.Details["error"] = "invalid connection type"
+		return health
+	}
 
-	health.Status = datasource.StatusUp
+	k.conn.mu.RLock()
+	defer k.conn.mu.RUnlock()
 
-	_, err := k.conn.Controller()
-	if err != nil {
-		health.Status = datasource.StatusDown
+	brokerStatus, allDown := k.evaluateBrokerHealth()
+
+	if !allDown {
+		health.Status = datasource.StatusUp
 	}
 
-	health.Details["host"] = k.config.Broker
+	health.Details["brokers"] = brokerStatus
 	health.Details["backend"] = "KAFKA"
-	health.Details["writers"] = k.getWriterStatsAsMap()
+	health.Details["writer"] = k.getWriterStatsAsMap()
 	health.Details["readers"] = k.getReaderStatsAsMap()
 
 	return health
 }
 
+func (k *kafkaClient) evaluateBrokerHealth() ([]map[string]any, bool) {
+	var (
+		statusList     = make([]map[string]any, 0)
+		controllerAddr string
+		allDown        = true
+	)
+
+	for _, conn := range k.conn.conns {
+		if conn == nil {
+			continue
+		}
+
+		status := checkBroker(conn, &controllerAddr)
+
+		if status["status"] == brokerStatusUp {
+			allDown = false
+		}
+
+		statusList = append(statusList, status)
+	}
+
+	return statusList, allDown
+}
+
+func checkBroker(conn Connection, controllerAddr *string) map[string]any {
+	brokerAddr := conn.RemoteAddr().String()
+	status := map[string]any{
+		"broker":       brokerAddr,
+		"status":       "DOWN",
+		"isController": false,
+		"error":        nil,
+	}
+
+	_, err := conn.ReadPartitions()
+	if err != nil {
+		status["error"] = err.Error()
+		return status
+	}
+
+	status["status"] = brokerStatusUp
+
+	if *controllerAddr == "" {
+		controller, err := conn.Controller()
+		if err != nil {
+			status["error"] = fmt.Sprintf("controller lookup failed: %v", err)
+		} else if controller.Host != "" {
+			*controllerAddr = net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port))
+		}
+	}
+
+	status["isController"] = brokerAddr == *controllerAddr
+
+	return status
+}
+
 func (k *kafkaClient) getReaderStatsAsMap() []any {
 	readerStats := make([]any, 0)
 
diff --git a/pkg/gofr/datasource/pubsub/kafka/helper.go b/pkg/gofr/datasource/pubsub/kafka/helper.go
new file mode 100644
index 00000000..03b21f0d
--- /dev/null
+++ b/pkg/gofr/datasource/pubsub/kafka/helper.go
@@ -0,0 +1,150 @@
+package kafka
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/segmentio/kafka-go"
+
+	"gofr.dev/pkg/gofr/datasource/pubsub"
+)
+
+func validateConfigs(conf *Config) error {
+	if err := validateRequiredFields(conf); err != nil {
+		return err
+	}
+
+	setDefaultSecurityProtocol(conf)
+
+	if err := validateSASLConfigs(conf); err != nil {
+		return err
+	}
+
+	if err := validateTLSConfigs(conf); err != nil {
+		return err
+	}
+
+	if err := validateSecurityProtocol(conf); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func validateRequiredFields(conf *Config) error {
+	if len(conf.Brokers) == 0 {
+		return errBrokerNotProvided
+	}
+
+	if conf.BatchSize <= 0 {
+		return fmt.Errorf("batch size must be greater than 0: %w", errBatchSize)
+	}
+
+	if conf.BatchBytes <= 0 {
+		return fmt.Errorf("batch bytes must be greater than 0: %w", errBatchBytes)
+	}
+
+	if conf.BatchTimeout <= 0 {
+		return fmt.Errorf("batch timeout must be greater than 0: %w", errBatchTimeout)
+	}
+
+	return nil
+}
+
+// retryConnect handles the retry mechanism for connecting to the Kafka broker.
+func (k *kafkaClient) retryConnect(ctx context.Context) {
+	for {
+		time.Sleep(defaultRetryTimeout)
+
+		err := k.initialize(ctx)
+		if err != nil {
+			var brokers any
+
+			if len(k.config.Brokers) > 1 {
+				brokers = k.config.Brokers
+			} else {
+				brokers = k.config.Brokers[0]
+			}
+
+			k.logger.Errorf("could not connect to Kafka at '%v', error: %v", brokers, err)
+
+			continue
+		}
+
+		return
+	}
+}
+
+func (k *kafkaClient) isConnected() bool {
+	if k.conn == nil {
+		return false
+	}
+
+	_, err := k.conn.Controller()
+
+	return err == nil
+}
+
+func setupDialer(conf *Config) (*kafka.Dialer, error) {
+	dialer := &kafka.Dialer{
+		Timeout:   10 * time.Second,
+		DualStack: true,
+	}
+
+	if conf.SecurityProtocol == protocolSASL || conf.SecurityProtocol == protocolSASLSSL {
+		mechanism, err := getSASLMechanism(conf.SASLMechanism, conf.SASLUser, conf.SASLPassword)
+		if err != nil {
+			return nil, err
+		}
+
+		dialer.SASLMechanism = mechanism
+	}
+
+	if conf.SecurityProtocol == "SSL" || conf.SecurityProtocol == "SASL_SSL" {
+		tlsConfig, err := createTLSConfig(&conf.TLS)
+		if err != nil {
+			return nil, err
+		}
+
+		dialer.TLS = tlsConfig
+	}
+
+	return dialer, nil
+}
+
+// connectToBrokers connects to Kafka brokers with context support.
+func connectToBrokers(ctx context.Context, brokers []string, dialer *kafka.Dialer, logger pubsub.Logger) ([]Connection, error) {
+	conns := make([]Connection, 0)
+
+	if len(brokers) == 0 {
+		return nil, errBrokerNotProvided
+	}
+
+	for _, broker := range brokers {
+		conn, err := dialer.DialContext(ctx, "tcp", broker)
+		if err != nil {
+			logger.Errorf("failed to connect to broker %s: %v", broker, err)
+			continue
+		}
+
+		conns = append(conns, conn)
+	}
+
+	if len(conns) == 0 {
+		return nil, errFailedToConnectBrokers
+	}
+
+	return conns, nil
+}
+
+func createKafkaWriter(conf *Config, dialer *kafka.Dialer, logger pubsub.Logger) Writer {
+	return kafka.NewWriter(kafka.WriterConfig{
+		Brokers:      conf.Brokers,
+		Dialer:       dialer,
+		BatchSize:    conf.BatchSize,
+		BatchBytes:   conf.BatchBytes,
+		BatchTimeout: time.Duration(conf.BatchTimeout),
+		Logger:       kafka.LoggerFunc(logger.Debugf),
+	})
+}
diff --git a/pkg/gofr/datasource/pubsub/kafka/interfaces.go b/pkg/gofr/datasource/pubsub/kafka/interfaces.go
index 1af87fa8..40aa4be3 100644
--- a/pkg/gofr/datasource/pubsub/kafka/interfaces.go
+++ b/pkg/gofr/datasource/pubsub/kafka/interfaces.go
@@ -2,6 +2,7 @@ package kafka
 
 import (
 	"context"
+	"net"
 
 	"github.com/segmentio/kafka-go"
 )
@@ -26,5 +27,7 @@ type Connection interface {
 	Controller() (broker kafka.Broker, err error)
 	CreateTopics(topics ...kafka.TopicConfig) error
 	DeleteTopics(topics ...string) error
+	RemoteAddr() net.Addr
+	ReadPartitions(topics ...string) (partitions []kafka.Partition, err error)
 	Close() error
 }
diff --git a/pkg/gofr/datasource/pubsub/kafka/kafka.go b/pkg/gofr/datasource/pubsub/kafka/kafka.go
index 61f0ea9a..5822ddee 100644
--- a/pkg/gofr/datasource/pubsub/kafka/kafka.go
+++ b/pkg/gofr/datasource/pubsub/kafka/kafka.go
@@ -5,7 +5,6 @@ package kafka
 import (
 	"context"
 	"errors"
-	"fmt"
 	"sync"
 	"time"
 
@@ -15,32 +14,21 @@ import (
 	"gofr.dev/pkg/gofr/datasource/pubsub"
 )
 
-var (
-	ErrConsumerGroupNotProvided    = errors.New("consumer group id not provided")
-	errBrokerNotProvided           = errors.New("kafka broker address not provided")
-	errPublisherNotConfigured      = errors.New("can't publish message. Publisher not configured or topic is empty")
-	errBatchSize                   = errors.New("KAFKA_BATCH_SIZE must be greater than 0")
-	errBatchBytes                  = errors.New("KAFKA_BATCH_BYTES must be greater than 0")
-	errBatchTimeout                = errors.New("KAFKA_BATCH_TIMEOUT must be greater than 0")
-	errClientNotConnected          = errors.New("kafka client not connected")
-	errUnsupportedSASLMechanism    = errors.New("unsupported SASL mechanism")
-	errSASLCredentialsMissing      = errors.New("SASL credentials missing")
-	errUnsupportedSecurityProtocol = errors.New("unsupported security protocol")
-)
-
 const (
-	DefaultBatchSize    = 100
-	DefaultBatchBytes   = 1048576
-	DefaultBatchTimeout = 1000
-	defaultRetryTimeout = 10 * time.Second
-	protocolPlainText   = "PLAINTEXT"
-	protocolSASL        = "SASL_PLAINTEXT"
-	protocolSSL         = "SSL"
-	protocolSASLSSL     = "SASL_SSL"
+	DefaultBatchSize       = 100
+	DefaultBatchBytes      = 1048576
+	DefaultBatchTimeout    = 1000
+	defaultRetryTimeout    = 10 * time.Second
+	protocolPlainText      = "PLAINTEXT"
+	protocolSASL           = "SASL_PLAINTEXT"
+	protocolSSL            = "SSL"
+	protocolSASLSSL        = "SASL_SSL"
+	messageMultipleBrokers = "MULTIPLE_BROKERS"
+	brokerStatusUp         = "UP"
 )
 
 type Config struct {
-	Broker           string
+	Brokers          []string
 	Partition        int
 	ConsumerGroupID  string
 	OffSet           int
@@ -57,7 +45,7 @@ type Config struct {
 
 type kafkaClient struct {
 	dialer *kafka.Dialer
-	conn   Connection
+	conn   *multiConn
 
 	writer Writer
 	reader map[string]Reader
@@ -69,8 +57,8 @@ type kafkaClient struct {
 	metrics Metrics
 }
 
-//nolint:revive // New allow returning unexported types as intended.
-func New(conf *Config, logger pubsub.Logger, metrics Metrics) *kafkaClient {
+func New(conf *Config, logger pubsub.Logger, metrics Metrics) *kafkaClient { //nolint:revive // New allows
+	// returning unexported types as intended.
 	err := validateConfigs(conf)
 	if err != nil {
 		logger.Errorf("could not initialize kafka, error: %v", err)
@@ -78,76 +66,31 @@ func New(conf *Config, logger pubsub.Logger, metrics Metrics) *kafkaClient {
 		return nil
 	}
 
-	logger.Debugf("connecting to Kafka broker '%s'", conf.Broker)
-
-	dialer, conn, writer, reader, err := initializeKafkaClient(conf, logger)
-	if err != nil {
-		logger.Errorf("failed to connect to kafka at %v, error: %v", conf.Broker, err)
-
-		client := &kafkaClient{
-			logger:  logger,
-			config:  *conf,
-			metrics: metrics,
-			mu:      &sync.RWMutex{},
-		}
-
-		go retryConnect(client, conf, logger)
-
-		return client
+	if len(conf.Brokers) == 1 {
+		logger.Debugf("connecting to Kafka broker: '%s'", conf.Brokers[0])
+	} else {
+		logger.Debugf("connecting to Kafka brokers: %v", conf.Brokers)
 	}
 
-	return &kafkaClient{
-		config:  *conf,
-		dialer:  dialer,
-		reader:  reader,
-		conn:    conn,
+	client := &kafkaClient{
 		logger:  logger,
-		writer:  writer,
-		mu:      &sync.RWMutex{},
+		config:  *conf,
 		metrics: metrics,
+		mu:      &sync.RWMutex{},
 	}
-}
-
-func validateConfigs(conf *Config) error {
-	if err := validateRequiredFields(conf); err != nil {
-		return err
-	}
-
-	setDefaultSecurityProtocol(conf)
-
-	if err := validateSASLConfigs(conf); err != nil {
-		return err
-	}
-
-	if err := validateTLSConfigs(conf); err != nil {
-		return err
-	}
-
-	if err := validateSecurityProtocol(conf); err != nil {
-		return err
-	}
+	ctx := context.Background()
 
-	return nil
-}
+	err = client.initialize(ctx)
 
-func validateRequiredFields(conf *Config) error {
-	if conf.Broker == "" {
-		return errBrokerNotProvided
-	}
-
-	if conf.BatchSize <= 0 {
-		return fmt.Errorf("batch size must be greater than 0: %w", errBatchSize)
-	}
+	if err != nil {
+		logger.Errorf("failed to connect to kafka at %v, error: %v", conf.Brokers, err)
 
-	if conf.BatchBytes <= 0 {
-		return fmt.Errorf("batch bytes must be greater than 0: %w", errBatchBytes)
-	}
+		go client.retryConnect(ctx)
 
-	if conf.BatchTimeout <= 0 {
-		return fmt.Errorf("batch timeout must be greater than 0: %w", errBatchTimeout)
+		return client
 	}
 
-	return nil
+	return client
 }
 
 func (k *kafkaClient) Publish(ctx context.Context, topic string, message []byte) error {
@@ -175,12 +118,20 @@ func (k *kafkaClient) Publish(ctx context.Context, topic string, message []byte)
 		return err
 	}
 
+	var hostName string
+
+	if len(k.config.Brokers) > 1 {
+		hostName = messageMultipleBrokers
+	} else {
+		hostName = k.config.Brokers[0]
+	}
+
 	k.logger.Debug(&pubsub.Log{
 		Mode:          "PUB",
 		CorrelationID: span.SpanContext().TraceID().String(),
 		MessageValue:  string(message),
 		Topic:         topic,
-		Host:          k.config.Broker,
+		Host:          hostName,
 		PubSubBackend: "KAFKA",
 		Time:          end.Microseconds(),
 	})
@@ -242,12 +193,20 @@ func (k *kafkaClient) Subscribe(ctx context.Context, topic string) (*pubsub.Mess
 
 	end := time.Since(start)
 
+	var hostName string
+
+	if len(k.config.Brokers) > 1 {
+		hostName = "multiple brokers"
+	} else {
+		hostName = k.config.Brokers[0]
+	}
+
 	k.logger.Debug(&pubsub.Log{
 		Mode:          "SUB",
 		CorrelationID: span.SpanContext().TraceID().String(),
 		MessageValue:  string(msg.Value),
 		Topic:         topic,
-		Host:          k.config.Broker,
+		Host:          hostName,
 		PubSubBackend: "KAFKA",
 		Time:          end.Microseconds(),
 	})
@@ -263,7 +222,7 @@ func (k *kafkaClient) Close() (err error) {
 	}
 
 	if k.writer != nil {
-		err = k.writer.Close()
+		err = errors.Join(err, k.writer.Close())
 	}
 
 	if k.conn != nil {
@@ -272,113 +231,3 @@ func (k *kafkaClient) Close() (err error) {
 
 	return err
 }
-
-func initializeKafkaClient(conf *Config, logger pubsub.Logger) (*kafka.Dialer, Connection,
-	Writer, map[string]Reader, error) {
-	dialer := &kafka.Dialer{
-		Timeout:   10 * time.Second,
-		DualStack: true,
-	}
-
-	if conf.SecurityProtocol == protocolSASL || conf.SecurityProtocol == protocolSASLSSL {
-		mechanism, err := getSASLMechanism(conf.SASLMechanism, conf.SASLUser, conf.SASLPassword)
-		if err != nil {
-			return nil, nil, nil, nil, err
-		}
-
-		dialer.SASLMechanism = mechanism
-	}
-
-	if conf.SecurityProtocol == "SSL" || conf.SecurityProtocol == "SASL_SSL" {
-		tlsConfig, err := createTLSConfig(&conf.TLS)
-		if err != nil {
-			return nil, nil, nil, nil, err
-		}
-
-		dialer.TLS = tlsConfig
-	}
-
-	conn, err := dialer.DialContext(context.Background(), "tcp", conf.Broker)
-	if err != nil {
-		return nil, nil, nil, nil, err
-	}
-
-	writer := kafka.NewWriter(kafka.WriterConfig{
-		Brokers:      []string{conf.Broker},
-		Dialer:       dialer,
-		BatchSize:    conf.BatchSize,
-		BatchBytes:   conf.BatchBytes,
-		BatchTimeout: time.Duration(conf.BatchTimeout),
-	})
-
-	reader := make(map[string]Reader)
-
-	logger.Logf("connected to Kafka broker '%s'", conf.Broker)
-
-	return dialer, conn, writer, reader, nil
-}
-
-func (k *kafkaClient) getNewReader(topic string) Reader {
-	reader := kafka.NewReader(kafka.ReaderConfig{
-		GroupID:     k.config.ConsumerGroupID,
-		Brokers:     []string{k.config.Broker},
-		Topic:       topic,
-		MinBytes:    10e3,
-		MaxBytes:    10e6,
-		Dialer:      k.dialer,
-		StartOffset: int64(k.config.OffSet),
-	})
-
-	return reader
-}
-
-func (k *kafkaClient) DeleteTopic(_ context.Context, name string) error {
-	return k.conn.DeleteTopics(name)
-}
-
-func (k *kafkaClient) Controller() (broker kafka.Broker, err error) {
-	return k.conn.Controller()
-}
-
-func (k *kafkaClient) CreateTopic(_ context.Context, name string) error {
-	topics := kafka.TopicConfig{Topic: name, NumPartitions: 1, ReplicationFactor: 1}
-
-	err := k.conn.CreateTopics(topics)
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
-// retryConnect handles the retry mechanism for connecting to the Kafka broker.
-func retryConnect(client *kafkaClient, conf *Config, logger pubsub.Logger) {
-	for {
-		time.Sleep(defaultRetryTimeout)
-
-		dialer, conn, writer, reader, err := initializeKafkaClient(conf, logger)
-		if err != nil {
-			logger.Errorf("could not connect to Kafka at '%v', error: %v", conf.Broker, err)
-			continue
-		}
-
-		client.mu.Lock()
-		client.conn = conn
-		client.dialer = dialer
-		client.writer = writer
-		client.reader = reader
-		client.mu.Unlock()
-
-		return
-	}
-}
-
-func (k *kafkaClient) isConnected() bool {
-	if k.conn == nil {
-		return false
-	}
-
-	_, err := k.conn.Controller()
-
-	return err == nil
-}
diff --git a/pkg/gofr/datasource/pubsub/kafka/mock_interfaces.go b/pkg/gofr/datasource/pubsub/kafka/mock_interfaces.go
index a6d08a94..4b1b6466 100644
--- a/pkg/gofr/datasource/pubsub/kafka/mock_interfaces.go
+++ b/pkg/gofr/datasource/pubsub/kafka/mock_interfaces.go
@@ -11,6 +11,7 @@ package kafka
 
 import (
 	context "context"
+	net "net"
 	reflect "reflect"
 
 	kafka "github.com/segmentio/kafka-go"
@@ -21,6 +22,7 @@ import (
 type MockReader struct {
 	ctrl     *gomock.Controller
 	recorder *MockReaderMockRecorder
+	isgomock struct{}
 }
 
 // MockReaderMockRecorder is the mock recorder for MockReader.
@@ -121,6 +123,7 @@ func (mr *MockReaderMockRecorder) Stats() *gomock.Call {
 type MockWriter struct {
 	ctrl     *gomock.Controller
 	recorder *MockWriterMockRecorder
+	isgomock struct{}
 }
 
 // MockWriterMockRecorder is the mock recorder for MockWriter.
@@ -191,6 +194,7 @@ func (mr *MockWriterMockRecorder) WriteMessages(ctx any, msg ...any) *gomock.Cal
 type MockConnection struct {
 	ctrl     *gomock.Controller
 	recorder *MockConnectionMockRecorder
+	isgomock struct{}
 }
 
 // MockConnectionMockRecorder is the mock recorder for MockConnection.
@@ -274,3 +278,36 @@ func (mr *MockConnectionMockRecorder) DeleteTopics(topics ...any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteTopics", reflect.TypeOf((*MockConnection)(nil).DeleteTopics), topics...)
 }
+
+// ReadPartitions mocks base method.
+func (m *MockConnection) ReadPartitions(topics ...string) ([]kafka.Partition, error) {
+	m.ctrl.T.Helper()
+	varargs := []any{}
+	for _, a := range topics {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "ReadPartitions", varargs...)
+	ret0, _ := ret[0].([]kafka.Partition)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ReadPartitions indicates an expected call of ReadPartitions.
+func (mr *MockConnectionMockRecorder) ReadPartitions(topics ...any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReadPartitions", reflect.TypeOf((*MockConnection)(nil).ReadPartitions), topics...)
+}
+
+// RemoteAddr mocks base method.
+func (m *MockConnection) RemoteAddr() net.Addr {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "RemoteAddr")
+	ret0, _ := ret[0].(net.Addr)
+	return ret0
+}
+
+// RemoteAddr indicates an expected call of RemoteAddr.
+func (mr *MockConnectionMockRecorder) RemoteAddr() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RemoteAddr", reflect.TypeOf((*MockConnection)(nil).RemoteAddr))
+}
diff --git a/pkg/gofr/datasource/pubsub/kafka/tls.go b/pkg/gofr/datasource/pubsub/kafka/tls.go
index 139ee084..19a81439 100644
--- a/pkg/gofr/datasource/pubsub/kafka/tls.go
+++ b/pkg/gofr/datasource/pubsub/kafka/tls.go
@@ -3,17 +3,11 @@ package kafka
 import (
 	"crypto/tls"
 	"crypto/x509"
-	"errors"
 	"fmt"
 	"os"
 	"strings"
 )
 
-var (
-	errCACertFileRead = errors.New("failed to read CA certificate file")
-	errClientCertLoad = errors.New("failed to load client certificate")
-)
-
 type TLSConfig struct {
 	CertFile           string
 	KeyFile            string



Test patch:
diff --git a/examples/using-subscriber/main_test.go b/examples/using-subscriber/main_test.go
index 31894499..00730cc7 100644
--- a/examples/using-subscriber/main_test.go
+++ b/examples/using-subscriber/main_test.go
@@ -25,7 +25,7 @@ func (m *mockMetrics) IncrementCounter(ctx context.Context, name string, labels
 
 func initializeTest(t *testing.T) {
 	c := kafka.New(&kafka.Config{
-		Broker:       "localhost:9092",
+		Brokers:      []string{"localhost:9092"},
 		OffSet:       1,
 		BatchSize:    kafka.DefaultBatchSize,
 		BatchBytes:   kafka.DefaultBatchBytes,
diff --git a/pkg/gofr/datasource/pubsub/kafka/health_test.go b/pkg/gofr/datasource/pubsub/kafka/health_test.go
index 0b9e923b..52222fa8 100644
--- a/pkg/gofr/datasource/pubsub/kafka/health_test.go
+++ b/pkg/gofr/datasource/pubsub/kafka/health_test.go
@@ -1,130 +1,177 @@
 package kafka
 
 import (
+	"context"
+	"net"
 	"testing"
 
 	"github.com/segmentio/kafka-go"
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"go.uber.org/mock/gomock"
+	"github.com/stretchr/testify/mock"
 
 	"gofr.dev/pkg/gofr/datasource"
-	"gofr.dev/pkg/gofr/logging"
-	"gofr.dev/pkg/gofr/testutil"
 )
 
-func TestKafkaClient_HealthStatusUP(t *testing.T) {
-	ctrl := gomock.NewController(t)
-
-	conn := NewMockConnection(ctrl)
-	reader := NewMockReader(ctrl)
+// MockConn simulates a Kafka connection.
+type MockConn struct {
+	mock.Mock
+	addr      string
+	isHealthy bool
+	isControl bool
+}
 
-	writer := NewMockWriter(ctrl)
+func (m *MockConn) RemoteAddr() net.Addr {
+	return mockAddr{m.addr}
+}
 
-	client := &kafkaClient{
-		conn:   conn,
-		reader: map[string]Reader{"test": reader},
-		writer: writer,
+func (m *MockConn) ReadPartitions(...string) ([]kafka.Partition, error) {
+	if m.isHealthy {
+		return []kafka.Partition{{}}, nil
 	}
 
-	expectedHealth := datasource.Health{
-		Status: datasource.StatusUp,
-		Details: map[string]any{
-			"host":    "",
-			"backend": "KAFKA",
-		},
-	}
+	return nil, errUnreachable
+}
 
-	conn.EXPECT().Controller().Return(kafka.Broker{}, nil)
-	writer.EXPECT().Stats().Return(kafka.WriterStats{Topic: "test"})
-	reader.EXPECT().Stats().Return(kafka.ReaderStats{Topic: "test"})
+func (m *MockConn) Controller() (kafka.Broker, error) {
+	if m.isControl {
+		host, _, _ := net.SplitHostPort(m.addr)
 
-	health := client.Health()
+		port := 9092
 
-	assert.Equal(t, expectedHealth.Details["host"], health.Details["host"])
-	assert.Equal(t, expectedHealth.Details["backend"], health.Details["backend"])
-	assert.Equal(t, expectedHealth.Status, health.Status)
+		return kafka.Broker{Host: host, Port: port}, nil
+	}
+
+	return kafka.Broker{}, errNotController
 }
 
-func TestKafkaClient_HealthStatusDown(t *testing.T) {
-	ctrl := gomock.NewController(t)
+// Add minimal required method stubs.
+func (*MockConn) Close() error                            { return nil }
+func (*MockConn) CreateTopics(...kafka.TopicConfig) error { return nil }
+func (*MockConn) DeleteTopics(...string) error            { return nil }
 
-	conn := NewMockConnection(ctrl)
-	reader := NewMockReader(ctrl)
+type mockAddr struct{ addr string }
 
-	writer := NewMockWriter(ctrl)
+func (mockAddr) Network() string  { return "tcp" }
+func (m mockAddr) String() string { return m.addr }
 
+func TestKafkaHealth_AllBrokersUp(t *testing.T) {
 	client := &kafkaClient{
-		conn:   conn,
-		reader: map[string]Reader{"test": reader},
-		writer: writer,
-	}
-
-	expectedHealth := datasource.Health{
-		Status: datasource.StatusDown,
-		Details: map[string]any{
-			"host":    "",
-			"backend": "KAFKA",
+		conn: &multiConn{
+			conns: []Connection{
+				&MockConn{addr: "127.0.0.1:9092", isHealthy: true, isControl: true},
+				&MockConn{addr: "127.0.0.2:9092", isHealthy: true, isControl: false},
+			},
 		},
+		reader: make(map[string]Reader),
+		writer: &mockWriter{},
+		logger: &mockLogger{},
 	}
 
-	conn.EXPECT().Controller().Return(kafka.Broker{}, testutil.CustomError{ErrorMessage: "connection failed"})
-	writer.EXPECT().Stats().Return(kafka.WriterStats{Topic: "test"})
-	reader.EXPECT().Stats().Return(kafka.ReaderStats{Topic: "test"})
-
 	health := client.Health()
 
-	assert.Equal(t, expectedHealth.Details["host"], health.Details["host"])
-	assert.Equal(t, expectedHealth.Details["backend"], health.Details["backend"])
-	assert.Equal(t, expectedHealth.Status, health.Status)
+	assert.Equal(t, datasource.StatusUp, health.Status)
+	assert.Len(t, health.Details["brokers"], 2)
+	assert.Contains(t, health.Details["brokers"], map[string]any{
+		"broker":       "127.0.0.1:9092",
+		"status":       "UP",
+		"isController": true,
+		"error":        nil,
+	})
 }
 
-func TestKafkaClient_getWriterStatsAsMap(t *testing.T) {
-	ctrl := gomock.NewController(t)
-
-	writer := NewMockWriter(ctrl)
-
+func TestKafkaHealth_SomeBrokersUpSomeDown(t *testing.T) {
 	client := &kafkaClient{
-		logger: logging.NewMockLogger(logging.DEBUG),
-		writer: writer,
+		conn: &multiConn{
+			conns: []Connection{
+				&MockConn{addr: "127.0.0.1:9092", isHealthy: true, isControl: false},
+				&MockConn{addr: "127.0.0.2:9092", isHealthy: false},
+				&MockConn{addr: "127.0.0.3:9092", isHealthy: true, isControl: true},
+			},
+		},
+		reader: make(map[string]Reader),
+		writer: &mockWriter{},
+		logger: &mockLogger{},
 	}
 
-	writer.EXPECT().Stats().Return(kafka.WriterStats{Topic: "test"})
+	health := client.Health()
 
-	writerStats := client.getWriterStatsAsMap()
+	assert.Equal(t, datasource.StatusUp, health.Status) // Because at least one broker is down
 
-	assert.NotNil(t, writerStats)
-}
+	brokers := health.Details["brokers"].([]map[string]any)
+	assert.Len(t, brokers, 3)
+
+	statusMap := map[string]string{}
 
-func TestKafkaClient_getReaderStatsAsMap(t *testing.T) {
-	ctrl := gomock.NewController(t)
+	for _, broker := range brokers {
+		addr := broker["broker"].(string)
+		status := broker["status"].(string)
+		statusMap[addr] = status
+	}
 
-	reader := NewMockReader(ctrl)
+	assert.Equal(t, "UP", statusMap["127.0.0.1:9092"])
+	assert.Equal(t, "DOWN", statusMap["127.0.0.2:9092"])
+	assert.Equal(t, "UP", statusMap["127.0.0.3:9092"])
+}
 
+func TestKafkaHealth_AllBrokersDown(t *testing.T) {
 	client := &kafkaClient{
-		logger: logging.NewMockLogger(logging.DEBUG),
-		reader: map[string]Reader{"test": reader},
+		conn: &multiConn{
+			conns: []Connection{
+				&MockConn{addr: "127.0.0.1:9092", isHealthy: false},
+			},
+		},
+		reader: make(map[string]Reader),
+		writer: &mockWriter{},
+		logger: &mockLogger{},
 	}
 
-	reader.EXPECT().Stats().Return(kafka.ReaderStats{Topic: "test"})
+	health := client.Health()
+
+	assert.Equal(t, datasource.StatusDown, health.Status)
+	assert.Len(t, health.Details["brokers"], 1)
 
-	writerStats := client.getReaderStatsAsMap()
+	brokerInfo := health.Details["brokers"].([]map[string]any)[0]
 
-	assert.NotNil(t, writerStats)
+	assert.Equal(t, "DOWN", brokerInfo["status"])
+	assert.NotNil(t, brokerInfo["error"])
 }
 
-func TestKafkaClint_convertStructToMap(t *testing.T) {
-	testCases := []struct {
-		desc   string
-		input  any
-		output any
-	}{
-		{"unmarshal error", make(chan int), nil},
+func TestKafkaHealth_InvalidConnType(t *testing.T) {
+	client := &kafkaClient{
+		conn:   nil,
+		reader: make(map[string]Reader),
+		writer: &mockWriter{},
+		logger: &mockLogger{},
 	}
 
-	for _, v := range testCases {
-		err := convertStructToMap(v.input, v.output)
+	health := client.Health()
+
+	assert.Equal(t, datasource.StatusDown, health.Status)
+	assert.Equal(t, "invalid connection type", health.Details["error"])
+}
+
+// --- Mock implementations for Writer/Reader/Logger/Stats
 
-		require.ErrorContains(t, err, "json: unsupported type: chan int")
+type mockWriter struct{}
+
+func (*mockWriter) Stats() kafka.WriterStats {
+	return kafka.WriterStats{
+		Dials:    1,
+		Writes:   1,
+		Messages: 1,
+		Bytes:    1024,
+		Errors:   0,
 	}
 }
+
+func (*mockWriter) WriteMessages(context.Context, ...kafka.Message) error { return nil }
+func (*mockWriter) Close() error                                          { return nil }
+
+type mockLogger struct{}
+
+func (*mockLogger) Errorf(string, ...any) {}
+func (*mockLogger) Debugf(string, ...any) {}
+func (*mockLogger) Logf(string, ...any)   {}
+func (*mockLogger) Log(...any)            {}
+func (*mockLogger) Error(...any)          {}
+func (*mockLogger) Debug(...any)          {}
diff --git a/pkg/gofr/datasource/pubsub/kafka/kafka_test.go b/pkg/gofr/datasource/pubsub/kafka/kafka_test.go
index d0fe13dc..24573cfc 100644
--- a/pkg/gofr/datasource/pubsub/kafka/kafka_test.go
+++ b/pkg/gofr/datasource/pubsub/kafka/kafka_test.go
@@ -2,6 +2,7 @@ package kafka
 
 import (
 	"errors"
+	"net"
 	"sync"
 	"testing"
 
@@ -24,7 +25,7 @@ func TestValidateConfigs_ValidCases(t *testing.T) {
 		{
 			name: "Valid Config",
 			config: Config{
-				Broker:           "kafkabroker",
+				Brokers:          []string{"kafkabroker"},
 				BatchSize:        1,
 				BatchBytes:       1,
 				BatchTimeout:     1,
@@ -38,7 +39,7 @@ func TestValidateConfigs_ValidCases(t *testing.T) {
 		{
 			name: "Valid PLAINTEXT Protocol",
 			config: Config{
-				Broker:           "kafkabroker",
+				Brokers:          []string{"kafkabroker"},
 				BatchSize:        1,
 				BatchBytes:       1,
 				BatchTimeout:     1,
@@ -49,7 +50,7 @@ func TestValidateConfigs_ValidCases(t *testing.T) {
 		{
 			name: "Valid SSL Protocol with TLS Configs",
 			config: Config{
-				Broker:           "kafkabroker",
+				Brokers:          []string{"kafkabroker"},
 				BatchSize:        1,
 				BatchBytes:       1,
 				BatchTimeout:     1,
@@ -65,7 +66,7 @@ func TestValidateConfigs_ValidCases(t *testing.T) {
 		{
 			name: "Valid SASL_SSL Protocol with TLS and SASL Configs",
 			config: Config{
-				Broker:           "kafkabroker",
+				Brokers:          []string{"kafkabroker"},
 				BatchSize:        1,
 				BatchBytes:       1,
 				BatchTimeout:     1,
@@ -84,7 +85,7 @@ func TestValidateConfigs_ValidCases(t *testing.T) {
 		{
 			name: "Valid SSL Protocol with InsecureSkipVerify",
 			config: Config{
-				Broker:           "kafkabroker",
+				Brokers:          []string{"kafkabroker"},
 				BatchSize:        1,
 				BatchBytes:       1,
 				BatchTimeout:     1,
@@ -125,7 +126,7 @@ func TestValidateConfigs_InvalidCases(t *testing.T) {
 		{
 			name: "Zero BatchSize",
 			config: Config{
-				Broker:       "kafkabroker",
+				Brokers:      []string{"kafkabroker"},
 				BatchSize:    0,
 				BatchBytes:   1,
 				BatchTimeout: 1,
@@ -135,7 +136,7 @@ func TestValidateConfigs_InvalidCases(t *testing.T) {
 		{
 			name: "Zero BatchBytes",
 			config: Config{
-				Broker:       "kafkabroker",
+				Brokers:      []string{"kafkabroker"},
 				BatchSize:    1,
 				BatchBytes:   0,
 				BatchTimeout: 1,
@@ -145,7 +146,7 @@ func TestValidateConfigs_InvalidCases(t *testing.T) {
 		{
 			name: "Zero BatchTimeout",
 			config: Config{
-				Broker:       "kafkabroker",
+				Brokers:      []string{"kafkabroker"},
 				BatchSize:    1,
 				BatchBytes:   1,
 				BatchTimeout: 0,
@@ -155,7 +156,7 @@ func TestValidateConfigs_InvalidCases(t *testing.T) {
 		{
 			name: "SASL_PLAINTEXT with Missing SASLMechanism",
 			config: Config{
-				Broker:           "kafkabroker",
+				Brokers:          []string{"kafkabroker"},
 				BatchSize:        1,
 				BatchBytes:       1,
 				BatchTimeout:     1,
@@ -168,7 +169,7 @@ func TestValidateConfigs_InvalidCases(t *testing.T) {
 		{
 			name: "Unsupported Security Protocol",
 			config: Config{
-				Broker:           "kafkabroker",
+				Brokers:          []string{"kafkabroker"},
 				BatchSize:        1,
 				BatchBytes:       1,
 				BatchTimeout:     1,
@@ -261,10 +262,16 @@ func TestKafkaClient_Publish(t *testing.T) {
 	logs := testutil.StdoutOutputForFunc(func() {
 		ctx := t.Context()
 		logger := logging.NewMockLogger(logging.DEBUG)
-		k := &kafkaClient{writer: mockWriter, logger: logger, metrics: mockMetrics}
+		k := &kafkaClient{
+			writer:  mockWriter,
+			logger:  logger,
+			metrics: mockMetrics,
+			config: Config{
+				Brokers: []string{"localhost:9092"}, // Make sure Broker is not empty
+			},
+		}
 
-		mockWriter.EXPECT().WriteMessages(gomock.Any(), gomock.Any()).
-			Return(nil)
+		mockWriter.EXPECT().WriteMessages(gomock.Any(), gomock.Any()).Return(nil)
 		mockMetrics.EXPECT().IncrementCounter(gomock.Any(), "app_pubsub_publish_total_count", "topic", "test")
 		mockMetrics.EXPECT().IncrementCounter(gomock.Any(), "app_pubsub_publish_success_count", "topic", "test")
 
@@ -298,11 +305,15 @@ func TestKafkaClient_SubscribeSuccess(t *testing.T) {
 		reader: map[string]Reader{
 			"test": mockReader,
 		},
-		conn:   mockConnection,
+		conn: &multiConn{
+			conns: []Connection{
+				mockConnection,
+			},
+		},
 		logger: nil,
 		config: Config{
 			ConsumerGroupID: "consumer",
-			Broker:          "kafkabroker",
+			Brokers:         []string{"kafkabroker"},
 			OffSet:          -1,
 		},
 		mu:      &sync.RWMutex{},
@@ -345,13 +356,19 @@ func TestKafkaClient_Subscribe_ErrConsumerGroupID(t *testing.T) {
 
 	mockConnection := NewMockConnection(ctrl)
 
+	m := &multiConn{
+		conns: []Connection{
+			mockConnection,
+		},
+	}
+
 	k := &kafkaClient{
 		dialer: &kafka.Dialer{},
 		config: Config{
-			Broker: "kafkabroker",
-			OffSet: -1,
+			Brokers: []string{"kafkabroker"},
+			OffSet:  -1,
 		},
-		conn:   mockConnection,
+		conn:   m,
 		logger: logging.NewMockLogger(logging.INFO),
 	}
 
@@ -377,17 +394,23 @@ func TestKafkaClient_SubscribeError(t *testing.T) {
 	mockMetrics := NewMockMetrics(ctrl)
 	mockConnection := NewMockConnection(ctrl)
 
+	m := &multiConn{
+		conns: []Connection{
+			mockConnection,
+		},
+	}
+
 	k := &kafkaClient{
 		dialer: &kafka.Dialer{},
 		writer: nil,
 		reader: map[string]Reader{
 			"test": mockReader,
 		},
-		conn:   mockConnection,
+		conn:   m,
 		logger: logging.NewMockLogger(logging.INFO),
 		config: Config{
 			ConsumerGroupID: "consumer",
-			Broker:          "kafkabroker",
+			Brokers:         []string{"kafkabroker"},
 			OffSet:          -1,
 		},
 		mu:      &sync.RWMutex{},
@@ -421,7 +444,11 @@ func TestKafkaClient_Close(t *testing.T) {
 	mockReader := NewMockReader(ctrl)
 	mockConn := NewMockConnection(ctrl)
 
-	k := kafkaClient{reader: map[string]Reader{"test-topic": mockReader}, writer: mockWriter, conn: mockConn}
+	k := kafkaClient{reader: map[string]Reader{"test-topic": mockReader}, writer: mockWriter, conn: &multiConn{
+		conns: []Connection{
+			mockConn,
+		},
+	}}
 
 	mockWriter.EXPECT().Close().Return(nil)
 	mockReader.EXPECT().Close().Return(nil)
@@ -452,14 +479,14 @@ func TestKafkaClient_CloseError(t *testing.T) {
 	err = k.Close()
 
 	require.Error(t, err)
-	assert.Equal(t, errClose, err)
+	assert.ErrorIs(t, err, errClose)
 }
 
 func TestKafkaClient_getNewReader(t *testing.T) {
 	k := &kafkaClient{
 		dialer: &kafka.Dialer{},
 		config: Config{
-			Broker:          "kafka-broker",
+			Brokers:         []string{"kafka-broker"},
 			ConsumerGroupID: "consumer",
 			OffSet:          -1,
 		},
@@ -482,14 +509,14 @@ func TestNewKafkaClient(t *testing.T) {
 		{
 			desc: "validation of configs fail (Empty Broker)",
 			config: Config{
-				Broker: "",
+				Brokers: []string{""},
 			},
 			expectNil: true,
 		},
 		{
 			desc: "validation of configs fail (Zero Batch Bytes)",
 			config: Config{
-				Broker:     "kafka-broker",
+				Brokers:    []string{"kafka-broker"},
 				BatchBytes: 0,
 			},
 			expectNil: true,
@@ -497,7 +524,7 @@ func TestNewKafkaClient(t *testing.T) {
 		{
 			desc: "validation of configs fail (Zero Batch Size)",
 			config: Config{
-				Broker:     "kafka-broker",
+				Brokers:    []string{"kafka-broker"},
 				BatchBytes: 1,
 				BatchSize:  0,
 			},
@@ -506,7 +533,7 @@ func TestNewKafkaClient(t *testing.T) {
 		{
 			desc: "validation of configs fail (Zero Batch Timeout)",
 			config: Config{
-				Broker:       "kafka-broker",
+				Brokers:      []string{"kafka-broker"},
 				BatchBytes:   1,
 				BatchSize:    1,
 				BatchTimeout: 0,
@@ -516,7 +543,7 @@ func TestNewKafkaClient(t *testing.T) {
 		{
 			desc: "successful initialization",
 			config: Config{
-				Broker:           "kafka-broker",
+				Brokers:          []string{"kafka-broker"},
 				ConsumerGroupID:  "consumer",
 				BatchBytes:       1,
 				BatchSize:        1,
@@ -548,7 +575,11 @@ func TestKafkaClient_Controller(t *testing.T) {
 	mockClient := NewMockConnection(ctrl)
 
 	client := kafkaClient{
-		conn: mockClient,
+		conn: &multiConn{
+			conns: []Connection{
+				mockClient,
+			},
+		},
 	}
 
 	mockClient.EXPECT().Controller().Return(kafka.Broker{}, nil)
@@ -565,9 +596,24 @@ func TestKafkaClient_DeleteTopic(t *testing.T) {
 	mockClient := NewMockConnection(ctrl)
 
 	client := kafkaClient{
-		conn: mockClient,
+		conn: &multiConn{
+			conns: []Connection{
+				mockClient,
+			},
+			dialer: &kafka.Dialer{}, // Needed if fallback dialing is triggered
+		},
 	}
 
+	mockClient.EXPECT().Controller().Return(kafka.Broker{
+		Host: "localhost",
+		Port: 9092,
+	}, nil).AnyTimes()
+
+	mockClient.EXPECT().RemoteAddr().Return(&net.TCPAddr{
+		IP:   net.ParseIP("127.0.0.1"),
+		Port: 9092,
+	}).AnyTimes()
+
 	mockClient.EXPECT().DeleteTopics("test").Return(nil)
 
 	err := client.DeleteTopic(t.Context(), "test")
@@ -577,28 +623,53 @@ func TestKafkaClient_DeleteTopic(t *testing.T) {
 
 func TestKafkaClient_CreateTopic(t *testing.T) {
 	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
 
-	mockClient := NewMockConnection(ctrl)
+	mockConn := NewMockConnection(ctrl)
+
+	// IP: 127.0.0.1 Port: 9092 -> controller's resolved address
+	controllerHost := "localhost"
+	controllerPort := 9092
 
 	client := kafkaClient{
-		conn: mockClient,
+		conn: &multiConn{
+			conns: []Connection{
+				mockConn,
+			},
+			dialer: &kafka.Dialer{}, // Only used if fallback occurs
+		},
 	}
 
-	testCases := []struct {
-		desc string
-		err  error
-	}{
-		{"create success", nil},
-		{"delete success", testutil.CustomError{ErrorMessage: "custom error"}},
-	}
+	t.Run("successfully creates topic", func(t *testing.T) {
+		mockConn.EXPECT().Controller().Return(kafka.Broker{
+			Host: controllerHost,
+			Port: controllerPort,
+		}, nil)
 
-	for _, tc := range testCases {
-		mockClient.EXPECT().CreateTopics(gomock.Any()).Return(tc.err)
+		// RemoteAddr should return IP resolved version of controller
+		mockConn.EXPECT().RemoteAddr().Return(&net.TCPAddr{
+			IP:   net.ParseIP("127.0.0.1"),
+			Port: 9092,
+		})
+
+		mockConn.EXPECT().CreateTopics([]kafka.TopicConfig{
+			{
+				Topic:             "test",
+				NumPartitions:     1,
+				ReplicationFactor: 1,
+			},
+		}).Return(nil)
 
 		err := client.CreateTopic(t.Context(), "test")
+		require.NoError(t, err)
+	})
 
-		assert.Equal(t, tc.err, err)
-	}
+	t.Run("controller returns error", func(t *testing.T) {
+		mockConn.EXPECT().Controller().Return(kafka.Broker{}, errNoActiveConnections)
+
+		err := client.CreateTopic(t.Context(), "test")
+		require.EqualError(t, err, errNoActiveConnections.Error())
+	})
 }
 
 func TestKafkaClient_Subscribe_NotConnected(t *testing.T) {
@@ -615,7 +686,11 @@ func TestKafkaClient_Subscribe_NotConnected(t *testing.T) {
 
 	k := &kafkaClient{
 		dialer: &kafka.Dialer{},
-		conn:   mockConnection,
+		conn: &multiConn{
+			conns: []Connection{
+				mockConnection,
+			},
+		},
 		logger: logging.NewMockLogger(logging.DEBUG),
 	}