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
108
109
110
|
//go:build !functional
package sarama
import "testing"
var (
offsetRequestNoBlocksV1 = []byte{
0xFF, 0xFF, 0xFF, 0xFF,
0x00, 0x00, 0x00, 0x00,
}
offsetRequestNoBlocksV2 = []byte{
0xFF, 0xFF, 0xFF, 0xFF,
0x00, 0x00, 0x00, 0x00,
0x00,
}
offsetRequestOneBlock = []byte{
0xFF, 0xFF, 0xFF, 0xFF,
0x00, 0x00, 0x00, 0x01,
0x00, 0x03, 'f', 'o', 'o',
0x00, 0x00, 0x00, 0x01,
0x00, 0x00, 0x00, 0x04,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01,
0x00, 0x00, 0x00, 0x02,
}
offsetRequestOneBlockV1 = []byte{
0xFF, 0xFF, 0xFF, 0xFF,
0x00, 0x00, 0x00, 0x01,
0x00, 0x03, 'b', 'a', 'r',
0x00, 0x00, 0x00, 0x01,
0x00, 0x00, 0x00, 0x04,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01,
}
offsetRequestOneBlockReadCommittedV2 = []byte{
0xFF, 0xFF, 0xFF, 0xFF,
0x01, 0x00, 0x00, 0x00, 0x01,
0x00, 0x03, 'b', 'a', 'r',
0x00, 0x00, 0x00, 0x01,
0x00, 0x00, 0x00, 0x04,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01,
}
offsetRequestReplicaID = []byte{
0x00, 0x00, 0x00, 0x2a,
0x00, 0x00, 0x00, 0x00,
}
offsetRequestV4 = []byte{
0xff, 0xff, 0xff, 0xff, // replicaID
0x01, // IsolationLevel
0x00, 0x00, 0x00, 0x01,
0x00, 0x04,
0x64, 0x6e, 0x77, 0x65, // topic name
0x00, 0x00, 0x00, 0x01,
0x00, 0x00, 0x00, 0x09, // partitionID
0xff, 0xff, 0xff, 0xff, // leader epoch
0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, // timestamp
}
)
func TestOffsetRequest(t *testing.T) {
request := new(OffsetRequest)
testRequest(t, "no blocks", request, offsetRequestNoBlocksV1)
request.AddBlock("foo", 4, 1, 2)
testRequest(t, "one block", request, offsetRequestOneBlock)
}
func TestOffsetRequestV1(t *testing.T) {
request := new(OffsetRequest)
request.Version = 1
testRequest(t, "no blocks", request, offsetRequestNoBlocksV1)
request.AddBlock("bar", 4, 1, 2) // Last argument is ignored for V1
testRequest(t, "one block", request, offsetRequestOneBlockV1)
}
func TestOffsetRequestV2(t *testing.T) {
request := new(OffsetRequest)
request.Version = 2
testRequest(t, "no blocks", request, offsetRequestNoBlocksV2)
request.IsolationLevel = ReadCommitted
request.AddBlock("bar", 4, 1, 2) // Last argument is ignored for V1
testRequest(t, "one block", request, offsetRequestOneBlockReadCommittedV2)
}
func TestOffsetRequestReplicaID(t *testing.T) {
request := new(OffsetRequest)
replicaID := int32(42)
request.SetReplicaID(replicaID)
if found := request.ReplicaID(); found != replicaID {
t.Errorf("replicaID: expected %v, found %v", replicaID, found)
}
testRequest(t, "with replica ID", request, offsetRequestReplicaID)
}
func TestOffsetRequestV4(t *testing.T) {
request := new(OffsetRequest)
request.Version = 4
request.IsolationLevel = ReadCommitted
request.AddBlock("dnwe", 9, -1, -1)
testRequest(t, "V4", request, offsetRequestV4)
}
|
