Server side config check

kafka/client.go

@@ -143,6 +143,36 @@ func (c *Client) UpdateTopic(topic Topic) error {
 	return nil
 }
 
+func (c *Client) CheckConfigDiff(topic Topic) error {
+	log.Printf("[DEBUG] Checking config")
+	broker, err := c.client.Controller()
+	if err != nil {
+		return err
+	}
+
+	r := &sarama.AlterConfigsRequest{
+		Resources:    configToResources(topic),
+		ValidateOnly: true,
+	}
+
+	res, err := broker.AlterConfigs(r)
+
+	if err != nil {
+		return err
+	}
+
+	if err == nil {
+		for _, e := range res.Resources {
+			if e.ErrorCode != int16(sarama.ErrNoError) {
+				log.Printf("[DEBUG] Config error detected %d %s", e.ErrorCode, e.ErrorMsg)
+				return errors.New(e.ErrorMsg)
+			}
+		}
+	}
+	log.Printf("[DEBUG] No config error detected")
+	return nil
+}
+
 func (c *Client) CreateTopic(t Topic) error {
 	broker, err := c.client.Controller()
 	if err != nil {

kafka/lazy_client.go

@@ -24,6 +24,14 @@ func (c *LazyClient) init() error {
 	return err
 }
 
+func (c *LazyClient) CheckConfigDiff(t Topic) error {
+	err := c.init()
+	if err != nil {
+		return err
+	}
+	return c.inner.CheckConfigDiff(t)
+}
+
 func (c *LazyClient) CreateTopic(t Topic) error {
 	err := c.init()
 	if err != nil {

kafka/resource_kafka_topic.go

@@ -3,6 +3,7 @@ package kafka
 import (
 	"fmt"
 	"log"
+	"strings"
 	"time"
 
 	"github.com/hashicorp/terraform-plugin-sdk/helper/resource"
@@ -26,7 +27,7 @@ func kafkaTopicResource() *schema.Resource {
 		Importer: &schema.ResourceImporter{
 			State: schema.ImportStatePassthrough,
 		},
-		CustomizeDiff: customPartitionDiff,
+		CustomizeDiff: customDiff,
 		Schema: map[string]*schema.Schema{
 			"name": {
 				Type:        schema.TypeString,
@@ -200,8 +201,9 @@ func topicRead(d *schema.ResourceData, meta interface{}) error {
 	return nil
 }
 
-func customPartitionDiff(diff *schema.ResourceDiff, v interface{}) error {
+func customDiff(diff *schema.ResourceDiff, meta interface{}) error {
 	log.Printf("[INFO] Checking the diff!")
+	log.Printf("[DEBUG] Resource Id = %s", diff.Id())
 	if diff.HasChange("partitions") {
 		log.Printf("[INFO] Partitions have changed!")
 		o, n := diff.GetChange("partitions")
@@ -212,7 +214,21 @@ func customPartitionDiff(diff *schema.ResourceDiff, v interface{}) error {
 			log.Printf("Partitions decreased from %d to %d. Forcing new resource", oi, ni)
 			diff.ForceNew("partitions")
 		}
-
+	}
+	if diff.HasChange("config") {
+		log.Printf("[INFO] Config has changed!")
+		if diff.Id() != "" {
+			topic := diffToTopic(diff, meta)
+			client := meta.(*LazyClient)
+			err := client.CheckConfigDiff(topic)
+			if err != nil {
+				if strings.Contains(err.Error(), "create a new topic") {
+					diff.ForceNew("config")
+				} else {
+					return err
+				}
+			}
+		}
 	}
 	return nil
 }

kafka/topic.go

@@ -3,6 +3,7 @@ package kafka
 import (
 	"errors"
 	"fmt"
+	"log"
 
 	"github.com/Shopify/sarama"
 	"github.com/hashicorp/terraform-plugin-sdk/helper/schema"
@@ -64,6 +65,31 @@ func isDefault(tc *sarama.ConfigEntry, version int) bool {
 }
 
 func metaToTopic(d *schema.ResourceData, meta interface{}) Topic {
+	log.Printf("meta = %+v", meta)
+	topicName := d.Get("name").(string)
+	partitions := d.Get("partitions").(int)
+	replicationFactor := d.Get("replication_factor").(int)
+	convertedPartitions := int32(partitions)
+	convertedRF := int16(replicationFactor)
+	config := d.Get("config").(map[string]interface{})
+
+	m2 := make(map[string]*string)
+	for key, value := range config {
+		switch value := value.(type) {
+		case string:
+			m2[key] = &value
+		}
+	}
+
+	return Topic{
+		Name:              topicName,
+		Partitions:        convertedPartitions,
+		ReplicationFactor: convertedRF,
+		Config:            m2,
+	}
+}
+
+func diffToTopic(d *schema.ResourceDiff, meta interface{}) Topic {
 	topicName := d.Get("name").(string)
 	partitions := d.Get("partitions").(int)
 	replicationFactor := d.Get("replication_factor").(int)