sarama/api_versions_test.go at main · IBM/sarama · 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
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())
	}
}