From 385cecb928e9ec3d9610c7beb223fcd1ed303fd0 Mon Sep 17 00:00:00 2001
From: Travis Bischel <twmb@protonmail.com>
Date: Fri, 4 Dec 2020 15:02:15 -0700
Subject: [PATCH] broker: add warning if connection dies on first read without
 sasl

If the successful reads is >0, or sasl is specified, then we log on
debug level. Otherwise, we log at warn. Hopefully this is not too noisy
if a broker is unreachable.

Kafka makes it impossible to determine if sasl is required if it is
missing; it just kills the connection. We can hint in a warning message
though to potentially look at missing sasl.

Closes #16.
---
 pkg/kgo/broker.go | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/pkg/kgo/broker.go b/pkg/kgo/broker.go
index bcdb936c..b37580e4 100644
--- a/pkg/kgo/broker.go
+++ b/pkg/kgo/broker.go
@@ -882,12 +882,19 @@ func (cxn *brokerCxn) waitResp(pr promisedResp) {
 func (cxn *brokerCxn) handleResps() {
 	defer cxn.die() // always track our death
 
+	var successes uint64
 	for pr := range cxn.resps {
 		raw, err := cxn.readResponse(time.Since(pr.enqueue), pr.resp.Key(), pr.corrID, pr.readTimeout, pr.flexibleHeader)
 		if err != nil {
+			if successes > 0 || len(cxn.b.cl.cfg.sasls) > 0 {
+				cxn.b.cl.cfg.logger.Log(LogLevelDebug, "read from broker errored, killing connection", "addr", cxn.b.addr, "id", cxn.b.meta.NodeID, "successful_reads", successes, "err", err)
+			} else {
+				cxn.b.cl.cfg.logger.Log(LogLevelWarn, "read from broker errored, killing connection after 0 successful responses (is sasl missing?)", "addr", cxn.b.addr, "id", cxn.b.meta.NodeID, "err", err)
+			}
 			pr.promise(nil, err)
 			return
 		}
+		successes++
 		readErr := pr.resp.ReadFrom(raw)
 
 		// If we had no error, we read the response successfully.