diff --git a/metadata_refresh.go b/metadata_refresh.go new file mode 100644 index 00000000..ad8cbb39 --- /dev/null +++ b/metadata_refresh.go @@ -0,0 +1,89 @@ +package kafka + +import ( + "errors" + + "github.com/segmentio/kafka-go/protocol" + fetchAPI "github.com/segmentio/kafka-go/protocol/fetch" + produceAPI "github.com/segmentio/kafka-go/protocol/produce" +) + +// isStaleMetadataError reports whether a kafka error code indicates that the +// cached cluster metadata is likely out of date and should be refreshed. +func isStaleMetadataError(err Error) bool { + switch err { + case UnknownTopicOrPartition, + LeaderNotAvailable, + NotLeaderForPartition, + ReplicaNotAvailable, + BrokerNotAvailable, + KafkaStorageError, + FencedLeaderEpoch, + UnknownLeaderEpoch, + UnknownTopicID: + return true + default: + return false + } +} + +// errorRequiresMetadataRefresh reports whether a transport-level error returned +// by a failed round trip likely indicates stale cluster metadata: transient +// network errors, routing failures over a stale layout (unknown topic, partition +// or leader), or kafka errors that signal the metadata is out of date. +func errorRequiresMetadataRefresh(err error) bool { + if isTransientNetworkError(err) { + return true + } + // Routing failures raised while resolving the target broker from the cached + // cluster layout indicate the layout is stale and should be refreshed. + if errors.Is(err, protocol.ErrNoTopic) || + errors.Is(err, protocol.ErrNoPartition) || + errors.Is(err, protocol.ErrNoLeader) { + return true + } + var kafkaErr Error + if errors.As(err, &kafkaErr) { + return isStaleMetadataError(kafkaErr) + } + return false +} + +// responseRequiresMetadataRefresh inspects a successful response body and +// reports whether any partition (or, for Fetch, the session-level error code) +// reported a stale-metadata error. Only response types routed to specific +// partition leaders are inspected, since those are impacted by leader changes. +func responseRequiresMetadataRefresh(r Response) bool { + switch resp := r.(type) { + case *produceAPI.Response: + if resp == nil { + return false + } + for i := range resp.Topics { + partitions := resp.Topics[i].Partitions + for j := range partitions { + if isStaleMetadataError(Error(partitions[j].ErrorCode)) { + return true + } + } + } + case *fetchAPI.Response: + if resp == nil { + return false + } + // Fetch v7+ may report a session-level error in addition to the + // per-partition error codes. + if isStaleMetadataError(Error(resp.ErrorCode)) { + return true + } + for i := range resp.Topics { + partitions := resp.Topics[i].Partitions + for j := range partitions { + if isStaleMetadataError(Error(partitions[j].ErrorCode)) { + return true + } + } + } + } + return false +} diff --git a/metadata_refresh_test.go b/metadata_refresh_test.go new file mode 100644 index 00000000..ffbc8bbf --- /dev/null +++ b/metadata_refresh_test.go @@ -0,0 +1,119 @@ +package kafka + +import ( + "io" + "testing" + + fetchAPI "github.com/segmentio/kafka-go/protocol/fetch" + meta "github.com/segmentio/kafka-go/protocol/metadata" + produceAPI "github.com/segmentio/kafka-go/protocol/produce" +) + +func TestIsStaleMetadataError(t *testing.T) { + staleErrors := []Error{ + UnknownTopicOrPartition, + LeaderNotAvailable, + NotLeaderForPartition, + ReplicaNotAvailable, + BrokerNotAvailable, + KafkaStorageError, + FencedLeaderEpoch, + UnknownLeaderEpoch, + UnknownTopicID, + } + + for _, err := range staleErrors { + if !isStaleMetadataError(err) { + t.Errorf("expected %v (%d) to be classified as a stale metadata error", err, int(err)) + } + } + + nonStaleErrors := []Error{ + Unknown, + InvalidMessage, + RequestTimedOut, + MessageSizeTooLarge, + TopicAuthorizationFailed, + InvalidRequiredAcks, + } + + for _, err := range nonStaleErrors { + if isStaleMetadataError(err) { + t.Errorf("did not expect %v (%d) to be classified as a stale metadata error", err, int(err)) + } + } +} + +func TestErrorRequiresMetadataRefresh(t *testing.T) { + tests := []struct { + scenario string + err error + want bool + }{ + {scenario: "nil", err: nil, want: false}, + {scenario: "transient network", err: io.ErrUnexpectedEOF, want: true}, + {scenario: "broker not available", err: BrokerNotAvailable, want: true}, + {scenario: "not leader for partition", err: NotLeaderForPartition, want: true}, + {scenario: "non retriable kafka error", err: TopicAuthorizationFailed, want: false}, + } + + for _, test := range tests { + t.Run(test.scenario, func(t *testing.T) { + if got := errorRequiresMetadataRefresh(test.err); got != test.want { + t.Errorf("errorRequiresMetadataRefresh(%v) = %v, want %v", test.err, got, test.want) + } + }) + } +} + +func TestResponseRequiresMetadataRefresh(t *testing.T) { + produceWith := func(code Error) *produceAPI.Response { + return &produceAPI.Response{ + Topics: []produceAPI.ResponseTopic{{ + Topic: "topic", + Partitions: []produceAPI.ResponsePartition{{ + Partition: 0, + ErrorCode: int16(code), + }}, + }}, + } + } + + fetchWith := func(topLevel, partition Error) *fetchAPI.Response { + return &fetchAPI.Response{ + ErrorCode: int16(topLevel), + Topics: []fetchAPI.ResponseTopic{{ + Topic: "topic", + Partitions: []fetchAPI.ResponsePartition{{ + Partition: 0, + ErrorCode: int16(partition), + }}, + }}, + } + } + + tests := []struct { + scenario string + resp Response + want bool + }{ + {scenario: "produce stale", resp: produceWith(NotLeaderForPartition), want: true}, + {scenario: "produce ok", resp: produceWith(0), want: false}, + {scenario: "produce non-stale error", resp: produceWith(InvalidMessageSize), want: false}, + {scenario: "fetch partition stale", resp: fetchWith(0, LeaderNotAvailable), want: true}, + {scenario: "fetch session stale", resp: fetchWith(FencedLeaderEpoch, 0), want: true}, + {scenario: "fetch ok", resp: fetchWith(0, 0), want: false}, + {scenario: "unrelated response", resp: &meta.Response{}, want: false}, + {scenario: "nil interface", resp: nil, want: false}, + {scenario: "typed nil produce", resp: (*produceAPI.Response)(nil), want: false}, + {scenario: "typed nil fetch", resp: (*fetchAPI.Response)(nil), want: false}, + } + + for _, test := range tests { + t.Run(test.scenario, func(t *testing.T) { + if got := responseRequiresMetadataRefresh(test.resp); got != test.want { + t.Errorf("responseRequiresMetadataRefresh(%T) = %v, want %v", test.resp, got, test.want) + } + }) + } +} diff --git a/transport.go b/transport.go index 685bdddb..1bd9edce 100644 --- a/transport.go +++ b/transport.go @@ -296,6 +296,9 @@ type connPool struct { ready event // triggered after the first metadata update wake chan event // used to force metadata updates cancel context.CancelFunc + // Unix-nanos timestamp of the last error-triggered metadata refresh + // request, used to throttle refreshes under cascading failures. + lastMetadataRefresh int64 // Mutable fields of the connection pool, access must be synchronized. mutex sync.RWMutex conns map[int32]*connGroup // data connections used for produce/fetch/etc... @@ -398,9 +401,20 @@ func (p *connPool) roundTrip(ctx context.Context, req Request) (Response, error) r, err := response.await(ctx) if err != nil { + // A communication or routing error likely means the cached cluster + // view is stale; refresh it so retries reach the right brokers. + if errorRequiresMetadataRefresh(err) { + p.requestMetadataUpdate() + } return r, err } + // A successful response may still report per-partition errors that + // indicate stale metadata (e.g. the partition leader moved). + if responseRequiresMetadataRefresh(r) { + p.requestMetadataUpdate() + } + switch resp := r.(type) { case *createtopics.Response: // Force an update of the metadata when adding topics, @@ -442,6 +456,31 @@ func (p *connPool) roundTrip(ctx context.Context, req Request) (Response, error) return r, nil } +// metadataRefreshThrottle is the minimum interval between two error-triggered +// metadata refreshes. It bounds the load on the cluster when many round trips +// fail in cascade. +const metadataRefreshThrottle = time.Second + +// requestMetadataUpdate triggers an asynchronous refresh of the cached cluster +// metadata without blocking the caller. The discover goroutine performs the +// refresh on its next iteration. Requests are coalesced and throttled to at +// most one per metadataRefreshThrottle to avoid storming the brokers under +// cascading failures. +func (p *connPool) requestMetadataUpdate() { + now := time.Now().UnixNano() + last := atomic.LoadInt64(&p.lastMetadataRefresh) + if now-last < int64(metadataRefreshThrottle) { + return + } + if !atomic.CompareAndSwapInt64(&p.lastMetadataRefresh, last, now) { + return + } + select { + case p.wake <- make(event): + default: + } +} + // refreshMetadata forces an update of the cached cluster metadata, and waits // for the given list of topics to appear. This waiting mechanism is necessary // to account for the fact that topic creation is asynchronous in kafka, and diff --git a/transport_test.go b/transport_test.go index eaae3329..5b6bebb2 100644 --- a/transport_test.go +++ b/transport_test.go @@ -5,12 +5,15 @@ import ( "crypto/tls" "errors" "net" + "sync" + "sync/atomic" "testing" "time" "github.com/segmentio/kafka-go/protocol" "github.com/segmentio/kafka-go/protocol/createtopics" meta "github.com/segmentio/kafka-go/protocol/metadata" + produceAPI "github.com/segmentio/kafka-go/protocol/produce" ) func TestIssue477(t *testing.T) { @@ -304,3 +307,150 @@ func TestIssue806(t *testing.T) { t.Fatalf("expected a meta.Response but got %T", r) } } + +// TestRoundTripRefreshesMetadataOnStaleError verifies that a Produce response +// carrying a stale-metadata error code (e.g. NotLeaderForPartition) triggers an +// asynchronous metadata refresh so the next attempt can be routed to the new +// partition leader. +func TestRoundTripRefreshesMetadataOnStaleError(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) + defer cancel() + + const topic = "topic" + + ready := make(chan struct{}) + close(ready) + + // A buffered wake channel lets requestMetadataUpdate's non-blocking send + // succeed without a concurrent reader, making the assertion deterministic. + wake := make(chan event, 1) + + // Resolve the produce request with a NotLeaderForPartition error. + requests := make(chan connRequest, 1) + defer close(requests) + go func() { + request := <-requests + request.res.resolve(&produceAPI.Response{ + Topics: []produceAPI.ResponseTopic{{ + Topic: topic, + Partitions: []produceAPI.ResponsePartition{{ + Partition: 0, + ErrorCode: int16(NotLeaderForPartition), + }}, + }}, + }) + }() + + pool := &connPool{ + ready: ready, + wake: wake, + conns: map[int32]*connGroup{}, + } + + pool.setState(connPoolState{ + layout: protocol.Cluster{ + Brokers: map[int32]protocol.Broker{ + 0: {ID: 0}, + }, + Topics: map[string]protocol.Topic{ + topic: { + Name: topic, + Partitions: map[int32]protocol.Partition{ + 0: {ID: 0, Leader: 0}, + }, + }, + }, + }, + }) + + // Produce requests are routed to the partition leader (broker 0). + pool.conns[0] = &connGroup{ + pool: pool, + broker: Broker{ID: 0}, + idleConns: []*conn{ + { + reqs: requests, + }, + }, + } + + r, err := pool.roundTrip(ctx, &produceAPI.Request{ + Topics: []produceAPI.RequestTopic{{ + Topic: topic, + Partitions: []produceAPI.RequestPartition{{ + Partition: 0, + }}, + }}, + }) + if err != nil { + t.Fatalf("unexpected error from roundTrip: %v", err) + } + if _, ok := r.(*produceAPI.Response); !ok { + t.Fatalf("expected a produce.Response but got %T", r) + } + + select { + case <-wake: + // expected: a metadata refresh was requested. + default: + t.Fatal("expected a metadata refresh to be requested after a stale-metadata produce error") + } +} + +// TestRequestMetadataUpdateNonBlocking verifies that requestMetadataUpdate never +// blocks the caller, even when no consumer is reading from the wake channel. +func TestRequestMetadataUpdateNonBlocking(t *testing.T) { + pool := &connPool{ + wake: make(chan event), // unbuffered, with no reader + } + + done := make(chan struct{}) + go func() { + pool.requestMetadataUpdate() + close(done) + }() + + select { + case <-done: + // expected: the call returned without blocking. + case <-time.After(time.Second): + t.Fatal("requestMetadataUpdate blocked when no consumer was reading the wake channel") + } +} + +// TestRequestMetadataUpdateThrottled verifies that consecutive refresh requests +// are throttled: only one wake is emitted per metadataRefreshThrottle window, +// even when many callers race, bounding the load on the cluster. +func TestRequestMetadataUpdateThrottled(t *testing.T) { + // Buffered so each accepted request is recorded without a reader. + wake := make(chan event, 8) + pool := &connPool{wake: wake} + + const callers = 50 + var wg sync.WaitGroup + wg.Add(callers) + for i := 0; i < callers; i++ { + go func() { + defer wg.Done() + pool.requestMetadataUpdate() + }() + } + wg.Wait() + + if got := len(wake); got != 1 { + t.Fatalf("expected a single metadata refresh within the throttle window, got %d", got) + } + + // A second burst within the window must be throttled out. + pool.requestMetadataUpdate() + if got := len(wake); got != 1 { + t.Fatalf("expected refreshes within the throttle window to be dropped, got %d", got) + } + + // Simulating an elapsed window allows a new refresh. + atomic.StoreInt64(&pool.lastMetadataRefresh, time.Now().Add(-2*metadataRefreshThrottle).UnixNano()) + pool.requestMetadataUpdate() + if got := len(wake); got != 2 { + t.Fatalf("expected a new metadata refresh after the throttle window elapsed, got %d", got) + } +}