diff --git a/broker.go b/broker.go index f0dfcff32..9587e6e18 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,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()) 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