From b56dfc7de3d0b1eec87805ed4cbd6d0a411703b4 Mon Sep 17 00:00:00 2001 From: Hugo Villarreal Date: Wed, 15 Jul 2026 10:51:11 -0500 Subject: [PATCH 1/2] fix: anchor Net.ReadTimeout deadline to request send time Previously, readFull called SetReadDeadline(time.Now().Add(ReadTimeout)) on every call. In responseReceiver this meant the deadline was reset between reading the header and reading the body, so a broker that dribbled bytes just within the per-read window could stall indefinitely without ever triggering a timeout. Fix by computing the deadline once per request as requestTime.Add(Net.ReadTimeout) and passing it through to readFull. The same pattern is applied to the ApiVersions negotiation, SASL handshake, and SCRAM paths which all have their own requestTime already. The config doc comment is updated to make clear that ReadTimeout is a per-request budget and must be set to at least the longest expected server-side operation (e.g. Consumer.Group.Rebalance.Timeout). Fixes #2820 Signed-off-by: Hugo Villarreal --- broker.go | 36 ++++++++----- broker_test.go | 143 +++++++++++++++++++++++++++++++++++++++++++++++++ config.go | 13 ++++- 3 files changed, 177 insertions(+), 15 deletions(-) diff --git a/broker.go b/broker.go index f0dfcff32..3af2d0373 100644 --- a/broker.go +++ b/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,8 @@ 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 +1371,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 +1455,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 +1469,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 +1600,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 +1611,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 +1685,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 +1768,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 +1781,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()) diff --git a/broker_test.go b/broker_test.go index 91ddef788..52c4cf7b3 100644 --- a/broker_test.go +++ b/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{} diff --git a/config.go b/config.go index b4530f23c..137a065a9 100644 --- a/config.go +++ b/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 From 2722d463a337d9eb7c9928793b6b75d57091b96a Mon Sep 17 00:00:00 2001 From: Hugo Villarreal Date: Thu, 16 Jul 2026 09:40:18 -0500 Subject: [PATCH 2/2] fixup! fix: anchor Net.ReadTimeout deadline to request send time Set deadline off in its own paragraph as suggested in review, making it visually clear that the value is shared across both readFull calls rather than being a convenience variable for a single call. Signed-off-by: Hugo Villarreal --- broker.go | 1 + 1 file changed, 1 insertion(+) diff --git a/broker.go b/broker.go index 3af2d0373..9587e6e18 100644 --- a/broker.go +++ b/broker.go @@ -1344,6 +1344,7 @@ func (b *Broker) responseReceiver() { header := make([]byte, headerLength) deadline := promise.requestTime.Add(b.conf.Net.ReadTimeout) + bytesReadHeader, err := b.readFull(header, deadline) requestLatency := time.Since(promise.requestTime) if err != nil {