diff --git a/aws/internal/service/mq/waiter/waiter.go b/aws/internal/service/mq/waiter/waiter.go
index 3bc226dc58df..d55ab76c36b5 100644
--- a/aws/internal/service/mq/waiter/waiter.go
+++ b/aws/internal/service/mq/waiter/waiter.go
@@ -35,9 +35,10 @@ func BrokerCreated(conn *mq.MQ, id string) (*mq.DescribeBrokerResponse, error) {
 func BrokerDeleted(conn *mq.MQ, id string) (*mq.DescribeBrokerResponse, error) {
 	stateConf := resource.StateChangeConf{
 		Pending: []string{
-			mq.BrokerStateRunning,
-			mq.BrokerStateRebootInProgress,
+			mq.BrokerStateCreationFailed,
 			mq.BrokerStateDeletionInProgress,
+			mq.BrokerStateRebootInProgress,
+			mq.BrokerStateRunning,
 		},
 		Target:  []string{},
 		Timeout: BrokerDeleteTimeout,
diff --git a/aws/resource_aws_mq_broker.go b/aws/resource_aws_mq_broker.go
index 9ba3f17dc688..b43e1f9128b1 100644
--- a/aws/resource_aws_mq_broker.go
+++ b/aws/resource_aws_mq_broker.go
@@ -146,6 +146,7 @@ func resourceAwsMqBroker() *schema.Resource {
 			"ldap_server_metadata": {
 				Type:     schema.TypeList,
 				Optional: true,
+				ForceNew: true,
 				MaxItems: 1,
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
@@ -171,8 +172,9 @@ func resourceAwsMqBroker() *schema.Resource {
 							Optional: true,
 						},
 						"service_account_password": {
-							Type:     schema.TypeString,
-							Optional: true,
+							Type:      schema.TypeString,
+							Optional:  true,
+							Sensitive: true,
 						},
 						"service_account_username": {
 							Type:     schema.TypeString,
@@ -327,7 +329,6 @@ func resourceAwsMqBrokerCreate(d *schema.ResourceData, meta interface{}) error {
 		AutoMinorVersionUpgrade: aws.Bool(d.Get("auto_minor_version_upgrade").(bool)),
 		BrokerName:              aws.String(name),
 		CreatorRequestId:        aws.String(requestId),
-		EncryptionOptions:       expandMqEncryptionOptions(d.Get("encryption_options").([]interface{})),
 		EngineType:              aws.String(d.Get("engine_type").(string)),
 		EngineVersion:           aws.String(d.Get("engine_version").(string)),
 		HostInstanceType:        aws.String(d.Get("host_instance_type").(string)),
@@ -338,19 +339,22 @@ func resourceAwsMqBrokerCreate(d *schema.ResourceData, meta interface{}) error {
 	if v, ok := d.GetOk("authentication_strategy"); ok {
 		input.AuthenticationStrategy = aws.String(v.(string))
 	}
-	if v, ok := d.GetOk("configuration"); ok {
+	if v, ok := d.GetOk("configuration"); ok && len(v.([]interface{})) > 0 && v.([]interface{})[0] != nil {
 		input.Configuration = expandMqConfigurationId(v.([]interface{}))
 	}
 	if v, ok := d.GetOk("deployment_mode"); ok {
 		input.DeploymentMode = aws.String(v.(string))
 	}
-	if v, ok := d.GetOk("logs"); ok && len(v.([]interface{})) > 0 {
+	if v, ok := d.GetOk("encryption_options"); ok && len(v.([]interface{})) > 0 && v.([]interface{})[0] != nil {
+		input.EncryptionOptions = expandMqEncryptionOptions(d.Get("encryption_options").([]interface{}))
+	}
+	if v, ok := d.GetOk("logs"); ok && len(v.([]interface{})) > 0 && v.([]interface{})[0] != nil {
 		input.Logs = expandMqLogs(v.([]interface{}))
 	}
 	if v, ok := d.GetOk("ldap_server_metadata"); ok && len(v.([]interface{})) > 0 && v.([]interface{})[0] != nil {
-		input.LdapServerMetadata = expandMQLDAPServerMetadata(v.([]interface{})[0].(map[string]interface{}))
+		input.LdapServerMetadata = expandMQLDAPServerMetadata(v.([]interface{}))
 	}
-	if v, ok := d.GetOk("maintenance_window_start_time"); ok {
+	if v, ok := d.GetOk("maintenance_window_start_time"); ok && len(v.([]interface{})) > 0 && v.([]interface{})[0] != nil {
 		input.MaintenanceWindowStartTime = expandMqWeeklyStartTime(v.([]interface{}))
 	}
 	if v, ok := d.GetOk("security_groups"); ok && v.(*schema.Set).Len() > 0 {
@@ -436,8 +440,12 @@ func resourceAwsMqBrokerRead(d *schema.ResourceData, meta interface{}) error {
 	}
 
 	if output.LdapServerMetadata != nil {
-		if err := d.Set("ldap_server_metadata", flattenMQLDAPServerMetadata(output.LdapServerMetadata)); err != nil {
-			return fmt.Errorf("error setting lda_server_metadata: %w", err)
+		password := ""
+		if v, ok := d.GetOk("ldap_server_metadata.0.service_account_password"); ok {
+			password = v.(string)
+		}
+		if err := d.Set("ldap_server_metadata", flattenMQLDAPServerMetadata(output.LdapServerMetadata, password)); err != nil {
+			return fmt.Errorf("error setting ldap_server_metadata: %w", err)
 		}
 	} else {
 		d.Set("ldap_server_metadata", nil)
@@ -491,21 +499,6 @@ func resourceAwsMqBrokerUpdate(d *schema.ResourceData, meta interface{}) error {
 
 	requiresReboot := false
 
-	if d.HasChange("ldap_server_metadata") {
-		input := &mq.UpdateBrokerRequest{
-			BrokerId:           aws.String(d.Id()),
-			LdapServerMetadata: nil,
-		}
-
-		if v, ok := d.GetOk("ldap_server_metadata"); ok && len(v.([]interface{})) > 0 && v.([]interface{})[0] != nil {
-			input.LdapServerMetadata = expandMQLDAPServerMetadata(v.([]interface{})[0].(map[string]interface{}))
-		}
-
-		if _, err := conn.UpdateBroker(input); err != nil {
-			return fmt.Errorf("error updating MQ Broker (%s) LDAP server metadata: %w", d.Id(), err)
-		}
-	}
-
 	if d.HasChange("security_groups") {
 		_, err := conn.UpdateBroker(&mq.UpdateBrokerRequest{
 			BrokerId:       aws.String(d.Id()),
@@ -947,7 +940,7 @@ func expandMqLogs(l []interface{}) *mq.Logs {
 	return logs
 }
 
-func flattenMQLDAPServerMetadata(apiObject *mq.LdapServerMetadataOutput) map[string]interface{} {
+func flattenMQLDAPServerMetadata(apiObject *mq.LdapServerMetadataOutput, password string) []interface{} {
 	if apiObject == nil {
 		return nil
 	}
@@ -969,6 +962,9 @@ func flattenMQLDAPServerMetadata(apiObject *mq.LdapServerMetadataOutput) map[str
 	if v := apiObject.RoleSearchSubtree; v != nil {
 		tfMap["role_search_subtree"] = aws.BoolValue(v)
 	}
+	if password != "" {
+		tfMap["service_account_password"] = password
+	}
 	if v := apiObject.ServiceAccountUsername; v != nil {
 		tfMap["service_account_username"] = aws.StringValue(v)
 	}
@@ -985,16 +981,18 @@ func flattenMQLDAPServerMetadata(apiObject *mq.LdapServerMetadataOutput) map[str
 		tfMap["user_search_subtree"] = aws.BoolValue(v)
 	}
 
-	return tfMap
+	return []interface{}{tfMap}
 }
 
-func expandMQLDAPServerMetadata(tfMap map[string]interface{}) *mq.LdapServerMetadataInput {
-	if tfMap == nil {
+func expandMQLDAPServerMetadata(tfList []interface{}) *mq.LdapServerMetadataInput {
+	if len(tfList) == 0 || tfList[0] == nil {
 		return nil
 	}
 
 	apiObject := &mq.LdapServerMetadataInput{}
 
+	tfMap := tfList[0].(map[string]interface{})
+
 	if v, ok := tfMap["hosts"]; ok && len(v.([]interface{})) > 0 {
 		apiObject.Hosts = expandStringList(v.([]interface{}))
 	}
diff --git a/aws/resource_aws_mq_broker_test.go b/aws/resource_aws_mq_broker_test.go
index 1a605c5c87fc..d0fa5141d236 100644
--- a/aws/resource_aws_mq_broker_test.go
+++ b/aws/resource_aws_mq_broker_test.go
@@ -1007,6 +1007,105 @@ func TestAccAWSMqBroker_rabbitmq(t *testing.T) {
 	})
 }
 
+func TestAccAWSMqBroker_clusterRabbitMQ(t *testing.T) {
+	var broker mq.DescribeBrokerResponse
+	rName := acctest.RandomWithPrefix("tf-acc-test")
+	resourceName := "aws_mq_broker.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck: func() {
+			testAccPreCheck(t)
+			testAccPartitionHasServicePreCheck(mq.EndpointsID, t)
+			testAccPreCheckAWSMq(t)
+		},
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckAwsMqBrokerDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccRabbitMqClusterBrokerConfig(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAwsMqBrokerExists(resourceName, &broker),
+					resource.TestCheckResourceAttr(resourceName, "auto_minor_version_upgrade", "false"),
+					resource.TestCheckResourceAttr(resourceName, "broker_name", rName),
+					resource.TestCheckResourceAttr(resourceName, "deployment_mode", "CLUSTER_MULTI_AZ"),
+					resource.TestCheckResourceAttr(resourceName, "encryption_options.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "encryption_options.0.use_aws_owned_key", "true"),
+					resource.TestCheckResourceAttr(resourceName, "engine_type", "RabbitMQ"),
+					resource.TestCheckResourceAttr(resourceName, "engine_version", "3.8.6"),
+					resource.TestCheckResourceAttr(resourceName, "host_instance_type", "mq.m5.large"),
+					resource.TestCheckResourceAttr(resourceName, "maintenance_window_start_time.#", "1"),
+					resource.TestCheckResourceAttrSet(resourceName, "maintenance_window_start_time.0.day_of_week"),
+					resource.TestCheckResourceAttrSet(resourceName, "maintenance_window_start_time.0.time_of_day"),
+					resource.TestCheckResourceAttr(resourceName, "logs.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "logs.0.general", "false"),
+					resource.TestCheckResourceAttr(resourceName, "maintenance_window_start_time.0.time_zone", "UTC"),
+					resource.TestCheckResourceAttr(resourceName, "publicly_accessible", "false"),
+					resource.TestCheckResourceAttr(resourceName, "security_groups.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "subnet_ids.#", "4"),
+					resource.TestCheckResourceAttr(resourceName, "user.#", "1"),
+					resource.TestCheckTypeSetElemNestedAttrs(resourceName, "user.*", map[string]string{
+						"console_access": "false",
+						"groups.#":       "0",
+						"username":       "Test",
+						"password":       "TestTest1234",
+					}),
+					testAccMatchResourceAttrRegionalARN(resourceName, "arn", "mq", regexp.MustCompile(`broker:+.`)),
+					resource.TestCheckResourceAttr(resourceName, "instances.#", "1"),
+					resource.TestMatchResourceAttr(resourceName, "instances.0.console_url",
+						regexp.MustCompile(`^https://[a-f0-9-]+\.mq.[a-z0-9-]+.amazonaws.com$`)),
+					resource.TestCheckResourceAttr(resourceName, "instances.0.endpoints.#", "1"),
+					resource.TestMatchResourceAttr(resourceName, "instances.0.endpoints.0", regexp.MustCompile(`^amqps://[a-z0-9-\.]+:5671$`)),
+				),
+			},
+			{
+				ResourceName:            resourceName,
+				ImportState:             true,
+				ImportStateVerify:       true,
+				ImportStateVerifyIgnore: []string{"apply_immediately", "user"},
+			},
+		},
+	})
+}
+
+func TestAccAWSMqBroker_ldap(t *testing.T) {
+	var broker mq.DescribeBrokerResponse
+	rName := acctest.RandomWithPrefix("tf-acc-test")
+	resourceName := "aws_mq_broker.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck: func() {
+			testAccPreCheck(t)
+			testAccPartitionHasServicePreCheck(mq.EndpointsID, t)
+			testAccPreCheckAWSMq(t)
+		},
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckAwsMqBrokerDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccMqBrokerConfig_ldap(rName, "anyusername"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAwsMqBrokerExists(resourceName, &broker),
+					resource.TestCheckResourceAttr(resourceName, "auto_minor_version_upgrade", "false"),
+					resource.TestCheckResourceAttr(resourceName, "broker_name", rName),
+					resource.TestCheckResourceAttr(resourceName, "authentication_strategy", "ldap"),
+					resource.TestCheckResourceAttr(resourceName, "ldap_server_metadata.0.hosts.#", "2"),
+					resource.TestCheckResourceAttr(resourceName, "ldap_server_metadata.0.hosts.0", "my.ldap.server-1.com"),
+					resource.TestCheckResourceAttr(resourceName, "ldap_server_metadata.0.hosts.1", "my.ldap.server-2.com"),
+					resource.TestCheckResourceAttr(resourceName, "ldap_server_metadata.0.role_base", "role.base"),
+					resource.TestCheckResourceAttr(resourceName, "ldap_server_metadata.0.role_name", "role.name"),
+					resource.TestCheckResourceAttr(resourceName, "ldap_server_metadata.0.role_search_matching", "role.search.matching"),
+					resource.TestCheckResourceAttr(resourceName, "ldap_server_metadata.0.role_search_subtree", "true"),
+					resource.TestCheckResourceAttr(resourceName, "ldap_server_metadata.0.service_account_username", "anyusername"),
+					resource.TestCheckResourceAttr(resourceName, "ldap_server_metadata.0.user_base", "user.base"),
+					resource.TestCheckResourceAttr(resourceName, "ldap_server_metadata.0.user_role_name", "user.role.name"),
+					resource.TestCheckResourceAttr(resourceName, "ldap_server_metadata.0.user_search_matching", "user.search.matching"),
+					resource.TestCheckResourceAttr(resourceName, "ldap_server_metadata.0.user_search_subtree", "true"),
+				),
+			},
+		},
+	})
+}
+
 func testAccCheckAwsMqBrokerDestroy(s *terraform.State) error {
 	conn := testAccProvider.Meta().(*AWSClient).mqconn
 
@@ -1617,3 +1716,67 @@ resource "aws_mq_broker" "test" {
 }
 `, rName)
 }
+
+func testAccRabbitMqClusterBrokerConfig(rName string) string {
+	return fmt.Sprintf(`
+resource "aws_security_group" "test" {
+  name = %[1]q
+}
+
+resource "aws_mq_broker" "test" {
+  broker_name        = %[1]q
+  engine_type        = "RabbitMQ"
+  engine_version     = "3.8.6"
+  host_instance_type = "mq.m5.large"
+  security_groups    = [aws_security_group.test.id]
+  storage_type       = "ebs"
+  deployment_mode    = "CLUSTER_MULTI_AZ"
+
+  user {
+    username = "Test"
+    password = "TestTest1234"
+  }
+}
+`, rName)
+}
+
+func testAccMqBrokerConfig_ldap(rName, ldapUsername string) string {
+	return fmt.Sprintf(`
+resource "aws_security_group" "test" {
+  name = %[1]q
+}
+
+resource "aws_mq_broker" "test" {
+  apply_immediately       = true
+  authentication_strategy = "ldap"
+  broker_name             = %[1]q
+  engine_type             = "ActiveMQ"
+  engine_version          = "5.15.0"
+  host_instance_type      = "mq.t2.micro"
+  security_groups         = [aws_security_group.test.id]
+
+  logs {
+    general = true
+  }
+
+  user {
+    username = "Test"
+    password = "TestTest1234"
+  }
+
+  ldap_server_metadata {
+    hosts                    = ["my.ldap.server-1.com", "my.ldap.server-2.com"]
+    role_base                = "role.base"
+    role_name                = "role.name"
+    role_search_matching     = "role.search.matching"
+    role_search_subtree      = true
+    service_account_password = "supersecret"
+    service_account_username = %[2]q
+    user_base                = "user.base"
+    user_role_name           = "user.role.name"
+    user_search_matching     = "user.search.matching"
+    user_search_subtree      = true
+  }
+}
+`, rName, ldapUsername)
+}
diff --git a/website/docs/r/mq_broker.html.markdown b/website/docs/r/mq_broker.html.markdown
index 40cff4e00ca6..7da2cbb79088 100644
--- a/website/docs/r/mq_broker.html.markdown
+++ b/website/docs/r/mq_broker.html.markdown
@@ -89,7 +89,7 @@ The following arguments are optional:
 * `configuration` - (Optional) Configuration block for broker configuration. Applies to `engine_type` of `ActiveMQ` only. Detailed below.
 * `deployment_mode` - (Optional) Deployment mode of the broker. Valid values are `SINGLE_INSTANCE`, `ACTIVE_STANDBY_MULTI_AZ`, and `CLUSTER_MULTI_AZ`. Default is `SINGLE_INSTANCE`.
 * `encryption_options` - (Optional) Configuration block containing encryption options. Detailed below.
-* `ldap_server_metadata` - (Optional) Configuration block for the LDAP server used to authenticate and authorize connections to the broker. Not supported for `engine_type` `RabbitMQ`. Detailed below.
+* `ldap_server_metadata` - (Optional) Configuration block for the LDAP server used to authenticate and authorize connections to the broker. Not supported for `engine_type` `RabbitMQ`. Detailed below. (Currently, AWS may not process changes to LDAP server metadata.)
 * `logs` - (Optional) Configuration block for the logging configuration of the broker. Detailed below.
 * `maintenance_window_start_time` - (Optional) Configuration block for the maintenance window start time. Detailed below.
 * `publicly_accessible` - (Optional) Whether to enable connections from applications outside of the VPC that hosts the broker's subnets.