diff --git a/internal/services/servicebus/servicebus_namespace_disaster_recovery_config_data_source.go b/internal/services/servicebus/servicebus_namespace_disaster_recovery_config_data_source.go
index 5313321e3263..27bee139b02f 100644
--- a/internal/services/servicebus/servicebus_namespace_disaster_recovery_config_data_source.go
+++ b/internal/services/servicebus/servicebus_namespace_disaster_recovery_config_data_source.go
@@ -12,6 +12,7 @@ import (
 	"github.com/hashicorp/terraform-provider-azurerm/internal/clients"
 	"github.com/hashicorp/terraform-provider-azurerm/internal/services/servicebus/validate"
 	"github.com/hashicorp/terraform-provider-azurerm/internal/tf/pluginsdk"
+	"github.com/hashicorp/terraform-provider-azurerm/internal/tf/validation"
 	"github.com/hashicorp/terraform-provider-azurerm/internal/timeouts"
 )
 
@@ -50,6 +51,12 @@ func dataSourceServiceBusNamespaceDisasterRecoveryConfig() *pluginsdk.Resource {
 				AtLeastOneOf: []string{"namespace_id", "resource_group_name", "namespace_name"},
 			},
 
+			"alias_authorization_rule_id": {
+				Type:         pluginsdk.TypeString,
+				Optional:     true,
+				ValidateFunc: validation.StringIsNotEmpty,
+			},
+
 			"partner_namespace_id": {
 				Type:     pluginsdk.TypeString,
 				Computed: true,
@@ -124,7 +131,16 @@ func dataSourceServiceBusNamespaceDisasterRecoveryConfigRead(d *pluginsdk.Resour
 
 	d.SetId(id.ID())
 
-	authRuleId := disasterrecoveryconfigs.NewAuthorizationRuleID(id.SubscriptionId, id.ResourceGroupName, id.NamespaceName, d.Get("name").(string))
+	// the auth rule cannot be retrieved by dr config name, the shared access policy should either be specified by user or using the default one which is `RootManageSharedAccessKey`
+	authRuleId := disasterrecoveryconfigs.NewAuthorizationRuleID(id.SubscriptionId, id.ResourceGroupName, id.NamespaceName, serviceBusNamespaceDefaultAuthorizationRule)
+	if input := d.Get("alias_authorization_rule_id").(string); input != "" {
+		ruleId, err := disasterrecoveryconfigs.ParseAuthorizationRuleID(input)
+		if err != nil {
+			return fmt.Errorf("parsing primary namespace auth rule id error: %+v", err)
+		}
+		authRuleId = *ruleId
+	}
+
 	keys, err := client.ListKeys(ctx, authRuleId)
 	if err != nil {
 		log.Printf("[WARN] listing default keys for %s: %+v", id, err)
diff --git a/internal/services/servicebus/servicebus_namespace_disaster_recovery_config_resource.go b/internal/services/servicebus/servicebus_namespace_disaster_recovery_config_resource.go
index 2593ce4314f3..55e4c8c5d654 100644
--- a/internal/services/servicebus/servicebus_namespace_disaster_recovery_config_resource.go
+++ b/internal/services/servicebus/servicebus_namespace_disaster_recovery_config_resource.go
@@ -15,6 +15,7 @@ import (
 	"github.com/hashicorp/terraform-provider-azurerm/internal/clients"
 	"github.com/hashicorp/terraform-provider-azurerm/internal/locks"
 	"github.com/hashicorp/terraform-provider-azurerm/internal/tf/pluginsdk"
+	"github.com/hashicorp/terraform-provider-azurerm/internal/tf/validation"
 	"github.com/hashicorp/terraform-provider-azurerm/internal/timeouts"
 	"github.com/hashicorp/terraform-provider-azurerm/utils"
 )
@@ -59,6 +60,12 @@ func resourceServiceBusNamespaceDisasterRecoveryConfig() *pluginsdk.Resource {
 				ValidateFunc: azure.ValidateResourceIDOrEmpty,
 			},
 
+			"alias_authorization_rule_id": {
+				Type:         pluginsdk.TypeString,
+				Optional:     true,
+				ValidateFunc: validation.StringIsNotEmpty,
+			},
+
 			"primary_connection_string_alias": {
 				Type:      pluginsdk.TypeString,
 				Computed:  true,
@@ -198,7 +205,15 @@ func resourceServiceBusNamespaceDisasterRecoveryConfigRead(d *pluginsdk.Resource
 		}
 	}
 
-	authRuleId := disasterrecoveryconfigs.NewAuthorizationRuleID(id.SubscriptionId, id.ResourceGroupName, id.NamespaceName, id.Alias)
+	// the auth rule cannot be retrieved by dr config name, the shared access policy should either be specified by user or using the default one which is `RootManageSharedAccessKey`
+	authRuleId := disasterrecoveryconfigs.NewAuthorizationRuleID(id.SubscriptionId, id.ResourceGroupName, id.NamespaceName, serviceBusNamespaceDefaultAuthorizationRule)
+	if input := d.Get("alias_authorization_rule_id").(string); input != "" {
+		ruleId, err := disasterrecoveryconfigs.ParseAuthorizationRuleID(input)
+		if err != nil {
+			return fmt.Errorf("parsing primary namespace auth rule id error: %+v", err)
+		}
+		authRuleId = *ruleId
+	}
 
 	keys, err := client.ListKeys(ctx, authRuleId)
 
diff --git a/internal/services/servicebus/servicebus_topic_resource_test.go b/internal/services/servicebus/servicebus_topic_resource_test.go
index 5d4b548dceff..b55f1fedb137 100644
--- a/internal/services/servicebus/servicebus_topic_resource_test.go
+++ b/internal/services/servicebus/servicebus_topic_resource_test.go
@@ -89,18 +89,21 @@ func TestAccServiceBusTopic_basicDisableEnable(t *testing.T) {
 				check.That(data.ResourceName).ExistsInAzure(r),
 			),
 		},
+		data.ImportStep(),
 		{
 			Config: r.basicDisabled(data),
 			Check: acceptance.ComposeTestCheckFunc(
 				check.That(data.ResourceName).ExistsInAzure(r),
 			),
 		},
+		data.ImportStep(),
 		{
 			Config: r.basic(data),
 			Check: acceptance.ComposeTestCheckFunc(
 				check.That(data.ResourceName).ExistsInAzure(r),
 			),
 		},
+		data.ImportStep(),
 	})
 }
 
