Repository: minio/minio (language: go)

Issue:
"fatal error: concurrent map writes" after upgrade from RELEASE.2023-10-24T04-42-36Z to RELEASE.2024-02-06T21-36-22Z
## NOTE
Running the minio cluster in kubernetes cluster with operator version 5.0.11. After upgrading minio image from RELEASE.2023-10-24T04-42-36Z to RELEASE.2024-02-06T21-36-22Z, all minio pods crash with error "fatal error: concurrent map writes".

Cluster is containing 10 nodes with Ec:3 parity scheme for the nvme disk pools, around 700TB usable capacity. 

We tested some other smaller clusters with this binary upgrade, all works. 

## Expected Behavior
Upgrade went smoothly, without crashloop.

## Current Behavior
All Minio pods start running for a few mins while waiting for reconnecting other nodes, then it will crash and error our "fatal error: concurrent map writes"

## Possible Solution
<!--- Not obligatory, but suggest a fix/reason for the bug, -->
<!--- or ideas how to implement the addition or change -->
N/A

## Steps to Reproduce (for bugs)
<!--- Provide a link to a live example, or an unambiguous set of steps to -->
<!--- reproduce this bug. Include code to reproduce, if relevant -->
<!--- and make sure you have followed https://github.com/minio/minio/tree/release/docs/debugging to capture relevant logs -->

1.Install minio operator/tenant 5.0.11
2. overwrite minio image to RELEASE.2024-02-06T21-36-22Z in values.yaml file
3. helm upgrade the tenant

Not reproducible in the smaller minio cluster


## Context
<!--- How has this issue affected you? What are you trying to accomplish? -->
<!--- Providing context helps us come up with a solution that is most useful in the real world -->
Unable to upgrade cluster to newer version minio. 

## Regression
<!-- Is this issue a regression? (Yes / No) -->
<!-- If Yes, optionally please include minio version or commit id or PR# that caused this regression, if you have these details. -->
No

## Your Environment
<!--- Include as many relevant details about the environment you experienced the bug in -->
* Version used (`minio --version`): MInio version RELEASE.2023-10-24T04-42-36Z prior the upgrade. Try to upgrade to RELEASE.2024-02-06T21-36-22Z. MInio operator version 5.0.10, minio Tenant version 5.0.11
* Server setup and configuration:  10 Dedicated node running kubernetes v1.24.10 containerd. Each host equiped 16x 7.3 TB NvMe drives mounted to file system with xfs. Minio tenant is configure use EC:3 storage class. 
* Operating System and version (`uname -a`): kernel 5.14.0-284.11.1.el9_2.x86_64, Rocky Linux 9.2 (Blue Onyx)



Discussion / hints:
Could you share the pod's yaml and pod's log @crazyeyes2023 ?
Hi @jiuker , 

Please find the pod yaml output (currently we rolled back to RELEASE.2023-10-24T04-42-36Z binary) and logs (was taken during the crashed loops when the pod running RELEASE.2024-02-06T21-36-22Z)

