From d57e817540ce6d55f7526c1cfb968eeb70075326 Mon Sep 17 00:00:00 2001
From: Conor Mongey <conor@mongey.net>
Date: Thu, 28 Dec 2017 21:18:04 +0000
Subject: [PATCH] Support importing config values

---
 kafka/resource_kafka_topic.go | 11 ++++++++++-
 kafka/utils.go                | 33 +++++++++++++++++++++++++++++++++
 main.tf                       |  2 +-
 3 files changed, 44 insertions(+), 2 deletions(-)

diff --git a/kafka/resource_kafka_topic.go b/kafka/resource_kafka_topic.go
index 9766da4c..21f67614 100644
--- a/kafka/resource_kafka_topic.go
+++ b/kafka/resource_kafka_topic.go
@@ -120,7 +120,8 @@ func topicDelete(d *schema.ResourceData, meta interface{}) error {
 
 func topicRead(d *schema.ResourceData, meta interface{}) error {
 	name := d.Id()
-	c := meta.(*Client).client
+	client := meta.(*Client)
+	c := client.client
 	topics, err := c.Topics()
 
 	if err != nil {
@@ -143,7 +144,15 @@ func topicRead(d *schema.ResourceData, meta interface{}) error {
 					log.Printf("[DEBUG] ReplicationFactor %d from Kafka", r)
 					d.Set("replication_factor", r)
 				}
+				configToSave, err := ConfigForTopic(t, client.config.Brokers)
+				if err != nil {
+					log.Printf("[ERROR] Could not get config for topic %s: %s", t, err)
+					return err
+				}
+
+				d.Set("config", configToSave)
 			}
+
 			return nil
 		}
 	}
diff --git a/kafka/utils.go b/kafka/utils.go
index dedc380c..2e855815 100644
--- a/kafka/utils.go
+++ b/kafka/utils.go
@@ -57,6 +57,39 @@ type topicConfig struct {
 	Config            map[string]*string
 }
 
+func ConfigForTopic(topic string, brokers []string) (map[string]string, error) {
+	confToSave := map[string]string{}
+	request := &samara.DescribeConfigsRequest{
+		Resources: []*samara.Resource{
+			&samara.Resource{
+				T:           samara.TopicResource,
+				Name:        topic,
+				ConfigNames: []string{"segment.ms"},
+			},
+		},
+	}
+
+	broker, err := AvailableBrokerFromList(brokers)
+	if err != nil {
+		return confToSave, err
+	}
+	cr, err := broker.DescribeConfigs(request)
+	if err != nil {
+		return confToSave, err
+	}
+
+	if len(cr.Resources) > 0 && len(cr.Resources[0].Configs) > 0 {
+		for _, conf := range cr.Resources[0].Configs {
+			if conf.Default {
+				continue
+			}
+			log.Printf("[DEBUG] configs %s", conf.Name)
+			log.Printf("[DEBUG] configs %s", conf.Value)
+			confToSave[conf.Name] = conf.Value
+		}
+	}
+	return confToSave, nil
+}
 func metaToTopicConfig(d *schema.ResourceData, meta interface{}) topicConfig {
 	topicName := d.Get("name").(string)
 	partitions := d.Get("partitions").(int)
diff --git a/main.tf b/main.tf
index 221d0a49..9712870c 100644
--- a/main.tf
+++ b/main.tf
@@ -14,6 +14,6 @@ resource "kafka_topic" "foo2" {
   partitions         = 1
 
   config = {
-    "segment.ms" = "10000"
+    "segment.ms" = "20000"
   }
 }