fix: anchor Net.ReadTimeout deadline to request send time

broker.go

@@ -1084,10 +1084,16 @@ func (b *Broker) AlterClientQuotas(request *AlterClientQuotasRequest) (*AlterCli
 	return response, nil
 }
 
-// readFull ensures the conn ReadDeadline has been setup before making a
-// call to io.ReadFull
-func (b *Broker) readFull(buf []byte) (n int, err error) {
-	if err := b.conn.SetReadDeadline(time.Now().Add(b.conf.Net.ReadTimeout)); err != nil {
+// readFull sets the read deadline to deadline and then calls io.ReadFull.
+//
+// The deadline is an absolute time.Time computed by the caller as
+// requestTime.Add(Net.ReadTimeout). Anchoring the deadline to the moment
+// the request was sent means a single deadline covers both the header and
+// body reads of one response; resetting it on every call would allow a
+// slow broker to stall indefinitely by returning bytes just before each
+// per-read timeout fires.
+func (b *Broker) readFull(buf []byte, deadline time.Time) (n int, err error) {
+	if err := b.conn.SetReadDeadline(deadline); err != nil {
 		return 0, err
 	}
 
@@ -1337,7 +1343,9 @@ func (b *Broker) responseReceiver() {
 		headerLength := getHeaderLength(promise.response.headerVersion())
 		header := make([]byte, headerLength)
 
-		bytesReadHeader, err := b.readFull(header)
+		deadline := promise.requestTime.Add(b.conf.Net.ReadTimeout)
+
+		bytesReadHeader, err := b.readFull(header, deadline)
 		requestLatency := time.Since(promise.requestTime)
 		if err != nil {
 			b.updateIncomingCommunicationMetrics(bytesReadHeader, requestLatency)
@@ -1364,7 +1372,7 @@ func (b *Broker) responseReceiver() {
 		}
 
 		buf := make([]byte, decodedHeader.length-int32(headerLength)+4)
-		bytesReadBody, err := b.readFull(buf)
+		bytesReadBody, err := b.readFull(buf, deadline)
 		b.updateIncomingCommunicationMetrics(bytesReadHeader+bytesReadBody, requestLatency)
 		if err != nil {
 			dead = err
@@ -1448,8 +1456,9 @@ func (b *Broker) sendAndReceiveApiVersions(v int16) (*ApiVersionsResponse, error
 	// Kafka protocol response structure:
 	// - Message length (4 bytes): Total length of the response excluding this field
 	// - ResponseHeader v0 (4 bytes): Contains correlation ID for request-response matching
+	deadline := requestTime.Add(b.conf.Net.ReadTimeout)
 	header := make([]byte, 8)
-	_, err = b.readFull(header)
+	_, err = b.readFull(header, deadline)
 	if err != nil {
 		b.addRequestInFlightMetrics(-1)
 		Logger.Printf("Failed to read ApiVersionsResponse V%d header from %s: %s\n", v, b.addr, err)
@@ -1461,7 +1470,7 @@ func (b *Broker) sendAndReceiveApiVersions(v int16) (*ApiVersionsResponse, error
 	// correlationID := binary.BigEndian.Uint32(header[4:])
 
 	payload := make([]byte, length-4)
-	n, err := b.readFull(payload)
+	n, err := b.readFull(payload, deadline)
 	if err != nil {
 		b.addRequestInFlightMetrics(-1)
 		Logger.Printf("Failed to read ApiVersionsResponse V%d payload from %s: %s\n", v, b.addr, err)
@@ -1592,8 +1601,9 @@ func (b *Broker) sendAndReceiveSASLHandshake(saslType SASLMechanism, version int
 	}
 	b.correlationID++
 
+	deadline := requestTime.Add(b.conf.Net.ReadTimeout)
 	header := make([]byte, 8) // response header
-	_, err = b.readFull(header)
+	_, err = b.readFull(header, deadline)
 	if err != nil {
 		b.addRequestInFlightMetrics(-1)
 		Logger.Printf("Failed to read SASL handshake header : %s\n", err.Error())
@@ -1602,7 +1612,7 @@ func (b *Broker) sendAndReceiveSASLHandshake(saslType SASLMechanism, version int
 
 	length := binary.BigEndian.Uint32(header[:4])
 	payload := make([]byte, length-4)
-	n, err := b.readFull(payload)
+	n, err := b.readFull(payload, deadline)
 	if err != nil {
 		b.addRequestInFlightMetrics(-1)
 		Logger.Printf("Failed to read SASL handshake payload : %s\n", err.Error())
@@ -1676,7 +1686,7 @@ func (b *Broker) sendAndReceiveSASLPlainAuthV0() error {
 	}
 
 	header := make([]byte, 4)
-	n, err := b.readFull(header)
+	n, err := b.readFull(header, requestTime.Add(b.conf.Net.ReadTimeout))
 	b.updateIncomingCommunicationMetrics(n, time.Since(requestTime))
 	// If the credentials are valid, we would get a 4 byte response filled with null characters.
 	// Otherwise, the broker closes the connection and we get an EOF
@@ -1759,8 +1769,9 @@ func (b *Broker) sendAndReceiveSASLSCRAMv0() error {
 			return err
 		}
 		b.correlationID++
+		deadline := requestTime.Add(b.conf.Net.ReadTimeout)
 		header := make([]byte, 4)
-		_, err = b.readFull(header)
+		_, err = b.readFull(header, deadline)
 		if err != nil {
 			b.addRequestInFlightMetrics(-1)
 			Logger.Printf("Failed to read response header while authenticating with SASL to broker %s: %s\n", b.addr, err.Error())
@@ -1771,7 +1782,7 @@ func (b *Broker) sendAndReceiveSASLSCRAMv0() error {
 			return PacketDecodingError{fmt.Sprintf("SASL response of length %d too large", payloadLength)}
 		}
 		payload := make([]byte, int(payloadLength))
-		n, err := b.readFull(payload)
+		n, err := b.readFull(payload, deadline)
 		if err != nil {
 			b.addRequestInFlightMetrics(-1)
 			Logger.Printf("Failed to read response payload while authenticating with SASL to broker %s: %s\n", b.addr, err.Error())

broker_test.go

@@ -4,8 +4,10 @@ package sarama
 
 import (
 	"bytes"
+	"encoding/binary"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"reflect"
 	"sync"
@@ -15,6 +17,7 @@ import (
 	"github.com/jcmturner/gokrb5/v8/krberror"
 	"github.com/rcrowley/go-metrics"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/net/proxy"
 )
 
 func ExampleBroker() {
@@ -1700,3 +1703,143 @@ func Test_handleThrottledResponse(t *testing.T) {
 		}
 	})
 }
+
+// dripDialer is a test proxy.Dialer whose server half writes a valid, framed
+// Kafka response in two chunks separated by a configurable pause. This lets
+// tests verify that Net.ReadTimeout is applied as a single per-request
+// deadline rather than being reset on every individual read call.
+//
+// The server goroutine writes:
+//
+//	[4-byte length][4-byte correlationID][body...]
+//
+// where body is a minimal, empty MetadataResponse v0 (4-byte broker-count of 0,
+// 4-byte topic-count of 0). It sends the length+correlationID header first,
+// sleeps for pauseBetweenChunks, then sends the body.
+type dripDialer struct {
+	pauseBetweenChunks time.Duration
+	correlationID      int32
+}
+
+func (d dripDialer) Dial(_, _ string) (net.Conn, error) {
+	client, server := net.Pipe()
+
+	go func() {
+		defer server.Close()
+
+		// Read and discard the incoming request so the client can proceed.
+		reqHeader := make([]byte, 4)
+		if _, err := io.ReadFull(server, reqHeader); err != nil {
+			return
+		}
+		reqLen := binary.BigEndian.Uint32(reqHeader)
+		reqBody := make([]byte, reqLen)
+		if _, err := io.ReadFull(server, reqBody); err != nil {
+			return
+		}
+
+		// Minimal MetadataResponse v0: zero brokers, zero topics.
+		body := []byte{
+			0x00, 0x00, 0x00, 0x00, // brokers array length = 0
+			0x00, 0x00, 0x00, 0x00, // topics array length = 0
+		}
+
+		// Frame header: [total-length (4)] [correlationID (4)]
+		// total-length covers everything after itself, so: 4 (corrID) + len(body)
+		frameLen := 4 + len(body)
+		header := make([]byte, 8)
+		binary.BigEndian.PutUint32(header[0:4], uint32(frameLen))
+		binary.BigEndian.PutUint32(header[4:8], uint32(d.correlationID))
+
+		if _, err := server.Write(header); err != nil {
+			return
+		}
+
+		// Pause between header and body to simulate a slow broker.
+		time.Sleep(d.pauseBetweenChunks)
+
+		_, _ = server.Write(body)
+	}()
+
+	return client, nil
+}
+
+// TestReadTimeoutDeadlineSetOncePerResponse verifies that Net.ReadTimeout acts
+// as a per-request budget rather than a per-read deadline.
+//
+// The server sends a valid response in two chunks: the frame header arrives
+// immediately but the body is delayed by pauseBetweenChunks. We configure
+// ReadTimeout to be longer than the total response time, so the request must
+// succeed even though each individual chunk took longer than zero time.
+//
+// Before the fix, readFull reset the deadline on every call, which could allow
+// this test to pass incidentally. The companion test
+// TestReadTimeoutExpiresAcrossChunks covers the failure path.
+func TestReadTimeoutDeadlineSetOncePerResponse(t *testing.T) {
+	t.Parallel()
+
+	const (
+		readTimeout        = 200 * time.Millisecond
+		pauseBetweenChunks = 50 * time.Millisecond // well within the budget
+	)
+
+	conf := NewTestConfig()
+	conf.Net.ReadTimeout = readTimeout
+	conf.Net.Proxy.Enable = true
+	conf.Net.Proxy.Dialer = dripDialer{
+		pauseBetweenChunks: pauseBetweenChunks,
+		correlationID:      1,
+	}
+
+	broker := NewBroker("localhost:0")
+	broker.correlationID = 1
+
+	if err := broker.Open(conf); err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	t.Cleanup(func() { _ = broker.Close() })
+
+	_, err := broker.GetMetadata(&MetadataRequest{})
+	if err != nil {
+		t.Errorf("expected successful response when response arrives within ReadTimeout, got: %v", err)
+	}
+}
+
+// TestReadTimeoutExpiresAcrossChunks verifies that Net.ReadTimeout is enforced
+// across the entire response, not just the first read. The server sends the
+// frame header but then pauses longer than ReadTimeout before sending the body.
+// The broker must surface a timeout error.
+func TestReadTimeoutExpiresAcrossChunks(t *testing.T) {
+	t.Parallel()
+
+	const (
+		readTimeout        = 50 * time.Millisecond
+		pauseBetweenChunks = 200 * time.Millisecond // exceeds the budget
+	)
+
+	conf := NewTestConfig()
+	conf.Net.ReadTimeout = readTimeout
+	conf.Net.Proxy.Enable = true
+	conf.Net.Proxy.Dialer = dripDialer{
+		pauseBetweenChunks: pauseBetweenChunks,
+		correlationID:      1,
+	}
+
+	broker := NewBroker("localhost:0")
+	broker.correlationID = 1
+
+	if err := broker.Open(conf); err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	t.Cleanup(func() { _ = broker.Close() })
+
+	_, err := broker.GetMetadata(&MetadataRequest{})
+
+	var nerr net.Error
+	if !errors.As(err, &nerr) || !nerr.Timeout() {
+		t.Errorf("expected net.Error timeout when body arrives after ReadTimeout, got: %v", err)
+	}
+}
+
+// Compile-time assertion that dripDialer satisfies the proxy.Dialer interface.
+var _ proxy.Dialer = dripDialer{}

config.go

@@ -51,8 +51,17 @@ type Config struct {
 		// All three of the below configurations are similar to the
 		// `socket.timeout.ms` setting in JVM kafka. All of them default
 		// to 30 seconds.
-		DialTimeout  time.Duration // How long to wait for the initial connection.
-		ReadTimeout  time.Duration // How long to wait for a response.
+		DialTimeout time.Duration // How long to wait for the initial connection.
+
+		// ReadTimeout is the maximum duration the broker will wait for a
+		// complete response after a request has been fully sent. The deadline
+		// is set once when the request is sent and applies to the entire
+		// response read (header + body). It must therefore be set to at least
+		// the longest server-side operation you expect, such as
+		// Consumer.Group.Rebalance.Timeout, or reads will time out before the
+		// broker has finished processing.
+		ReadTimeout time.Duration
+
 		WriteTimeout time.Duration // How long to wait for a transmit.
 
 		// ResolveCanonicalBootstrapServers turns each bootstrap broker address