From b1945ce7d38681cdc3c1707464c59925a16db4dd Mon Sep 17 00:00:00 2001
From: Conor Mongey <conor@mongey.net>
Date: Fri, 8 Feb 2019 17:28:12 +0000
Subject: [PATCH] Get latest version supported for DescribeConfigsRequest

---
 kafka/client.go      | 58 +++++++++++++++++++++++++++++++++++++++-----
 kafka/client_test.go | 28 +++++++++++++++++++++
 2 files changed, 80 insertions(+), 6 deletions(-)
 create mode 100644 kafka/client_test.go

diff --git a/kafka/client.go b/kafka/client.go
index 404a3af1..77b38759 100644
--- a/kafka/client.go
+++ b/kafka/client.go
@@ -19,9 +19,10 @@ type TopicMissingError struct {
 func (e TopicMissingError) Error() string { return e.msg }
 
 type Client struct {
-	client      sarama.Client
-	kafkaConfig *sarama.Config
-	config      *Config
+	client        sarama.Client
+	kafkaConfig   *sarama.Config
+	config        *Config
+	supportedAPIs map[int]int
 }
 
 type Config struct {
@@ -56,11 +57,41 @@ func NewClient(config *Config) (*Client, error) {
 		return nil, err
 	}
 
-	return &Client{
+	client := &Client{
 		client:      c,
 		config:      config,
 		kafkaConfig: kc,
-	}, kc.Validate()
+	}
+
+	err = kc.Validate()
+	if err != nil {
+		return client, err
+	}
+
+	client.populateAPIVersions()
+	return client, nil
+}
+
+func (c *Client) populateAPIVersions() {
+	log.Printf("[DEBUG] retrieving supported APIs from broker")
+	broker, err := c.client.Controller()
+	if err != nil {
+		log.Printf("[ERROR] Unable to populate supported API versions. Error retrieving controller: %s", err)
+		return
+	}
+
+	resp, err := broker.ApiVersions(&sarama.ApiVersionsRequest{})
+	if err != nil {
+		log.Printf("[ERROR] Unable to populate supported API versions. %s", err)
+		return
+	}
+
+	m := map[int]int{}
+	for _, v := range resp.ApiVersions {
+		log.Printf("[TRACE] API key %d. Min %d, Max %d", v.ApiKey, v.MinVersion, v.MaxVersion)
+		m[int(v.ApiKey)] = int(v.MaxVersion)
+	}
+	c.supportedAPIs = m
 }
 
 func (c *Client) DeleteTopic(t string) error {
@@ -237,10 +268,25 @@ func (client *Client) ReadTopic(name string) (Topic, error) {
 	return topic, err
 }
 
+func (c *Client) getDescribeConfigAPIVersion() int16 {
+	return int16(c.versionForKey(32, 1))
+}
+
+func (c *Client) versionForKey(apiKey, wantedMaxVersion int) int {
+	if maxSupportedVersion, ok := c.supportedAPIs[apiKey]; ok {
+		if maxSupportedVersion < wantedMaxVersion {
+			return maxSupportedVersion
+		}
+		return wantedMaxVersion
+	}
+
+	return 0
+}
+
 func (c *Client) topicConfig(topic string) (map[string]*string, error) {
 	conf := map[string]*string{}
 	request := &sarama.DescribeConfigsRequest{
-		Version: 1,
+		Version: c.getDescribeConfigAPIVersion(),
 		Resources: []*sarama.ConfigResource{
 			{
 				Type: sarama.TopicResource,
diff --git a/kafka/client_test.go b/kafka/client_test.go
new file mode 100644
index 00000000..6fd3577e
--- /dev/null
+++ b/kafka/client_test.go
@@ -0,0 +1,28 @@
+package kafka
+
+import "testing"
+
+func Test_ClientAPIVersion(t *testing.T) {
+	// Default to 0
+	client := &Client{}
+	maxVersion := client.versionForKey(32, 1)
+	if maxVersion != 0 {
+		t.Errorf("Got %d, expected %d", maxVersion, 0)
+	}
+
+	// use the max version the broker supports, if it's less than the requested
+	// version
+	client.supportedAPIs = map[int]int{32: 0}
+	maxVersion = client.versionForKey(32, 1)
+	if maxVersion != 0 {
+		t.Errorf("Got %d, expected %d", maxVersion, 0)
+	}
+
+	// while the broker supports 2, terraform-provider-kafka only supports 1, so
+	// use that
+	client.supportedAPIs = map[int]int{32: 2}
+	maxVersion = client.versionForKey(32, 1)
+	if maxVersion != 1 {
+		t.Errorf("Got %d, expected %d", maxVersion, 1)
+	}
+}