Function TestClusterAdminDescribeConfig

admin_test.go:878–946

(t *testing.T)

}

func TestClusterAdminDescribeConfig(t *testing.T) {
	seedBroker := NewMockBroker(t, 1)
	defer seedBroker.Close()

	seedBroker.SetHandlerByMap(map[string]MockResponse{
		"MetadataRequest": NewMockMetadataResponse(t).
			SetController(seedBroker.BrokerID()).
			SetBroker(seedBroker.Addr(), seedBroker.BrokerID()),
		"DescribeConfigsRequest": NewMockDescribeConfigsResponse(t),
	})

	tests := []struct {
		saramaVersion   KafkaVersion
		requestVersion  int16
		includeSynonyms bool
	}{
		{V1_0_0_0, 0, false},
		{V1_1_0_0, 1, true},
		{V1_1_1_0, 1, true},
		{V2_0_0_0, 2, true},
		{V2_6_0_0, 3, true},
		{V2_8_0_0, 4, true},
	}
	for _, tt := range tests {
		t.Run(tt.saramaVersion.String(), func(t *testing.T) {
			config := NewTestConfig()
			config.Version = tt.saramaVersion
			admin, err := NewClusterAdmin([]string{seedBroker.Addr()}, config)
			if err != nil {
				t.Fatal(err)
			}
			defer func() {
				_ = admin.Close()
			}()

			resource := ConfigResource{
				Name:        "r1",
				Type:        TopicResource,
				ConfigNames: []string{"my_topic"},
			}

			entries, err := admin.DescribeConfig(resource)
			if err != nil {
				t.Fatal(err)
			}

			history := seedBroker.History()
			describeReq, ok := history[len(history)-1].Request.(*DescribeConfigsRequest)
			if !ok {
				t.Fatal("failed to find DescribeConfigsRequest in mockBroker history")
			}

			if describeReq.Version != tt.requestVersion {
				t.Fatalf(
					"requestVersion %v did not match expected %v",
					describeReq.Version, tt.requestVersion)
			}