kafka-go/fetch_kip392_test.go at 545fe4e3863c4971920143a3731173ee6332a867 · segmentio/kafka-go · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
package kafka

import (
	"context"
	"net"
	"testing"
	"time"

	"github.com/segmentio/kafka-go/protocol"
	fetchAPI "github.com/segmentio/kafka-go/protocol/fetch"
)

// stubRoundTripper returns a canned response for any request it receives.
type stubRoundTripper struct {
	resp protocol.Message
	err  error
}

func (s *stubRoundTripper) RoundTrip(_ context.Context, _ net.Addr, _ protocol.Message) (protocol.Message, error) {
	return s.resp, s.err
}

// TestClientFetch_PreferredReadReplicaDefaultsMinusOneWithoutRackID asserts
// that Client.Fetch surfaces PreferredReadReplica = -1 when the caller did
// not opt into KIP-392 (req.RackID == ""). Without this guard, a v10 broker
// response (or any response that omits the v11 field) would surface the Go
// zero value 0, indistinguishable from broker id 0 being preferred.
func TestClientFetch_PreferredReadReplicaDefaultsMinusOneWithoutRackID(t *testing.T) {
	stub := &stubRoundTripper{
		resp: &fetchAPI.Response{
			ThrottleTimeMs: 0,
			Topics: []fetchAPI.ResponseTopic{{
				Topic: "topic-x",
				Partitions: []fetchAPI.ResponsePartition{{
					Partition:            0,
					ErrorCode:            0,
					HighWatermark:        100,
					LastStableOffset:     100,
					LogStartOffset:       0,
					PreferredReadReplica: 0, // simulates v10 zero-value field
				}},
			}},
		},
	}

	client := &Client{
		Addr:      TCP("127.0.0.1:9092"),
		Timeout:   time.Second,
		Transport: stub,
	}

	resp, err := client.Fetch(context.Background(), &FetchRequest{
		Topic:     "topic-x",
		Partition: 0,
		Offset:    0,
		MinBytes:  1,
		MaxBytes:  1024,
		MaxWait:   100 * time.Millisecond,
		// RackID intentionally left empty -- caller did not opt in.
	})
	if err != nil {
		t.Fatalf("Client.Fetch: %v", err)
	}
	if resp.PreferredReadReplica != -1 {
		t.Fatalf("PreferredReadReplica: got %d, want -1 when RackID is empty", resp.PreferredReadReplica)
	}
}

// TestClientFetch_PreferredReadReplicaPassThroughWithRackID asserts that when
// the caller did opt into KIP-392 by setting RackID, broker id 0 is passed
// through verbatim (broker 0 may legitimately be the preferred replica).
func TestClientFetch_PreferredReadReplicaPassThroughWithRackID(t *testing.T) {
	stub := &stubRoundTripper{
		resp: &fetchAPI.Response{
			Topics: []fetchAPI.ResponseTopic{{
				Topic: "topic-x",
				Partitions: []fetchAPI.ResponsePartition{{
					Partition:            0,
					HighWatermark:        100,
					PreferredReadReplica: 7,
				}},
			}},
		},
	}

	client := &Client{
		Addr:      TCP("127.0.0.1:9092"),
		Timeout:   time.Second,
		Transport: stub,
	}

	resp, err := client.Fetch(context.Background(), &FetchRequest{
		Topic:     "topic-x",
		Partition: 0,
		Offset:    0,
		MinBytes:  1,
		MaxBytes:  1024,
		MaxWait:   100 * time.Millisecond,
		RackID:    "rack-nl",
	})
	if err != nil {
		t.Fatalf("Client.Fetch: %v", err)
	}
	if resp.PreferredReadReplica != 7 {
		t.Fatalf("PreferredReadReplica: got %d, want 7 (passed through verbatim)", resp.PreferredReadReplica)
	}
}