resource/aws_mq_broker: Add RabbitMQ engine type

.changelog/16108.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+resource/aws_mq_broker: Add RabbitMQ as option for `engine_type`, and new arguments `authentication_strategy`, `ldap_server_metadata`, and `storage_type`. Improve handling of eventual consistency.
+```

aws/data_source_aws_mq_broker.go

@@ -13,6 +13,18 @@ func dataSourceAwsMqBroker() *schema.Resource {
 		Read: dataSourceAwsmQBrokerRead,
 
 		Schema: map[string]*schema.Schema{
+			"arn": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+			"authentication_strategy": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+			"auto_minor_version_upgrade": {
+				Type:     schema.TypeBool,
+				Computed: true,
+			},
 			"broker_id": {
 				Type:          schema.TypeString,
 				Optional:      true,
@@ -25,14 +37,6 @@ func dataSourceAwsMqBroker() *schema.Resource {
 				Computed:      true,
 				ConflictsWith: []string{"broker_id"},
 			},
-			"auto_minor_version_upgrade": {
-				Type:     schema.TypeBool,
-				Computed: true,
-			},
-			"arn": {
-				Type:     schema.TypeString,
-				Computed: true,
-			},
 			"configuration": {
 				Type:     schema.TypeList,
 				Computed: true,
@@ -90,15 +94,68 @@ func dataSourceAwsMqBroker() *schema.Resource {
 							Type:     schema.TypeString,
 							Computed: true,
 						},
+						"endpoints": {
+							Type:     schema.TypeList,
+							Computed: true,
+							Elem:     &schema.Schema{Type: schema.TypeString},
+						},
 						"ip_address": {
 							Type:     schema.TypeString,
 							Computed: true,
 						},
-						"endpoints": {
+					},
+				},
+			},
+			"ldap_server_metadata": {
+				Type:     schema.TypeList,
+				Computed: true,
+				Elem: &schema.Resource{
+					Schema: map[string]*schema.Schema{
+						"hosts": {
 							Type:     schema.TypeList,
 							Computed: true,
 							Elem:     &schema.Schema{Type: schema.TypeString},
 						},
+						"role_base": {
+							Type:     schema.TypeString,
+							Computed: true,
+						},
+						"role_name": {
+							Type:     schema.TypeString,
+							Computed: true,
+						},
+						"role_search_matching": {
+							Type:     schema.TypeString,
+							Computed: true,
+						},
+						"role_search_subtree": {
+							Type:     schema.TypeBool,
+							Computed: true,
+						},
+						"service_account_password": {
+							Type:     schema.TypeString,
+							Computed: true,
+						},
+						"service_account_username": {
+							Type:     schema.TypeString,
+							Computed: true,
+						},
+						"user_base": {
+							Type:     schema.TypeString,
+							Computed: true,
+						},
+						"user_role_name": {
+							Type:     schema.TypeString,
+							Computed: true,
+						},
+						"user_search_matching": {
+							Type:     schema.TypeString,
+							Computed: true,
+						},
+						"user_search_subtree": {
+							Type:     schema.TypeBool,
+							Computed: true,
+						},
 					},
 				},
 			},
@@ -155,6 +212,10 @@ func dataSourceAwsMqBroker() *schema.Resource {
 				Elem:     &schema.Schema{Type: schema.TypeString},
 				Computed: true,
 			},
+			"storage_type": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
 			"subnet_ids": {
 				Type:     schema.TypeSet,
 				Elem:     &schema.Schema{Type: schema.TypeString},

aws/data_source_aws_mq_broker_test.go

@@ -27,6 +27,7 @@ func TestAccDataSourceAWSMqBroker_basic(t *testing.T) {
 				Check: resource.ComposeAggregateTestCheckFunc(
 					resource.TestCheckResourceAttrPair(dataSourceByIdName, "arn", resourceName, "arn"),
 					resource.TestCheckResourceAttrPair(dataSourceByIdName, "broker_name", resourceName, "broker_name"),
+					resource.TestCheckResourceAttrPair(dataSourceByIdName, "authentication_strategy", resourceName, "authentication_strategy"),
 					resource.TestCheckResourceAttrPair(dataSourceByIdName, "auto_minor_version_upgrade", resourceName, "auto_minor_version_upgrade"),
 					resource.TestCheckResourceAttrPair(dataSourceByIdName, "deployment_mode", resourceName, "deployment_mode"),
 					resource.TestCheckResourceAttrPair(dataSourceByIdName, "configuration.#", resourceName, "configuration.#"),
@@ -40,6 +41,7 @@ func TestAccDataSourceAWSMqBroker_basic(t *testing.T) {
 					resource.TestCheckResourceAttrPair(dataSourceByIdName, "maintenance_window_start_time.#", resourceName, "maintenance_window_start_time.#"),
 					resource.TestCheckResourceAttrPair(dataSourceByIdName, "publicly_accessible", resourceName, "publicly_accessible"),
 					resource.TestCheckResourceAttrPair(dataSourceByIdName, "security_groups.#", resourceName, "security_groups.#"),
+					resource.TestCheckResourceAttrPair(dataSourceByIdName, "storage_type", resourceName, "storage_type"),
 					resource.TestCheckResourceAttrPair(dataSourceByIdName, "subnet_ids.#", resourceName, "subnet_ids.#"),
 					resource.TestCheckResourceAttrPair(dataSourceByIdName, "tags.%", resourceName, "tags.%"),
 					resource.TestCheckResourceAttrPair(dataSourceByIdName, "user.#", resourceName, "user.#"),

aws/internal/service/mq/waiter/status.go

@@ -0,0 +1,30 @@
+package waiter
+
+import (
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/mq"
+	"github.com/hashicorp/aws-sdk-go-base/tfawserr"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+)
+
+func BrokerStatus(conn *mq.MQ, id string) resource.StateRefreshFunc {
+	return func() (interface{}, string, error) {
+		output, err := conn.DescribeBroker(&mq.DescribeBrokerInput{
+			BrokerId: aws.String(id),
+		})
+
+		if tfawserr.ErrCodeEquals(err, mq.ErrCodeNotFoundException) {
+			return nil, "", nil
+		}
+
+		if err != nil {
+			return nil, "", err
+		}
+
+		if output == nil {
+			return nil, "", nil
+		}
+
+		return output, aws.StringValue(output.BrokerState), nil
+	}
+}

aws/internal/service/mq/waiter/waiter.go

@@ -0,0 +1,71 @@
+package waiter
+
+import (
+	"time"
+
+	"github.com/aws/aws-sdk-go/service/mq"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+)
+
+const (
+	BrokerCreateTimeout = 30 * time.Minute
+	BrokerDeleteTimeout = 30 * time.Minute
+	BrokerRebootTimeout = 30 * time.Minute
+)
+
+func BrokerCreated(conn *mq.MQ, id string) (*mq.DescribeBrokerResponse, error) {
+	stateConf := resource.StateChangeConf{
+		Pending: []string{
+			mq.BrokerStateCreationInProgress,
+			mq.BrokerStateRebootInProgress,
+		},
+		Target:  []string{mq.BrokerStateRunning},
+		Timeout: BrokerCreateTimeout,
+		Refresh: BrokerStatus(conn, id),
+	}
+	outputRaw, err := stateConf.WaitForState()
+
+	if output, ok := outputRaw.(*mq.DescribeBrokerResponse); ok {
+		return output, err
+	}
+
+	return nil, err
+}
+
+func BrokerDeleted(conn *mq.MQ, id string) (*mq.DescribeBrokerResponse, error) {
+	stateConf := resource.StateChangeConf{
+		Pending: []string{
+			mq.BrokerStateRunning,
+			mq.BrokerStateRebootInProgress,
+			mq.BrokerStateDeletionInProgress,
+		},
+		Target:  []string{},
+		Timeout: BrokerDeleteTimeout,
+		Refresh: BrokerStatus(conn, id),
+	}
+	outputRaw, err := stateConf.WaitForState()
+
+	if output, ok := outputRaw.(*mq.DescribeBrokerResponse); ok {
+		return output, err
+	}
+
+	return nil, err
+}
+
+func BrokerRebooted(conn *mq.MQ, id string) (*mq.DescribeBrokerResponse, error) {
+	stateConf := resource.StateChangeConf{
+		Pending: []string{
+			mq.BrokerStateRebootInProgress,
+		},
+		Target:  []string{mq.BrokerStateRunning},
+		Timeout: BrokerRebootTimeout,
+		Refresh: BrokerStatus(conn, id),
+	}
+	outputRaw, err := stateConf.WaitForState()
+
+	if output, ok := outputRaw.(*mq.DescribeBrokerResponse); ok {
+		return output, err
+	}
+
+	return nil, err
+}