fix: add broker connection check in metadata refresh loop. Closes bro…

…ken broker connections

broker.go

@@ -13,6 +13,7 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"syscall"
 	"time"
 
 	"github.com/rcrowley/go-metrics"
@@ -1765,3 +1766,19 @@ func validServerNameTLS(addr string, cfg *tls.Config) *tls.Config {
 	c.ServerName = sn
 	return c
 }
+
+// isBroken checks if the connection on the broker is still working. For this purpose it sends
+// an ApiVersions request. If the connection returns an EOF, ECONNRESET or EPIPE error, it is broken.
+func (b *Broker) isBroken() bool {
+	_, err := b.ApiVersions(&ApiVersionsRequest{
+		ClientSoftwareName:    defaultClientSoftwareName,
+		ClientSoftwareVersion: version(),
+	})
+
+	if err == io.EOF || errors.Is(err, syscall.ECONNRESET) || errors.Is(err, syscall.EPIPE) {
+		DebugLogger.Printf("Connection is broken on broker %s: %v", b.addr, err)
+		return true
+	}
+
+	return false
+}

client.go

@@ -987,6 +987,8 @@ func (client *client) refreshMetadata() error {
 		}
 	}
 
+	client.checkAndCloseBrokenBroker()
+
 	if err := client.RefreshMetadata(topics...); err != nil {
 		return err
 	}
@@ -1090,6 +1092,25 @@ func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int,
 	return retry(error)
 }
 
+func (client *client) checkAndCloseBrokenBroker() {
+	client.lock.RLock()
+	defer client.lock.RUnlock()
+
+	DebugLogger.Println("check broker connections and close broken brokers")
+
+	for _, broker := range client.brokers {
+		broker.lock.Lock()
+		if broker.conn != nil {
+			broker.lock.Unlock()
+			if broker.isBroken() {
+				_ = broker.Close()
+			}
+		} else {
+			broker.lock.Unlock()
+		}
+	}
+}
+
 // if no fatal error, returns a list of topics that need retrying due to ErrLeaderNotAvailable
 func (client *client) updateMetadata(data *MetadataResponse, allKnownMetaData bool) (retry bool, err error) {
 	if client.Closed() {