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
|
package sarama
import "testing"
func TestRestrictApiVersionLowersVersionToBrokerMax(t *testing.T) {
request := NewMetadataRequest(V2_8_0_0, []string{"test-topic"})
if request.version() != 10 {
t.Errorf("Expected MetadataRequest version to be 10, got %d", request.version())
}
brokerVersions := apiVersionMap{
apiKeyMetadata: &apiVersionRange{
minVersion: 0,
maxVersion: 8,
},
}
err := restrictApiVersion(request, brokerVersions)
if err != nil {
t.Errorf("restrictApiVersion returned unexpected error: %v", err)
}
if request.version() != 8 {
t.Errorf("Expected version to be restricted to 8, got %d", request.version())
}
}
func TestRestrictApiVersionLeavesVersionUnchangedWhenWithinRange(t *testing.T) {
request := NewMetadataRequest(V2_4_0_0, []string{"test-topic"})
originalVersion := request.version()
if originalVersion != 9 {
t.Errorf("Expected MetadataRequest version to be 9, got %d", originalVersion)
}
brokerVersions := apiVersionMap{
apiKeyMetadata: &apiVersionRange{
minVersion: 0,
maxVersion: 10,
},
}
err := restrictApiVersion(request, brokerVersions)
if err != nil {
t.Errorf("restrictApiVersion returned unexpected error: %v", err)
}
if request.version() != originalVersion {
t.Errorf("Expected version to remain %d, got %d", originalVersion, request.version())
}
}
func TestRestrictApiVersionDoesNotRaiseVersionBeyondUserSetMax(t *testing.T) {
// the Kafka version comes from conf.Version, which is the user-set max Kafka API version to use
request := NewMetadataRequest(V0_10_0_0, []string{"test-topic"})
if request.version() != 1 {
t.Errorf("Expected MetadataRequest version to be 1, got %d", request.version())
}
// broker doesn't support versions below 5
brokerVersions := apiVersionMap{
apiKeyMetadata: &apiVersionRange{
minVersion: 5,
maxVersion: 10,
},
}
// we expect the user's preference to be respected even when it's below the broker's minimum
err := restrictApiVersion(request, brokerVersions)
if err != nil {
t.Errorf("restrictApiVersion returned unexpected error: %v", err)
}
if request.version() != 1 {
t.Errorf("Expected version to be set to minimum 1, got %d", request.version())
}
}
func TestRestrictApiVersionDoesNothingIfBrokerVersionRangeMissing(t *testing.T) {
request := NewMetadataRequest(V2_8_0_0, []string{"test-topic"})
originalVersion := request.version()
brokerVersions := apiVersionMap{
// no entry for apiKeyMetadata
}
err := restrictApiVersion(request, brokerVersions)
if err != nil {
t.Errorf("restrictApiVersion returned unexpected error: %v", err)
}
if request.version() != originalVersion {
t.Errorf("Expected version to remain %d, got %d", originalVersion, request.version())
}
}
|