[minio-research-ss-2-9.log](https://github.com/minio/minio/files/14338800/minio-research-ss-2-9.log)
[minio-research-ss-2-8.log](https://github.com/minio/minio/files/14338801/minio-research-ss-2-8.log)
[minio-research-ss-2-8.yaml.zip](https://github.com/minio/minio/files/14338818/minio-research-ss-2-8.yaml.zip)
[minio-research-ss-2-9.yaml.zip](https://github.com/minio/minio/files/14338819/minio-research-ss-2-9.yaml.zip)

THanks a lot for your help in advance looking into this issue. 

Best regards, 
Kyd
Please upgrade all pods to latest release. RELEASE.2024-02-17T01-15-57Z
> Cluster is containing 10 nodes with Ec:3 parity scheme for the nvme disk pools, around 700TB usable capacity.

Large clusters like this require a subscription. Please subscribe at https://min.io/pricing 
Also as a workaround - turn-off kafka notification and upgrade the setup. 

```
fatal error: concurrent map writes
fatal error: concurrent map writes

goroutine 37496 [running]:
github.com/minio/minio/internal/event/target.(*KafkaTarget).pingBrokers.func1({{0xc10f29c503?, 0xc359b29fd0?}, 0x6747?, 0xfb?}, 0x9)
        github.com/minio/minio/internal/event/target/kafka.go:349 +0x1cd
created by github.com/minio/minio/internal/event/target.(*KafkaTarget).pingBrokers in goroutine 4776
        github.com/minio/minio/internal/event/target/kafka.go:341 +0xb5

```


Fix patch:
diff --git a/internal/event/target/kafka.go b/internal/event/target/kafka.go
index 3173e9c10dbed..c57837593e29c 100644
--- a/internal/event/target/kafka.go
+++ b/internal/event/target/kafka.go
@@ -24,12 +24,10 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"net"
 	"net/url"
 	"os"
 	"path/filepath"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/minio/minio/internal/event"
@@ -154,15 +152,15 @@ func (k KafkaArgs) Validate() error {
 type KafkaTarget struct {
 	initOnce once.Init
 
-	id          event.TargetID
-	args        KafkaArgs
-	producer    sarama.SyncProducer
-	config      *sarama.Config
-	store       store.Store[event.Event]
-	batch       *store.Batch[string, *sarama.ProducerMessage]
-	loggerOnce  logger.LogOnce
-	brokerConns map[string]net.Conn
-	quitCh      chan struct{}
+	id         event.TargetID
+	args       KafkaArgs
+	client     sarama.Client
+	producer   sarama.SyncProducer
+	config     *sarama.Config
+	store      store.Store[event.Event]
+	batch      *store.Batch[string, *sarama.ProducerMessage]
+	loggerOnce logger.LogOnce
+	quitCh     chan struct{}
 }
 
 // ID - returns target ID.
@@ -189,7 +187,9 @@ func (target *KafkaTarget) IsActive() (bool, error) {
 }
 
 func (target *KafkaTarget) isActive() (bool, error) {
-	if err := target.pingBrokers(); err != nil {
+	// Refer https://github.com/IBM/sarama/issues/1341
+	brokers := target.client.Brokers()
+	if len(brokers) == 0 {
 		return false, store.ErrNotConnected
 	}
 	return true, nil
@@ -317,54 +317,13 @@ func (target *KafkaTarget) toProducerMessage(eventData event.Event) (*sarama.Pro
 // Close - closes underneath kafka connection.
 func (target *KafkaTarget) Close() error {
 	close(target.quitCh)
-	if target.producer != nil {
-		return target.producer.Close()
-	}
-	for _, conn := range target.brokerConns {
-		if conn != nil {
-			conn.Close()
-		}
-	}
-	return nil
-}
 
-// Check if at least one broker in cluster is active
-func (target *KafkaTarget) pingBrokers() (err error) {
-	d := net.Dialer{Timeout: 1 * time.Second}
-
-	errs := make([]error, len(target.args.Brokers))
-	var wg sync.WaitGroup
-	for idx, broker := range target.args.Brokers {
-		broker := broker
-		idx := idx
-		wg.Add(1)
-		go func(broker xnet.Host, idx int) {
-			defer wg.Done()
-			conn, ok := target.brokerConns[broker.String()]
-			if !ok || conn == nil {
-				conn, errs[idx] = d.Dial("tcp", broker.String())
-				if errs[idx] != nil {
-					return
-				}
-				target.brokerConns[broker.String()] = conn
-			}
-			if _, errs[idx] = conn.Write([]byte("")); errs[idx] != nil {
-				conn.Close()
-				target.brokerConns[broker.String()] = nil
-			}
-		}(broker, idx)
+	if target.producer != nil {
+		target.producer.Close()
+		return target.client.Close()
 	}
-	wg.Wait()
 
-	var retErr error
-	for _, err := range errs {
-		if err == nil {
-			// if one of them is active we are good.
-			return nil
-		}
-		retErr = err
-	}
-	return retErr
+	return nil
 }
 
 func (target *KafkaTarget) init() error {
@@ -431,13 +390,22 @@ func (target *KafkaTarget) initKafka() error {
 		brokers = append(brokers, broker.String())
 	}
 
-	producer, err := sarama.NewSyncProducer(brokers, config)
+	client, err := sarama.NewClient(brokers, config)
+	if err != nil {
+		if !errors.Is(err, sarama.ErrOutOfBrokers) {
+			target.loggerOnce(context.Background(), err, target.ID().String())
+		}
+		return err
+	}
+
+	producer, err := sarama.NewSyncProducerFromClient(client)
 	if err != nil {
-		if err != sarama.ErrOutOfBrokers {
+		if !errors.Is(err, sarama.ErrOutOfBrokers) {
 			target.loggerOnce(context.Background(), err, target.ID().String())
 		}
 		return err
 	}
+	target.client = client
 	target.producer = producer
 
 	yes, err := target.isActive()
@@ -463,12 +431,11 @@ func NewKafkaTarget(id string, args KafkaArgs, loggerOnce logger.LogOnce) (*Kafk
 	}
 
 	target := &KafkaTarget{
-		id:          event.TargetID{ID: id, Name: "kafka"},
-		args:        args,
-		store:       queueStore,
-		loggerOnce:  loggerOnce,
-		quitCh:      make(chan struct{}),
-		brokerConns: make(map[string]net.Conn, len(args.Brokers)),
+		id:         event.TargetID{ID: id, Name: "kafka"},
+		args:       args,
+		store:      queueStore,
+		loggerOnce: loggerOnce,
+		quitCh:     make(chan struct{}),
 	}
 
 	if target.store != nil {
diff --git a/internal/logger/target/kafka/kafka.go b/internal/logger/target/kafka/kafka.go
index f96976435dc5c..b6a77abf869fb 100644
--- a/internal/logger/target/kafka/kafka.go
+++ b/internal/logger/target/kafka/kafka.go
@@ -24,7 +24,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"net"
 	"os"
 	"path/filepath"
 	"sync"
@@ -78,44 +77,6 @@ type Config struct {
 	LogOnce func(ctx context.Context, err error, id string, errKind ...interface{}) `json:"-"`
 }
 
-func (h *Target) pingBrokers() (err error) {
-	d := net.Dialer{Timeout: 1 * time.Second}
-
-	errs := make([]error, len(h.kconfig.Brokers))
-	var wg sync.WaitGroup
-	for idx, broker := range h.kconfig.Brokers {
-		broker := broker
-		idx := idx
-		wg.Add(1)
-		go func(broker xnet.Host, idx int) {
-			defer wg.Done()
-			conn, ok := h.brokerConns[broker.String()]
-			if !ok || conn == nil {
-				conn, errs[idx] = d.Dial("tcp", broker.String())
-				if errs[idx] != nil {
-					return
-				}
-				h.brokerConns[broker.String()] = conn
-			}
-			if _, errs[idx] = conn.Write([]byte("")); errs[idx] != nil {
-				conn.Close()
-				h.brokerConns[broker.String()] = nil
-			}
-		}(broker, idx)
-	}
-	wg.Wait()
-
-	var retErr error
-	for _, err := range errs {
-		if err == nil {
-			// if one of them is active we are good.
-			return nil
-		}
-		retErr = err
-	}
-	return retErr
-}
-
 // Target - Kafka target.
 type Target struct {
 	status int32
@@ -139,10 +100,10 @@ type Target struct {
 	initKafkaOnce      once.Init
 	initQueueStoreOnce once.Init
 
-	producer    sarama.SyncProducer
-	kconfig     Config
-	config      *sarama.Config
-	brokerConns map[string]net.Conn
+	client   sarama.Client
+	producer sarama.SyncProducer
+	kconfig  Config
+	config   *sarama.Config
 }
 
 func (h *Target) validate() error {
@@ -273,10 +234,6 @@ func (h *Target) send(entry interface{}) error {
 
 // Init initialize kafka target
 func (h *Target) init() error {
-	if err := h.pingBrokers(); err != nil {
-		return err
-	}
-
 	sconfig := sarama.NewConfig()
 	if h.kconfig.Version != "" {
 		kafkaVersion, err := sarama.ParseKafkaVersion(h.kconfig.Version)
@@ -325,13 +282,23 @@ func (h *Target) init() error {
 		brokers = append(brokers, broker.String())
 	}
 
-	producer, err := sarama.NewSyncProducer(brokers, sconfig)
+	client, err := sarama.NewClient(brokers, sconfig)
 	if err != nil {
 		return err
 	}
 
+	producer, err := sarama.NewSyncProducerFromClient(client)
+	if err != nil {
+		return err
+	}
+	h.client = client
 	h.producer = producer
-	atomic.StoreInt32(&h.status, statusOnline)
+
+	if len(h.client.Brokers()) > 0 {
+		// Refer https://github.com/IBM/sarama/issues/1341
+		atomic.StoreInt32(&h.status, statusOnline)
+	}
+
 	return nil
 }
 
@@ -409,12 +376,7 @@ func (h *Target) Cancel() {
 
 	if h.producer != nil {
 		h.producer.Close()
-	}
-
-	for _, conn := range h.brokerConns {
-		if conn != nil {
-			conn.Close()
-		}
+		h.client.Close()
 	}
 
 	// Wait for messages to be sent...
@@ -425,10 +387,9 @@ func (h *Target) Cancel() {
 // sends log over http to the specified endpoint
 func New(config Config) *Target {
 	target := &Target{
-		logCh:       make(chan interface{}, config.QueueSize),
-		kconfig:     config,
-		status:      statusOffline,
-		brokerConns: make(map[string]net.Conn, len(config.Brokers)),
+		logCh:   make(chan interface{}, config.QueueSize),
+		kconfig: config,
+		status:  statusOffline,
 	}
 	return target
 }
diff --git a/internal/s3select/unused-errors.go b/internal/s3select/unused-errors.go
index 818215e9e123e..d40310af6ab13 100644
--- a/internal/s3select/unused-errors.go
+++ b/internal/s3select/unused-errors.go
@@ -20,11 +20,11 @@
 
 package s3select
 
-///////////////////////////////////////////////////////////////////////
+// /////////////////////////////////////////////////////////////////////
 //
-//             Validation errors.
+//	Validation errors.
 //
-///////////////////////////////////////////////////////////////////////
+// /////////////////////////////////////////////////////////////////////
 func errExpressionTooLong(err error) *s3Error {
 	return &s3Error{
 		code:       "ExpressionTooLong",
@@ -151,11 +151,11 @@ func errUnrecognizedFormatException(err error) *s3Error {
 	}
 }
 
-//////////////////////////////////////////////////////////////////////////////////////
+// ////////////////////////////////////////////////////////////////////////////////////
 //
-//                      SQL parsing errors.
+//	SQL parsing errors.
 //
-//////////////////////////////////////////////////////////////////////////////////////
+// ////////////////////////////////////////////////////////////////////////////////////
 func errLexerInvalidChar(err error) *s3Error {
 	return &s3Error{
 		code:       "LexerInvalidChar",
@@ -498,11 +498,11 @@ func errParseCannotMixSqbAndWildcardInSelectList(err error) *s3Error {
 	}
 }
 
-//////////////////////////////////////////////////////////////////////////////////////
+// ////////////////////////////////////////////////////////////////////////////////////
 //
-//                             CAST() related errors.
+//	CAST() related errors.
 //
-//////////////////////////////////////////////////////////////////////////////////////
+// ////////////////////////////////////////////////////////////////////////////////////
 func errCastFailed(err error) *s3Error {
 	return &s3Error{
 		code:       "CastFailed",
@@ -593,11 +593,11 @@ func errEvaluatorInvalidTimestampFormatPatternSymbol(err error) *s3Error {
 	}
 }
 
-////////////////////////////////////////////////////////////////////////
+// //////////////////////////////////////////////////////////////////////
 //
-//               Generic S3 HTTP handler errors.
+//	Generic S3 HTTP handler errors.
 //
-////////////////////////////////////////////////////////////////////////
+// //////////////////////////////////////////////////////////////////////
 func errBusy(err error) *s3Error {
 	return &s3Error{
 		code:       "Busy",


Test patch: