Merge pull request #8740 from terraform-providers/f-aws_msk_configura…

…tion

New Resource: aws_msk_configuration

aws/provider.go

@@ -563,6 +563,7 @@ func Provider() terraform.ResourceProvider {
 			"aws_media_store_container":                               resourceAwsMediaStoreContainer(),
 			"aws_media_store_container_policy":                        resourceAwsMediaStoreContainerPolicy(),
 			"aws_msk_cluster":                                         resourceAwsMskCluster(),
+			"aws_msk_configuration":                                   resourceAwsMskConfiguration(),
 			"aws_nat_gateway":                                         resourceAwsNatGateway(),
 			"aws_network_acl":                                         resourceAwsNetworkAcl(),
 			"aws_default_network_acl":                                 resourceAwsDefaultNetworkAcl(),

aws/resource_aws_msk_configuration.go

@@ -0,0 +1,137 @@
+package aws
+
+import (
+	"fmt"
+	"log"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/kafka"
+	"github.com/hashicorp/terraform/helper/schema"
+)
+
+func resourceAwsMskConfiguration() *schema.Resource {
+	return &schema.Resource{
+		Create: resourceAwsMskConfigurationCreate,
+		Read:   resourceAwsMskConfigurationRead,
+		Delete: schema.Noop,
+
+		Importer: &schema.ResourceImporter{
+			State: schema.ImportStatePassthrough,
+		},
+
+		Schema: map[string]*schema.Schema{
+			"arn": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+			"description": {
+				Type:     schema.TypeString,
+				Optional: true,
+				ForceNew: true,
+			},
+			"kafka_versions": {
+				Type:     schema.TypeSet,
+				Required: true,
+				ForceNew: true,
+				Elem: &schema.Schema{
+					Type: schema.TypeString,
+				},
+			},
+			"latest_revision": {
+				Type:     schema.TypeInt,
+				Computed: true,
+			},
+			"name": {
+				Type:     schema.TypeString,
+				Required: true,
+				ForceNew: true,
+			},
+			"server_properties": {
+				Type:     schema.TypeString,
+				Required: true,
+				ForceNew: true,
+			},
+		},
+	}
+}
+
+func resourceAwsMskConfigurationCreate(d *schema.ResourceData, meta interface{}) error {
+	conn := meta.(*AWSClient).kafkaconn
+
+	input := &kafka.CreateConfigurationInput{
+		KafkaVersions:    expandStringSet(d.Get("kafka_versions").(*schema.Set)),
+		Name:             aws.String(d.Get("name").(string)),
+		ServerProperties: []byte(d.Get("server_properties").(string)),
+	}
+
+	if v, ok := d.GetOk("description"); ok {
+		input.Description = aws.String(v.(string))
+	}
+
+	output, err := conn.CreateConfiguration(input)
+
+	if err != nil {
+		return fmt.Errorf("error creating MSK Configuration: %s", err)
+	}
+
+	d.SetId(aws.StringValue(output.Arn))
+
+	return resourceAwsMskConfigurationRead(d, meta)
+}
+
+func resourceAwsMskConfigurationRead(d *schema.ResourceData, meta interface{}) error {
+	conn := meta.(*AWSClient).kafkaconn
+
+	configurationInput := &kafka.DescribeConfigurationInput{
+		Arn: aws.String(d.Id()),
+	}
+
+	configurationOutput, err := conn.DescribeConfiguration(configurationInput)
+
+	if isAWSErr(err, kafka.ErrCodeNotFoundException, "") {
+		log.Printf("[WARN] MSK Configuration (%s) not found, removing from state", d.Id())
+		d.SetId("")
+		return nil
+	}
+
+	if err != nil {
+		return fmt.Errorf("error describing MSK Configuration (%s): %s", d.Id(), err)
+	}
+
+	if configurationOutput == nil {
+		return fmt.Errorf("error describing MSK Configuration (%s): missing result", d.Id())
+	}
+
+	if configurationOutput.LatestRevision == nil {
+		return fmt.Errorf("error describing MSK Configuration (%s): missing latest revision", d.Id())
+	}
+
+	revision := configurationOutput.LatestRevision.Revision
+	revisionInput := &kafka.DescribeConfigurationRevisionInput{
+		Arn:      aws.String(d.Id()),
+		Revision: revision,
+	}
+
+	revisionOutput, err := conn.DescribeConfigurationRevision(revisionInput)
+
+	if err != nil {
+		return fmt.Errorf("error describing MSK Configuration (%s) Revision (%d): %s", d.Id(), aws.Int64Value(revision), err)
+	}
+
+	if revisionOutput == nil {
+		return fmt.Errorf("error describing MSK Configuration (%s) Revision (%d): missing result", d.Id(), aws.Int64Value(revision))
+	}
+
+	d.Set("arn", aws.StringValue(configurationOutput.Arn))
+	d.Set("description", aws.StringValue(configurationOutput.Description))
+
+	if err := d.Set("kafka_versions", aws.StringValueSlice(configurationOutput.KafkaVersions)); err != nil {
+		return fmt.Errorf("error setting kafka_versions: %s", err)
+	}
+
+	d.Set("latest_revision", aws.Int64Value(revision))
+	d.Set("name", aws.StringValue(configurationOutput.Name))
+	d.Set("server_properties", string(revisionOutput.ServerProperties))
+
+	return nil
+}

aws/resource_aws_msk_configuration_test.go

@@ -0,0 +1,213 @@
+package aws
+
+import (
+	"fmt"
+	"regexp"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/kafka"
+	"github.com/hashicorp/terraform/helper/acctest"
+	"github.com/hashicorp/terraform/helper/resource"
+	"github.com/hashicorp/terraform/terraform"
+)
+
+func TestAccAWSMskConfiguration_basic(t *testing.T) {
+	var configuration1 kafka.DescribeConfigurationOutput
+	rName := acctest.RandomWithPrefix("tf-acc-test")
+	resourceName := "aws_msk_configuration.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckMskConfigurationDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccMskConfigurationConfig(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckMskConfigurationExists(resourceName, &configuration1),
+					testAccMatchResourceAttrRegionalARN(resourceName, "arn", "kafka", regexp.MustCompile(`configuration/.+`)),
+					resource.TestCheckResourceAttr(resourceName, "description", ""),
+					resource.TestCheckResourceAttr(resourceName, "kafka_versions.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "latest_revision", "1"),
+					resource.TestCheckResourceAttr(resourceName, "name", rName),
+					resource.TestMatchResourceAttr(resourceName, "server_properties", regexp.MustCompile(`auto.create.topics.enable = true`)),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
+func TestAccAWSMskConfiguration_Description(t *testing.T) {
+	var configuration1 kafka.DescribeConfigurationOutput
+	rName := acctest.RandomWithPrefix("tf-acc-test")
+	resourceName := "aws_msk_configuration.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckMskConfigurationDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccMskConfigurationConfigDescription(rName, "description1"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckMskConfigurationExists(resourceName, &configuration1),
+					resource.TestCheckResourceAttr(resourceName, "description", "description1"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
+func TestAccAWSMskConfiguration_KafkaVersions(t *testing.T) {
+	var configuration1 kafka.DescribeConfigurationOutput
+	rName := acctest.RandomWithPrefix("tf-acc-test")
+	resourceName := "aws_msk_configuration.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckMskConfigurationDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccMskConfigurationConfigKafkaVersions(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckMskConfigurationExists(resourceName, &configuration1),
+					resource.TestCheckResourceAttr(resourceName, "kafka_versions.#", "2"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
+func TestAccAWSMskConfiguration_ServerProperties(t *testing.T) {
+	var configuration1 kafka.DescribeConfigurationOutput
+	rName := acctest.RandomWithPrefix("tf-acc-test")
+	resourceName := "aws_msk_configuration.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckMskConfigurationDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccMskConfigurationConfigServerProperties(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckMskConfigurationExists(resourceName, &configuration1),
+					resource.TestMatchResourceAttr(resourceName, "server_properties", regexp.MustCompile(`auto.create.topics.enable = false`)),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
+func testAccCheckMskConfigurationDestroy(s *terraform.State) error {
+	for _, rs := range s.RootModule().Resources {
+		if rs.Type != "aws_msk_configuration" {
+			continue
+		}
+
+		// The API does not support deletions at this time
+	}
+
+	return nil
+}
+
+func testAccCheckMskConfigurationExists(resourceName string, configuration *kafka.DescribeConfigurationOutput) resource.TestCheckFunc {
+	return func(s *terraform.State) error {
+		rs, ok := s.RootModule().Resources[resourceName]
+		if !ok {
+			return fmt.Errorf("Not found: %s", resourceName)
+		}
+
+		if rs.Primary.ID == "" {
+			return fmt.Errorf("Resource ID not set: %s", resourceName)
+		}
+
+		conn := testAccProvider.Meta().(*AWSClient).kafkaconn
+
+		input := &kafka.DescribeConfigurationInput{
+			Arn: aws.String(rs.Primary.ID),
+		}
+
+		output, err := conn.DescribeConfiguration(input)
+
+		if err != nil {
+			return fmt.Errorf("error describing MSK Cluster (%s): %s", rs.Primary.ID, err)
+		}
+
+		*configuration = *output
+
+		return nil
+	}
+}
+
+func testAccMskConfigurationConfig(rName string) string {
+	return fmt.Sprintf(`
+resource "aws_msk_configuration" "test" {
+  kafka_versions    = ["2.1.0"]
+  name              = %[1]q
+  server_properties = <<PROPERTIES
+auto.create.topics.enable = true
+delete.topic.enable = true
+PROPERTIES
+}
+`, rName)
+}
+
+func testAccMskConfigurationConfigDescription(rName, description string) string {
+	return fmt.Sprintf(`
+resource "aws_msk_configuration" "test" {
+  description       = %[2]q
+  kafka_versions    = ["2.1.0"]
+  name              = %[1]q
+  server_properties = <<PROPERTIES
+auto.create.topics.enable = true
+PROPERTIES
+}
+`, rName, description)
+}
+
+func testAccMskConfigurationConfigKafkaVersions(rName string) string {
+	return fmt.Sprintf(`
+resource "aws_msk_configuration" "test" {
+  kafka_versions    = ["1.1.1", "2.1.0"]
+  name              = %[1]q
+  server_properties = <<PROPERTIES
+auto.create.topics.enable = true
+PROPERTIES
+}
+`, rName)
+}
+
+func testAccMskConfigurationConfigServerProperties(rName string) string {
+	return fmt.Sprintf(`
+resource "aws_msk_configuration" "test" {
+  kafka_versions    = ["2.1.0"]
+  name              = %[1]q
+  server_properties = <<PROPERTIES
+auto.create.topics.enable = false
+PROPERTIES
+}
+`, rName)
+}

website/aws.erb

@@ -1896,6 +1896,10 @@
                         <li>
                             <a href="/docs/providers/aws/r/msk_cluster.html">aws_msk_cluster</a>
                         </li>
+
+                        <li>
+                            <a href="/docs/providers/aws/r/msk_configuration.html">aws_msk_configuration</a>
+                        </li>
                     </ul>
                 </li>