@@ -115,6 +118,7 @@ func TestAccServiceBusTopic_update(t *testing.T) {
 				check.That(data.ResourceName).ExistsInAzure(r),
 			),
 		},
+		data.ImportStep(),
 		{
 			Config: r.update(data),
 			Check: acceptance.ComposeTestCheckFunc(
@@ -122,6 +126,7 @@ func TestAccServiceBusTopic_update(t *testing.T) {
 				check.That(data.ResourceName).Key("enable_express").HasValue("true"),
 			),
 		},
+		data.ImportStep(),
 	})
 }
 
@@ -136,6 +141,7 @@ func TestAccServiceBusTopic_enablePartitioningStandard(t *testing.T) {
 				check.That(data.ResourceName).ExistsInAzure(r),
 			),
 		},
+		data.ImportStep(),
 		{
 			Config: r.enablePartitioningStandard(data),
 			Check: acceptance.ComposeTestCheckFunc(
@@ -174,6 +180,7 @@ func TestAccServiceBusTopic_enablePartitioningPremium(t *testing.T) {
 				check.That(data.ResourceName).ExistsInAzure(r),
 			),
 		},
+		data.ImportStep(),
 		{
 			Config: r.enablePartitioningPremium(data),
 			Check: acceptance.ComposeTestCheckFunc(
@@ -196,6 +203,7 @@ func TestAccServiceBusTopic_enableDuplicateDetection(t *testing.T) {
 				check.That(data.ResourceName).ExistsInAzure(r),
 			),
 		},
+		data.ImportStep(),
 		{
 			Config: r.enableDuplicateDetection(data),
 			Check: acceptance.ComposeTestCheckFunc(
diff --git a/website/docs/r/servicebus_namespace_disaster_recovery_config.html.markdown b/website/docs/r/servicebus_namespace_disaster_recovery_config.html.markdown
index 30f561825eda..d18b57ac17e0 100644
--- a/website/docs/r/servicebus_namespace_disaster_recovery_config.html.markdown
+++ b/website/docs/r/servicebus_namespace_disaster_recovery_config.html.markdown
@@ -10,7 +10,7 @@ description: |-
 
 Manages a Disaster Recovery Config for a Service Bus Namespace.
 
-~> **NOTE:** Disaster Recovery Config is a Premium SKU only capability. 
+~> **NOTE:** Disaster Recovery Config is a Premium SKU only capability.
 
 ## Example Usage
 
@@ -36,10 +36,20 @@ resource "azurerm_servicebus_namespace" "secondary" {
   capacity            = "1"
 }
 
+resource "azurerm_servicebus_namespace_authorization_rule" "example" {
+  name         = "examplerule"
+  namespace_id = azurerm_servicebus_namespace.example.id
+
+  listen = true
+  send   = true
+  manage = false
+}
+
 resource "azurerm_servicebus_namespace_disaster_recovery_config" "example" {
-  name                 = "servicebus-alias-name"
-  primary_namespace_id = azurerm_servicebus_namespace.primary.id
-  partner_namespace_id = azurerm_servicebus_namespace.secondary.id
+  name                        = "servicebus-alias-name"
+  primary_namespace_id        = azurerm_servicebus_namespace.primary.id
+  partner_namespace_id        = azurerm_servicebus_namespace.secondary.id
+  alias_authorization_rule_id = azurerm_servicebus_namespace_authorization_rule.example.id
 }
 
 ```
@@ -54,6 +64,8 @@ The following arguments are supported:
 
 * `partner_namespace_id` - (Required) The ID of the Service Bus Namespace to replicate to.
 
+* `alias_authorization_rule_id` - (Optional) The Shared access policies used to access the connection string for the alias. Defaults to `RootManageSharedAccessKey`.
+
 ## Attributes Reference
 
 The following attributes are exported:
@@ -62,7 +74,7 @@ The following attributes are exported:
 
 * `primary_connection_string_alias` - The alias Primary Connection String for the ServiceBus Namespace.
 
-* `secondary_connection_string_alias` - The alias Secondary Connection String for the ServiceBus Namespace 
+* `secondary_connection_string_alias` - The alias Secondary Connection String for the ServiceBus Namespace
 
 * `default_primary_key` - The primary access key for the authorization rule `RootManageSharedAccessKey`.