Removing tight Resource Manager Provider coupling

sdk/spring/azure-spring-boot-samples/azure-spring-boot-sample-storage-resource/README.adoc

@@ -76,6 +76,8 @@ spring.cloud.azure.resource-group=[resource-group]
 spring.cloud.azure.storage.account=[account-name]
 ....
 
+
+
 == How to run
 
 5. Update link:src/main/resources/application.properties[application.properties]

sdk/spring/azure-spring-boot-samples/azure-spring-boot-sample-storage-resource/src/main/resources/application.properties

@@ -10,13 +10,17 @@
 
 # Storage account name length should be between 3 and 24
 # and use numbers and lower-case letters only.
-spring.cloud.azure.storage.account=accountName
+spring.cloud.azure.storage.account=ybstortst2
+
+spring.cloud.azure.resource-group=test
 
 # Change into your containerName and blobName
-blob=azure-blob://containerName/blobName
+blob=azure-blob://ybstrcntnr/testblob.txt
 
 # Default environment is GLOBAL. Provide your own if in another environment
 # Example environment is China, GLOBAL
 # spring.cloud.azure.environment=[environment]
 # Example region is westUS, northchina
-spring.cloud.azure.region=centralus
+spring.cloud.azure.region=eastus2
+
+spring.cloud.azure.auto-create-resources=true

sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/microsoft/azure/spring/cloud/autoconfigure/cache/AzureRedisAutoConfiguration.java

@@ -3,12 +3,11 @@
 
 package com.microsoft.azure.spring.cloud.autoconfigure.cache;
 
-import com.microsoft.azure.management.redis.RedisCache;
-import com.microsoft.azure.spring.cloud.autoconfigure.context.AzureContextAutoConfiguration;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
-import com.microsoft.azure.spring.cloud.telemetry.TelemetryCollector;
+import java.util.Arrays;
+
+import javax.annotation.PostConstruct;
+
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -19,8 +18,12 @@
 import org.springframework.context.annotation.Primary;
 import org.springframework.data.redis.core.RedisOperations;
 
-import javax.annotation.PostConstruct;
-import java.util.Arrays;
+import com.microsoft.azure.management.Azure;
+import com.microsoft.azure.management.redis.RedisCache;
+import com.microsoft.azure.spring.cloud.autoconfigure.context.AzureContextAutoConfiguration;
+import com.microsoft.azure.spring.cloud.context.core.config.AzureProperties;
+import com.microsoft.azure.spring.cloud.context.core.impl.RedisCacheManager;
+import com.microsoft.azure.spring.cloud.telemetry.TelemetryCollector;
 
 /**
  * An auto-configuration for Spring cache using Azure redis cache
@@ -31,7 +34,6 @@
 @AutoConfigureAfter(AzureContextAutoConfiguration.class)
 @ConditionalOnProperty(value = "spring.cloud.azure.redis.enabled", matchIfMissing = true)
 @ConditionalOnClass(RedisOperations.class)
-@ConditionalOnBean(ResourceManagerProvider.class)
 @EnableConfigurationProperties(AzureRedisProperties.class)
 public class AzureRedisAutoConfiguration {
     private static final String REDIS = "Redis";
@@ -41,14 +43,19 @@ public void collectTelemetry() {
         TelemetryCollector.getInstance().addService(REDIS);
     }
 
+    @ConditionalOnMissingBean
+    @Bean
+    public RedisCacheManager redisCacheManager(Azure azure, AzureProperties azureProperties) {
+        return new RedisCacheManager(azure, azureProperties);
+    }
+
     @ConditionalOnMissingBean
     @Primary
     @Bean
-    public RedisProperties redisProperties(ResourceManagerProvider resourceManagerProvider,
-                                           AzureRedisProperties azureRedisProperties) {
+    public RedisProperties redisProperties(AzureRedisProperties azureRedisProperties, RedisCacheManager redisCacheManager) {
         String cacheName = azureRedisProperties.getName();
 
-        RedisCache redisCache = resourceManagerProvider.getRedisCacheManager().getOrCreate(cacheName);
+        RedisCache redisCache = redisCacheManager.getOrCreate(cacheName);
 
         RedisProperties redisProperties = new RedisProperties();
 

sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/microsoft/azure/spring/cloud/autoconfigure/eventhub/AzureEventHubAutoConfiguration.java

@@ -3,30 +3,32 @@
 
 package com.microsoft.azure.spring.cloud.autoconfigure.eventhub;
 
+import javax.annotation.PostConstruct;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.util.StringUtils;
+
 import com.azure.messaging.eventhubs.EventHubConsumerAsyncClient;
+import com.azure.resourcemanager.storage.models.StorageAccount;
 import com.microsoft.azure.management.eventhub.EventHubNamespace;
-import com.microsoft.azure.management.storage.StorageAccount;
 import com.microsoft.azure.spring.cloud.autoconfigure.context.AzureContextAutoConfiguration;
 import com.microsoft.azure.spring.cloud.context.core.api.EnvironmentProvider;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
+import com.microsoft.azure.spring.cloud.context.core.impl.EventHubNamespaceManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.StorageAccountManager;
 import com.microsoft.azure.spring.cloud.context.core.storage.StorageConnectionStringProvider;
 import com.microsoft.azure.spring.cloud.telemetry.TelemetryCollector;
 import com.microsoft.azure.spring.integration.eventhub.api.EventHubClientFactory;
 import com.microsoft.azure.spring.integration.eventhub.api.EventHubOperation;
 import com.microsoft.azure.spring.integration.eventhub.factory.DefaultEventHubClientFactory;
 import com.microsoft.azure.spring.integration.eventhub.factory.EventHubConnectionStringProvider;
 import com.microsoft.azure.spring.integration.eventhub.impl.EventHubTemplate;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.AutoConfigureAfter;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.util.StringUtils;
-
-import javax.annotation.PostConstruct;
 
 /**
  * An auto-configuration for Event Hub, which provides {@link EventHubOperation}
@@ -43,7 +45,10 @@ public class AzureEventHubAutoConfiguration {
     private static final String NAMESPACE = "Namespace";
 
     @Autowired(required = false)
-    private ResourceManagerProvider resourceManagerProvider;
+    private EventHubNamespaceManager eventHubNamespaceManager;
+
+    @Autowired(required = false)
+    private StorageAccountManager storageAccountManager;
 
     @PostConstruct
     public void collectTelemetry() {
@@ -59,10 +64,9 @@ public EventHubOperation eventHubOperation(EventHubClientFactory clientFactory)
     @Bean
     @ConditionalOnMissingBean
     public EventHubConnectionStringProvider eventHubConnectionStringProvider(
-        AzureEventHubProperties eventHubProperties) {
-        if (resourceManagerProvider != null) {
-            EventHubNamespace namespace = resourceManagerProvider.getEventHubNamespaceManager()
-                .getOrCreate(eventHubProperties.getNamespace());
+            AzureEventHubProperties eventHubProperties) {
+        if (eventHubNamespaceManager != null) {
+            EventHubNamespace namespace = eventHubNamespaceManager.getOrCreate(eventHubProperties.getNamespace());
             return new EventHubConnectionStringProvider(namespace);
         } else {
             String connectionString = eventHubProperties.getConnectionString();
@@ -71,29 +75,29 @@ public EventHubConnectionStringProvider eventHubConnectionStringProvider(
                 throw new IllegalArgumentException("Event hubs connection string cannot be empty");
             }
 
-            TelemetryCollector.getInstance()
-                .addProperty(EVENT_HUB, NAMESPACE, EventHubUtils.getNamespace(connectionString));
+            TelemetryCollector.getInstance().addProperty(EVENT_HUB, NAMESPACE,
+                    EventHubUtils.getNamespace(connectionString));
             return new EventHubConnectionStringProvider(connectionString);
         }
     }
 
     @Bean
     @ConditionalOnMissingBean
     public EventHubClientFactory clientFactory(EventHubConnectionStringProvider connectionStringProvider,
-                                               AzureEventHubProperties eventHubProperties, EnvironmentProvider environmentProvider) {
+            AzureEventHubProperties eventHubProperties, EnvironmentProvider environmentProvider) {
         String checkpointConnectionString;
-        if (resourceManagerProvider != null) {
-            StorageAccount checkpointStorageAccount = resourceManagerProvider.getStorageAccountManager().getOrCreate(
-                eventHubProperties.getCheckpointStorageAccount());
-            checkpointConnectionString = StorageConnectionStringProvider
-                .getConnectionString(checkpointStorageAccount, environmentProvider.getEnvironment());
+        if (storageAccountManager != null) {
+            StorageAccount checkpointStorageAccount = storageAccountManager
+                    .getOrCreate(eventHubProperties.getCheckpointStorageAccount());
+            checkpointConnectionString = StorageConnectionStringProvider.getConnectionString(checkpointStorageAccount,
+                    environmentProvider.getEnvironment());
         } else {
-            checkpointConnectionString = StorageConnectionStringProvider
-                .getConnectionString(eventHubProperties.getCheckpointStorageAccount(),
-                    eventHubProperties.getCheckpointAccessKey(), environmentProvider.getEnvironment());
+            checkpointConnectionString = StorageConnectionStringProvider.getConnectionString(
+                    eventHubProperties.getCheckpointStorageAccount(), eventHubProperties.getCheckpointAccessKey(),
+                    environmentProvider.getEnvironment());
         }
 
         return new DefaultEventHubClientFactory(connectionStringProvider, checkpointConnectionString,
-            eventHubProperties.getCheckpointContainer());
+                eventHubProperties.getCheckpointContainer());
     }
 }

sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/microsoft/azure/spring/cloud/autoconfigure/eventhub/AzureEventHubKafkaAutoConfiguration.java

@@ -3,12 +3,10 @@
 
 package com.microsoft.azure.spring.cloud.autoconfigure.eventhub;
 
-import com.microsoft.azure.management.eventhub.AuthorizationRule;
-import com.microsoft.azure.management.eventhub.EventHubAuthorizationKey;
-import com.microsoft.azure.management.eventhub.EventHubNamespace;
-import com.microsoft.azure.spring.cloud.autoconfigure.context.AzureContextAutoConfiguration;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
-import com.microsoft.azure.spring.cloud.telemetry.TelemetryCollector;
+import java.util.Arrays;
+
+import javax.annotation.PostConstruct;
+
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -19,8 +17,12 @@
 import org.springframework.context.annotation.Primary;
 import org.springframework.kafka.core.KafkaTemplate;
 
-import javax.annotation.PostConstruct;
-import java.util.Arrays;
+import com.microsoft.azure.management.eventhub.AuthorizationRule;
+import com.microsoft.azure.management.eventhub.EventHubAuthorizationKey;
+import com.microsoft.azure.management.eventhub.EventHubNamespace;
+import com.microsoft.azure.spring.cloud.autoconfigure.context.AzureContextAutoConfiguration;
+import com.microsoft.azure.spring.cloud.context.core.impl.EventHubNamespaceManager;
+import com.microsoft.azure.spring.cloud.telemetry.TelemetryCollector;
 
 /**
  * An auto-configuration for Event Hub, which provides {@link KafkaProperties}
@@ -36,8 +38,7 @@ public class AzureEventHubKafkaAutoConfiguration {
     private static final String SECURITY_PROTOCOL = "security.protocol";
     private static final String SASL_SSL = "SASL_SSL";
     private static final String SASL_JAAS_CONFIG = "sasl.jaas.config";
-    private static final String SASL_CONFIG_VALUE =
-        "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"$ConnectionString\" "
+    private static final String SASL_CONFIG_VALUE = "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"$ConnectionString\" "
             + "password=\"%s\";%n";
     private static final String SASL_MECHANISM = "sasl.mechanism";
     private static final String SASL_MECHANISM_PLAIN = "PLAIN";
@@ -52,22 +53,21 @@ public void collectTelemetry() {
     @SuppressWarnings("rawtypes")
     @Primary
     @Bean
-    public KafkaProperties kafkaProperties(ResourceManagerProvider resourceManagerProvider,
-                                           AzureEventHubProperties eventHubProperties) {
+    public KafkaProperties kafkaProperties(EventHubNamespaceManager eventHubNamespaceManager,
+            AzureEventHubProperties eventHubProperties) {
         KafkaProperties kafkaProperties = new KafkaProperties();
-        EventHubNamespace namespace =
-            resourceManagerProvider.getEventHubNamespaceManager().getOrCreate(eventHubProperties.getNamespace());
-        String connectionString =
-            namespace.listAuthorizationRules().stream().findFirst().map(AuthorizationRule::getKeys)
-                .map(EventHubAuthorizationKey::primaryConnectionString).orElseThrow(() -> new RuntimeException(
-                String.format("Failed to fetch connection string of namespace '%s'", namespace), null));
+        EventHubNamespace namespace = eventHubNamespaceManager.getOrCreate(eventHubProperties.getNamespace());
+        String connectionString = namespace.listAuthorizationRules().stream().findFirst()
+                .map(AuthorizationRule::getKeys).map(EventHubAuthorizationKey::primaryConnectionString)
+                .orElseThrow(() -> new RuntimeException(
+                        String.format("Failed to fetch connection string of namespace '%s'", namespace), null));
         String endpoint = namespace.serviceBusEndpoint();
         String endpointHost = endpoint.substring("https://".length(), endpoint.lastIndexOf(':'));
         kafkaProperties.setBootstrapServers(Arrays.asList(endpointHost + ":" + PORT));
         kafkaProperties.getProperties().put(SECURITY_PROTOCOL, SASL_SSL);
         kafkaProperties.getProperties().put(SASL_MECHANISM, SASL_MECHANISM_PLAIN);
         kafkaProperties.getProperties().put(SASL_JAAS_CONFIG,
-            String.format(SASL_CONFIG_VALUE, connectionString, System.getProperty("line.separator")));
+                String.format(SASL_CONFIG_VALUE, connectionString, System.getProperty("line.separator")));
         return kafkaProperties;
     }
 }

sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/microsoft/azure/spring/cloud/autoconfigure/servicebus/AzureServiceBusAutoConfiguration.java

@@ -3,14 +3,8 @@
 
 package com.microsoft.azure.spring.cloud.autoconfigure.servicebus;
 
-import com.microsoft.azure.management.servicebus.AuthorizationKeys;
-import com.microsoft.azure.management.servicebus.ServiceBusNamespace;
-import com.microsoft.azure.servicebus.IMessage;
-import com.microsoft.azure.servicebus.primitives.ConnectionStringBuilder;
-import com.microsoft.azure.spring.cloud.autoconfigure.context.AzureContextAutoConfiguration;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManager;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
-import com.microsoft.azure.spring.cloud.telemetry.TelemetryCollector;
+import javax.annotation.PostConstruct;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
@@ -22,7 +16,14 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.util.StringUtils;
 
-import javax.annotation.PostConstruct;
+import com.microsoft.azure.management.servicebus.AuthorizationKeys;
+import com.microsoft.azure.management.servicebus.ServiceBusNamespace;
+import com.microsoft.azure.servicebus.IMessage;
+import com.microsoft.azure.servicebus.primitives.ConnectionStringBuilder;
+import com.microsoft.azure.spring.cloud.autoconfigure.context.AzureContextAutoConfiguration;
+import com.microsoft.azure.spring.cloud.context.core.api.ResourceManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusNamespaceManager;
+import com.microsoft.azure.spring.cloud.telemetry.TelemetryCollector;
 
 /**
  * An auto-configuration for Service Bus
@@ -55,15 +56,13 @@ public void collectTelemetry() {
     }
 
     @Bean
-    @ConditionalOnBean(ResourceManagerProvider.class)
-    public AzureServiceBusProperties serviceBusProperties(ResourceManagerProvider resourceManagerProvider,
+    @ConditionalOnBean(ServiceBusNamespaceManager.class)
+    public AzureServiceBusProperties serviceBusProperties(ServiceBusNamespaceManager serviceBusNamespaceManager,
                                                           AzureServiceBusProperties serviceBusProperties) {
         if (!StringUtils.hasText(serviceBusProperties.getConnectionString())) {
-            ResourceManager<ServiceBusNamespace, String> serviceBusNamespaceManager =
-                resourceManagerProvider.getServiceBusNamespaceManager();
             serviceBusNamespaceManager.getOrCreate(serviceBusProperties.getNamespace());
             serviceBusProperties.setConnectionString(
-                buildConnectionString(resourceManagerProvider.getServiceBusNamespaceManager(),
+                buildConnectionString(serviceBusNamespaceManager,
                     serviceBusProperties.getNamespace()));
             LOG.info("'spring.cloud.azure.servicebus.connection-string' auto configured");
         }

sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/microsoft/azure/spring/cloud/autoconfigure/servicebus/AzureServiceBusQueueAutoConfiguration.java

@@ -3,14 +3,8 @@
 
 package com.microsoft.azure.spring.cloud.autoconfigure.servicebus;
 
-import com.microsoft.azure.servicebus.QueueClient;
-import com.microsoft.azure.spring.cloud.autoconfigure.context.AzureContextAutoConfiguration;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
-import com.microsoft.azure.spring.cloud.telemetry.TelemetryCollector;
-import com.microsoft.azure.spring.integration.servicebus.factory.DefaultServiceBusQueueClientFactory;
-import com.microsoft.azure.spring.integration.servicebus.factory.ServiceBusQueueClientFactory;
-import com.microsoft.azure.spring.integration.servicebus.queue.ServiceBusQueueOperation;
-import com.microsoft.azure.spring.integration.servicebus.queue.ServiceBusQueueTemplate;
+import javax.annotation.PostConstruct;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
@@ -19,7 +13,18 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
-import javax.annotation.PostConstruct;
+import com.microsoft.azure.management.Azure;
+import com.microsoft.azure.servicebus.QueueClient;
+import com.microsoft.azure.spring.cloud.autoconfigure.context.AzureContextAutoConfiguration;
+import com.microsoft.azure.spring.cloud.context.core.config.AzureProperties;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusNamespaceManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusQueueManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusTopicManager;
+import com.microsoft.azure.spring.cloud.telemetry.TelemetryCollector;
+import com.microsoft.azure.spring.integration.servicebus.factory.DefaultServiceBusQueueClientFactory;
+import com.microsoft.azure.spring.integration.servicebus.factory.ServiceBusQueueClientFactory;
+import com.microsoft.azure.spring.integration.servicebus.queue.ServiceBusQueueOperation;
+import com.microsoft.azure.spring.integration.servicebus.queue.ServiceBusQueueTemplate;
 
 /**
  * An auto-configuration for Service Bus queue
@@ -28,14 +33,20 @@
  */
 @Configuration
 @AutoConfigureAfter(AzureContextAutoConfiguration.class)
-@ConditionalOnClass(value = {QueueClient.class, ServiceBusQueueClientFactory.class})
+@ConditionalOnClass(value = { QueueClient.class, ServiceBusQueueClientFactory.class })
 @ConditionalOnProperty(value = "spring.cloud.azure.servicebus.enabled", matchIfMissing = true)
 public class AzureServiceBusQueueAutoConfiguration {
     private static final String SERVICE_BUS_QUEUE = "ServiceBusQueue";
     private static final String NAMESPACE = "Namespace";
 
     @Autowired(required = false)
-    private ResourceManagerProvider resourceManagerProvider;
+    private ServiceBusNamespaceManager serviceBusNamespaceManager;
+
+    @Autowired(required = false)
+    private ServiceBusQueueManager serviceBusQueueManager;
+
+    @Autowired(required = false)
+    private ServiceBusTopicManager serviceBusTopicManager;
 
     @PostConstruct
     public void collectTelemetry() {
@@ -46,15 +57,16 @@ public void collectTelemetry() {
     @ConditionalOnMissingBean
     public ServiceBusQueueClientFactory queueClientFactory(AzureServiceBusProperties serviceBusProperties) {
         String connectionString = serviceBusProperties.getConnectionString();
-        DefaultServiceBusQueueClientFactory clientFactory =
-            new DefaultServiceBusQueueClientFactory(serviceBusProperties.getConnectionString());
+        DefaultServiceBusQueueClientFactory clientFactory = new DefaultServiceBusQueueClientFactory(
+                serviceBusProperties.getConnectionString());
 
-        if (resourceManagerProvider != null) {
-            clientFactory.setResourceManagerProvider(resourceManagerProvider);
+        if (serviceBusNamespaceManager != null && serviceBusQueueManager != null && serviceBusTopicManager != null) {
+            clientFactory.setServiceBusNamespaceManager(serviceBusNamespaceManager);
+            clientFactory.setServiceBusQueueManager(serviceBusQueueManager);
             clientFactory.setNamespace(serviceBusProperties.getNamespace());
         } else {
             TelemetryCollector.getInstance().addProperty(SERVICE_BUS_QUEUE, NAMESPACE,
-                ServiceBusUtils.getNamespace(connectionString));
+                    ServiceBusUtils.getNamespace(connectionString));
         }
 
         return clientFactory;
@@ -65,4 +77,16 @@ public ServiceBusQueueClientFactory queueClientFactory(AzureServiceBusProperties
     public ServiceBusQueueOperation queueOperation(ServiceBusQueueClientFactory factory) {
         return new ServiceBusQueueTemplate(factory);
     }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public ServiceBusNamespaceManager serviceBusNamespaceManager(Azure azure, AzureProperties azureProperties) {
+        return new ServiceBusNamespaceManager(azure, azureProperties);
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public ServiceBusQueueManager serviceBusQueueManager(Azure azure, AzureProperties azureProperties) {
+        return new ServiceBusQueueManager(azure, azureProperties);
+    }
 }

sdk/spring/azure-spring-cloud-autoconfigure/src/main/java/com/microsoft/azure/spring/cloud/autoconfigure/servicebus/AzureServiceBusTopicAutoConfiguration.java

@@ -3,14 +3,8 @@
 
 package com.microsoft.azure.spring.cloud.autoconfigure.servicebus;
 
-import com.microsoft.azure.servicebus.TopicClient;
-import com.microsoft.azure.spring.cloud.autoconfigure.context.AzureContextAutoConfiguration;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
-import com.microsoft.azure.spring.cloud.telemetry.TelemetryCollector;
-import com.microsoft.azure.spring.integration.servicebus.factory.DefaultServiceBusTopicClientFactory;
-import com.microsoft.azure.spring.integration.servicebus.factory.ServiceBusTopicClientFactory;
-import com.microsoft.azure.spring.integration.servicebus.topic.ServiceBusTopicOperation;
-import com.microsoft.azure.spring.integration.servicebus.topic.ServiceBusTopicTemplate;
+import javax.annotation.PostConstruct;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
@@ -19,7 +13,15 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
-import javax.annotation.PostConstruct;
+import com.microsoft.azure.servicebus.TopicClient;
+import com.microsoft.azure.spring.cloud.autoconfigure.context.AzureContextAutoConfiguration;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusNamespaceManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusTopicSubscriptionManager;
+import com.microsoft.azure.spring.cloud.telemetry.TelemetryCollector;
+import com.microsoft.azure.spring.integration.servicebus.factory.DefaultServiceBusTopicClientFactory;
+import com.microsoft.azure.spring.integration.servicebus.factory.ServiceBusTopicClientFactory;
+import com.microsoft.azure.spring.integration.servicebus.topic.ServiceBusTopicOperation;
+import com.microsoft.azure.spring.integration.servicebus.topic.ServiceBusTopicTemplate;
 
 /**
  * An auto-configuration for Service Bus topic
@@ -34,8 +36,12 @@ public class AzureServiceBusTopicAutoConfiguration {
     private static final String SERVICE_BUS_TOPIC = "ServiceBusTopic";
     private static final String NAMESPACE = "Namespace";
 
+
+    @Autowired(required = false)
+    private ServiceBusTopicSubscriptionManager serviceBusTopicSubscriptionManager;
+
     @Autowired(required = false)
-    private ResourceManagerProvider resourceManagerProvider;
+    private ServiceBusNamespaceManager serviceBusNamespaceManager;
 
     @PostConstruct
     public void collectTelemetry() {
@@ -49,9 +55,10 @@ public ServiceBusTopicClientFactory topicClientFactory(AzureServiceBusProperties
         DefaultServiceBusTopicClientFactory clientFactory =
             new DefaultServiceBusTopicClientFactory(serviceBusProperties.getConnectionString());
 
-        if (resourceManagerProvider != null) {
+        if (serviceBusTopicSubscriptionManager != null && serviceBusNamespaceManager != null) {
             clientFactory.setNamespace(serviceBusProperties.getNamespace());
-            clientFactory.setResourceManagerProvider(resourceManagerProvider);
+            clientFactory.setServiceBusNamespaceManager(serviceBusNamespaceManager);
+            clientFactory.setServiceBusTopicSubscriptionManager(serviceBusTopicSubscriptionManager);
         } else {
             TelemetryCollector.getInstance().addProperty(SERVICE_BUS_TOPIC, NAMESPACE,
                 ServiceBusUtils.getNamespace(connectionString));

sdk/spring/azure-spring-cloud-autoconfigure/src/test/java/com/microsoft/azure/spring/cloud/autoconfigure/cache/AzureRedisAutoConfigurationTest.java

@@ -3,10 +3,11 @@
 
 package com.microsoft.azure.spring.cloud.autoconfigure.cache;
 
-import com.microsoft.azure.management.redis.RedisAccessKeys;
-import com.microsoft.azure.management.redis.RedisCache;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
-import com.microsoft.azure.spring.cloud.context.core.impl.RedisCacheManager;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.isA;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
 import org.junit.Test;
 import org.springframework.boot.autoconfigure.AutoConfigurations;
 import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
@@ -16,10 +17,9 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.data.redis.core.RedisOperations;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.isA;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
+import com.microsoft.azure.management.redis.RedisAccessKeys;
+import com.microsoft.azure.management.redis.RedisCache;
+import com.microsoft.azure.spring.cloud.context.core.impl.RedisCacheManager;
 
 public class AzureRedisAutoConfigurationTest {
     private static final String KEY = "KEY";
@@ -66,9 +66,9 @@ public void testAzureRedisPropertiesConfigured() {
     static class TestConfiguration {
 
         @Bean
-        ResourceManagerProvider resourceManagerProvider() {
+        RedisCacheManager redisCacheManager() {
 
-            ResourceManagerProvider resourceManagerProvider = mock(ResourceManagerProvider.class);
+
             RedisCacheManager redisCacheManager = mock(RedisCacheManager.class);
             RedisCache redisCache = mock(RedisCache.class);
             RedisAccessKeys accessKeys = mock(RedisAccessKeys.class);
@@ -78,9 +78,8 @@ ResourceManagerProvider resourceManagerProvider() {
             when(redisCache.sslPort()).thenReturn(PORT);
             when(redisCache.shardCount()).thenReturn(0);
             when(redisCache.getKeys()).thenReturn(accessKeys);
-            when(resourceManagerProvider.getRedisCacheManager()).thenReturn(redisCacheManager);
             when(redisCacheManager.getOrCreate(isA(String.class))).thenReturn(redisCache);
-            return resourceManagerProvider;
+            return redisCacheManager;
         }
 
     }

sdk/spring/azure-spring-cloud-autoconfigure/src/test/java/com/microsoft/azure/spring/cloud/autoconfigure/context/AzureContextAutoConfigurationTest.java

@@ -3,21 +3,21 @@
 
 package com.microsoft.azure.spring.cloud.autoconfigure.context;
 
-import com.microsoft.azure.AzureEnvironment;
-import com.microsoft.azure.credentials.AzureTokenCredentials;
-import com.microsoft.azure.management.Azure;
-import com.microsoft.azure.spring.cloud.context.core.api.CredentialsProvider;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
-import com.microsoft.azure.spring.cloud.context.core.config.AzureProperties;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
 import org.junit.Test;
 import org.springframework.boot.autoconfigure.AutoConfigurations;
 import org.springframework.boot.test.context.FilteredClassLoader;
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.Mockito.mock;
+import com.microsoft.azure.AzureEnvironment;
+import com.microsoft.azure.credentials.AzureTokenCredentials;
+import com.microsoft.azure.management.Azure;
+import com.microsoft.azure.spring.cloud.context.core.api.CredentialsProvider;
+import com.microsoft.azure.spring.cloud.context.core.config.AzureProperties;
 
 public class AzureContextAutoConfigurationTest {
     private ApplicationContextRunner contextRunner =
@@ -80,11 +80,6 @@ CredentialsProvider credentialsProvider() {
             return mock(CredentialsProvider.class);
         }
 
-        @Bean
-        ResourceManagerProvider resourceManagerProvider() {
-            return mock(ResourceManagerProvider.class);
-        }
-
         @Bean
         AzureTokenCredentials credentials() {
             return mock(AzureTokenCredentials.class);

sdk/spring/azure-spring-cloud-autoconfigure/src/test/java/com/microsoft/azure/spring/cloud/autoconfigure/eventhub/AzureEventHubKafkaAutoConfigurationTest.java

@@ -3,13 +3,12 @@
 
 package com.microsoft.azure.spring.cloud.autoconfigure.eventhub;
 
-import com.microsoft.azure.Page;
-import com.microsoft.azure.PagedList;
-import com.microsoft.azure.management.eventhub.EventHubAuthorizationKey;
-import com.microsoft.azure.management.eventhub.EventHubNamespace;
-import com.microsoft.azure.management.eventhub.EventHubNamespaceAuthorizationRule;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
-import com.microsoft.rest.RestException;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+
 import org.junit.Ignore;
 import org.junit.Test;
 import org.springframework.boot.autoconfigure.AutoConfigurations;
@@ -20,11 +19,13 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.core.KafkaTemplate;
 
-import java.io.IOException;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
+import com.microsoft.azure.Page;
+import com.microsoft.azure.PagedList;
+import com.microsoft.azure.management.eventhub.EventHubAuthorizationKey;
+import com.microsoft.azure.management.eventhub.EventHubNamespace;
+import com.microsoft.azure.management.eventhub.EventHubNamespaceAuthorizationRule;
+import com.microsoft.azure.spring.cloud.context.core.impl.EventHubNamespaceManager;
+import com.microsoft.rest.RestException;
 
 public class AzureEventHubKafkaAutoConfigurationTest {
     private ApplicationContextRunner contextRunner = new ApplicationContextRunner()
@@ -72,9 +73,9 @@ public void testAzureEventHubPropertiesConfigured() {
     static class TestConfiguration {
 
         @Bean
-        ResourceManagerProvider resourceManagerProvider() {
+        EventHubNamespaceManager eventHubNamespaceManager() {
 
-            ResourceManagerProvider resourceManagerProvider = mock(ResourceManagerProvider.class);
+
             EventHubNamespace namespace = mock(EventHubNamespace.class);
             EventHubAuthorizationKey key = mock(EventHubAuthorizationKey.class);
             when(key.primaryConnectionString()).thenReturn("connectionString1");
@@ -90,7 +91,10 @@ public Page<EventHubNamespaceAuthorizationRule> nextPage(String nextPageLink)
             rules.add(rule);
             when(namespace.listAuthorizationRules()).thenReturn(rules);
             when(namespace.serviceBusEndpoint()).thenReturn("localhost");
-            return resourceManagerProvider;
+            EventHubNamespaceManager eventHubNamespaceManager = mock(EventHubNamespaceManager.class);
+            //This previously returned a ResourceManagerProvider that was in no way connected to any of the objets created above.
+            //Maintaining similar behavior in refactoring.
+            return eventHubNamespaceManager;
         }
 
     }

sdk/spring/azure-spring-cloud-autoconfigure/src/test/java/com/microsoft/azure/spring/cloud/autoconfigure/servicebus/ServiceBusTestConfiguration.java

@@ -3,24 +3,22 @@
 
 package com.microsoft.azure.spring.cloud.autoconfigure.servicebus;
 
-import com.microsoft.azure.management.servicebus.ServiceBusNamespace;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
-import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusNamesapceManager;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import com.microsoft.azure.management.servicebus.ServiceBusNamespace;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusNamespaceManager;
+
 @Configuration
 public class ServiceBusTestConfiguration {
     @Bean
-    ResourceManagerProvider resourceManagerProvider() {
-        ResourceManagerProvider resourceManagerProvider = mock(ResourceManagerProvider.class);
-        ServiceBusNamesapceManager namespaceManager = mock(ServiceBusNamesapceManager.class);
+    ServiceBusNamespaceManager namespaceManager() {
+        ServiceBusNamespaceManager namespaceManager = mock(ServiceBusNamespaceManager.class);
         when(namespaceManager.getOrCreate(anyString())).thenReturn(mock(ServiceBusNamespace.class));
-        when(resourceManagerProvider.getServiceBusNamespaceManager()).thenReturn(namespaceManager);
-        return resourceManagerProvider;
+        return namespaceManager;
     }
 }

sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/autoconfigure/context/AzureContextAutoConfiguration.java

@@ -3,6 +3,15 @@
 
 package com.microsoft.azure.spring.cloud.autoconfigure.context;
 
+import java.io.IOException;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
 import com.microsoft.azure.AzureEnvironment;
 import com.microsoft.azure.AzureResponseBuilder;
 import com.microsoft.azure.credentials.AzureTokenCredentials;
@@ -11,54 +20,42 @@
 import com.microsoft.azure.management.resources.fluentcore.utils.ResourceManagerThrottlingInterceptor;
 import com.microsoft.azure.serializer.AzureJacksonAdapter;
 import com.microsoft.azure.spring.cloud.context.core.api.CredentialsProvider;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
 import com.microsoft.azure.spring.cloud.context.core.config.AzureProperties;
-import com.microsoft.azure.spring.cloud.context.core.impl.AzureResourceManagerProvider;
 import com.microsoft.azure.spring.cloud.context.core.impl.DefaultCredentialsProvider;
 import com.microsoft.rest.RestClient;
-import java.io.IOException;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
 
 /**
- * Auto-config to provide default {@link CredentialsProvider} for all Azure services
+ * Auto-config to provide default {@link CredentialsProvider} for all Azure
+ * services
  *
  * @author Warren Zhu
  */
 @Configuration
 @EnableConfigurationProperties(AzureProperties.class)
 @ConditionalOnClass(Azure.class)
-@ConditionalOnProperty(prefix = "spring.cloud.azure", value = {"resource-group"})
+@ConditionalOnProperty(prefix = "spring.cloud.azure", value = { "resource-group" })
 public class AzureContextAutoConfiguration {
 
-    private static final String PROJECT_VERSION =
-        AzureContextAutoConfiguration.class.getPackage().getImplementationVersion();
+    private static final String PROJECT_VERSION = AzureContextAutoConfiguration.class.getPackage()
+            .getImplementationVersion();
     private static final String SPRING_CLOUD_USER_AGENT = "spring-cloud-azure/" + PROJECT_VERSION;
 
-    @Bean
-    @ConditionalOnMissingBean
-    public ResourceManagerProvider resourceManagerProvider(Azure azure, AzureProperties azureProperties) {
-        return new AzureResourceManagerProvider(azure, azureProperties);
-    }
 
     @Bean
     @ConditionalOnMissingBean
     public Azure azure(AzureTokenCredentials credentials) throws IOException {
         RestClient restClient = new RestClient.Builder()
-            .withBaseUrl(credentials.environment(), AzureEnvironment.Endpoint.RESOURCE_MANAGER)
-            .withCredentials(credentials).withSerializerAdapter(new AzureJacksonAdapter())
-            .withResponseBuilderFactory(new AzureResponseBuilder.Factory())
-            .withInterceptor(new ProviderRegistrationInterceptor(credentials))
-            .withInterceptor(new ResourceManagerThrottlingInterceptor()).withUserAgent(SPRING_CLOUD_USER_AGENT)
-            .build();
+                .withBaseUrl(credentials.environment(), AzureEnvironment.Endpoint.RESOURCE_MANAGER)
+                .withCredentials(credentials).withSerializerAdapter(new AzureJacksonAdapter())
+                .withResponseBuilderFactory(new AzureResponseBuilder.Factory())
+                .withInterceptor(new ProviderRegistrationInterceptor(credentials))
+                .withInterceptor(new ResourceManagerThrottlingInterceptor()).withUserAgent(SPRING_CLOUD_USER_AGENT)
+                .build();
 
         return Azure.authenticate(restClient, credentials.domain()).withDefaultSubscription();
     }
 
+
     @Bean
     @ConditionalOnMissingBean
     public AzureTokenCredentials credentials(AzureProperties azureProperties) {

sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/autoconfigure/context/AzureResourceManager20AutoConfiguration.java

@@ -0,0 +1,43 @@
+package com.microsoft.azure.spring.cloud.autoconfigure.context;
+
+import java.util.Arrays;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.env.Environment;
+
+import com.azure.core.credential.TokenCredential;
+import com.azure.core.management.profile.AzureProfile;
+import com.azure.resourcemanager.Azure;
+import com.microsoft.azure.AzureEnvironment;
+import com.microsoft.azure.identity.spring.SpringEnvironmentTokenBuilder;
+import com.microsoft.azure.spring.cloud.context.core.config.AzureProperties;
+
+@Configuration
+@EnableConfigurationProperties(AzureProperties.class)
+@ConditionalOnClass(Azure.class)
+@ConditionalOnProperty(prefix = "spring.cloud.azure", value = { "resource-group" })
+public class AzureResourceManager20AutoConfiguration {
+
+    @Bean
+    @ConditionalOnMissingBean
+    public com.azure.resourcemanager.Azure.Authenticated azure20(TokenCredential tokenCredential,
+            AzureProperties azureProperties) {
+        AzureEnvironment legacyEnvironment = azureProperties.getEnvironment();
+        com.azure.core.management.AzureEnvironment azureEnvironment = Arrays
+                .stream(com.azure.core.management.AzureEnvironment.knownEnvironments())
+                .filter(env -> env.getManagementEndpoint().equals(legacyEnvironment.managementEndpoint())).findFirst()
+                .get();
+        return com.azure.resourcemanager.Azure.authenticate(tokenCredential, new AzureProfile(azureEnvironment));
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public TokenCredential tokenCredential(Environment environment) {
+        return new SpringEnvironmentTokenBuilder().fromEnvironment(environment).build();
+    }
+}

sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/context/core/impl/AzureManager.java

@@ -3,24 +3,22 @@
 
 package com.microsoft.azure.spring.cloud.context.core.impl;
 
-import com.microsoft.azure.CloudException;
-import com.microsoft.azure.management.Azure;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManager;
-import com.microsoft.azure.spring.cloud.context.core.config.AzureProperties;
 import org.apache.commons.lang3.time.StopWatch;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.lang.NonNull;
 
+import com.microsoft.azure.CloudException;
+import com.microsoft.azure.spring.cloud.context.core.api.ResourceManager;
+import com.microsoft.azure.spring.cloud.context.core.config.AzureProperties;
+
 public abstract class AzureManager<T, K> implements ResourceManager<T, K> {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(AzureManager.class);
 
     protected final AzureProperties azureProperties;
-    protected final Azure azure;
 
-    public AzureManager(@NonNull Azure azure, @NonNull AzureProperties azureProperties) {
-        this.azure = azure;
+    public AzureManager(@NonNull AzureProperties azureProperties) {
         this.azureProperties = azureProperties;
     }
 

sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/context/core/impl/EventHubConsumerGroupManager.java

@@ -11,8 +11,11 @@
 
 public class EventHubConsumerGroupManager extends AzureManager<EventHubConsumerGroup, Tuple<EventHub, String>> {
 
+    private final Azure azure;
+
     public EventHubConsumerGroupManager(Azure azure, AzureProperties azureProperties) {
-        super(azure, azureProperties);
+        super(azureProperties);
+        this.azure = azure;
     }
 
     @Override

sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/context/core/impl/EventHubManager.java

@@ -10,9 +10,12 @@
 import com.microsoft.azure.spring.cloud.context.core.util.Tuple;
 
 public class EventHubManager extends AzureManager<EventHub, Tuple<EventHubNamespace, String>> {
+
+    private final Azure azure;
 
     public EventHubManager(Azure azure, AzureProperties azureProperties) {
-        super(azure, azureProperties);
+        super(azureProperties);
+        this.azure = azure;
     }
 
     @Override

sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/context/core/impl/EventHubNamesapceManager.java → sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/context/core/impl/EventHubNamespaceManager.java

@@ -7,10 +7,13 @@
 import com.microsoft.azure.management.eventhub.EventHubNamespace;
 import com.microsoft.azure.spring.cloud.context.core.config.AzureProperties;
 
-public class EventHubNamesapceManager extends AzureManager<EventHubNamespace, String> {
+public class EventHubNamespaceManager extends AzureManager<EventHubNamespace, String> {
 
-    public EventHubNamesapceManager(Azure azure, AzureProperties azureProperties) {
-        super(azure, azureProperties);
+    private final Azure azure;
+
+    public EventHubNamespaceManager(Azure azure, AzureProperties azureProperties) {
+        super(azureProperties);
+        this.azure = azure;
     }
 
     @Override
@@ -28,7 +31,8 @@ public EventHubNamespace internalGet(String namespace) {
         try {
             return azure.eventHubNamespaces().getByResourceGroup(azureProperties.getResourceGroup(), namespace);
         } catch (NullPointerException e) {
-            // azure management api has no way to determine whether an eventhub namespace exists
+            // azure management api has no way to determine whether an eventhub namespace
+            // exists
             // Workaround for this is by catching NPE
             return null;
         }
@@ -37,6 +41,6 @@ public EventHubNamespace internalGet(String namespace) {
     @Override
     public EventHubNamespace internalCreate(String namespace) {
         return azure.eventHubNamespaces().define(namespace).withRegion(azureProperties.getRegion())
-            .withExistingResourceGroup(azureProperties.getResourceGroup()).create();
+                .withExistingResourceGroup(azureProperties.getResourceGroup()).create();
     }
 }

sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/context/core/impl/RedisCacheManager.java

@@ -9,8 +9,11 @@
 
 public class RedisCacheManager extends AzureManager<RedisCache, String> {
 
+    private final Azure azure;
+
     public RedisCacheManager(Azure azure, AzureProperties azureProperties) {
-        super(azure, azureProperties);
+        super(azureProperties);
+        this.azure = azure;
     }
 
     @Override
@@ -31,6 +34,6 @@ public RedisCache internalGet(String name) {
     @Override
     public RedisCache internalCreate(String name) {
         return azure.redisCaches().define(name).withRegion(azureProperties.getRegion())
-            .withExistingResourceGroup(azureProperties.getResourceGroup()).withBasicSku().create();
+                .withExistingResourceGroup(azureProperties.getResourceGroup()).withBasicSku().create();
     }
 }

sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/context/core/impl/ResourceGroupManager.java

@@ -9,8 +9,11 @@
 
 public class ResourceGroupManager extends AzureManager<ResourceGroup, String> {
 
+    private final Azure azure;
+
     public ResourceGroupManager(Azure azure, AzureProperties azureProperties) {
-        super(azure, azureProperties);
+        super(azureProperties);
+        this.azure = azure;
     }
 
     @Override

sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/context/core/impl/ServiceBusNamesapceManager.java → sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/context/core/impl/ServiceBusNamespaceManager.java

@@ -7,10 +7,13 @@
 import com.microsoft.azure.management.servicebus.ServiceBusNamespace;
 import com.microsoft.azure.spring.cloud.context.core.config.AzureProperties;
 
-public class ServiceBusNamesapceManager extends AzureManager<ServiceBusNamespace, String> {
+public class ServiceBusNamespaceManager extends AzureManager<ServiceBusNamespace, String> {
 
-    public ServiceBusNamesapceManager(Azure azure, AzureProperties azureProperties) {
-        super(azure, azureProperties);
+    private final Azure azure;
+
+    public ServiceBusNamespaceManager(Azure azure, AzureProperties azureProperties) {
+        super(azureProperties);
+        this.azure = azure;
     }
 
     @Override
@@ -28,7 +31,8 @@ public ServiceBusNamespace internalGet(String namespace) {
         try {
             return azure.serviceBusNamespaces().getByResourceGroup(azureProperties.getResourceGroup(), namespace);
         } catch (NullPointerException e) {
-            // azure management api has no way to determine whether an eventhub namespace exists
+            // azure management api has no way to determine whether an eventhub namespace
+            // exists
             // Workaround for this is by catching NPE
             return null;
         }
@@ -37,6 +41,6 @@ public ServiceBusNamespace internalGet(String namespace) {
     @Override
     public ServiceBusNamespace internalCreate(String namespace) {
         return azure.serviceBusNamespaces().define(namespace).withRegion(azureProperties.getRegion())
-            .withExistingResourceGroup(azureProperties.getResourceGroup()).create();
+                .withExistingResourceGroup(azureProperties.getResourceGroup()).create();
     }
 }

sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/context/core/impl/ServiceBusQueueManager.java

@@ -11,8 +11,11 @@
 
 public class ServiceBusQueueManager extends AzureManager<Queue, Tuple<ServiceBusNamespace, String>> {
 
+    private final Azure azure;
+
     public ServiceBusQueueManager(Azure azure, AzureProperties azureProperties) {
-        super(azure, azureProperties);
+        super(azureProperties);
+        this.azure = azure;
     }
 
     @Override

sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/context/core/impl/ServiceBusTopicManager.java

@@ -11,8 +11,10 @@
 
 public class ServiceBusTopicManager extends AzureManager<Topic, Tuple<ServiceBusNamespace, String>> {
 
+    private final Azure azure;
     public ServiceBusTopicManager(Azure azure, AzureProperties azureProperties) {
-        super(azure, azureProperties);
+        super(azureProperties);
+        this.azure = azure;
     }
 
     @Override

sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/context/core/impl/ServiceBusTopicSubscriptionManager.java

@@ -11,8 +11,11 @@
 
 public class ServiceBusTopicSubscriptionManager extends AzureManager<ServiceBusSubscription, Tuple<Topic, String>> {
 
+    private final Azure azure;
+
     public ServiceBusTopicSubscriptionManager(Azure azure, AzureProperties azureProperties) {
-        super(azure, azureProperties);
+        super(azureProperties);
+        this.azure = azure;
     }
 
     @Override
@@ -30,7 +33,8 @@ public ServiceBusSubscription internalGet(Tuple<Topic, String> topicAndSubscript
         try {
             return topicAndSubscriptionName.getFirst().subscriptions().getByName(topicAndSubscriptionName.getSecond());
         } catch (NullPointerException ignore) {
-            // azure management api has no way to determine whether an service bus topic subscription exists
+            // azure management api has no way to determine whether an service bus topic
+            // subscription exists
             // Workaround for this is by catching NPE
             return null;
         }
@@ -39,6 +43,6 @@ public ServiceBusSubscription internalGet(Tuple<Topic, String> topicAndSubscript
     @Override
     public ServiceBusSubscription internalCreate(Tuple<Topic, String> topicAndSubscriptionName) {
         return topicAndSubscriptionName.getFirst().subscriptions().define(topicAndSubscriptionName.getSecond())
-            .create();
+                .create();
     }
 }

sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/context/core/impl/StorageAccountManager.java

@@ -3,14 +3,19 @@
 
 package com.microsoft.azure.spring.cloud.context.core.impl;
 
-import com.microsoft.azure.management.Azure;
-import com.microsoft.azure.management.storage.StorageAccount;
+import javax.annotation.Nonnull;
+
+import com.azure.resourcemanager.Azure;
+import com.azure.resourcemanager.storage.models.StorageAccount;
 import com.microsoft.azure.spring.cloud.context.core.config.AzureProperties;
 
 public class StorageAccountManager extends AzureManager<StorageAccount, String> {
+
+    private final Azure azure;
 
-    public StorageAccountManager(Azure azure, AzureProperties azureProperties) {
-        super(azure, azureProperties);
+    public StorageAccountManager(@Nonnull Azure azure, AzureProperties azureProperties) {
+        super(azureProperties);
+        this.azure = azure;
     }
 
     @Override
@@ -25,6 +30,7 @@ String getResourceType() {
 
     @Override
     public StorageAccount internalGet(String key) {
+
         return azure.storageAccounts().getByResourceGroup(azureProperties.getResourceGroup(), key);
     }
 

sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/context/core/impl/StorageQueueManager.java

@@ -14,8 +14,11 @@
 
 public class StorageQueueManager extends AzureManager<CloudQueue, Tuple<CloudStorageAccount, String>> {
 
+    private final Azure azure;
+
     public StorageQueueManager(Azure azure, AzureProperties azureProperties) {
-        super(azure, azureProperties);
+        super(azureProperties);
+        this.azure = azure;
     }
 
     @Override

sdk/spring/azure-spring-cloud-context/src/main/java/com/microsoft/azure/spring/cloud/context/core/storage/StorageConnectionStringProvider.java

@@ -3,13 +3,13 @@
 
 package com.microsoft.azure.spring.cloud.context.core.storage;
 
+import com.azure.resourcemanager.storage.models.StorageAccount;
 import com.microsoft.azure.AzureEnvironment;
-import com.microsoft.azure.management.storage.StorageAccount;
 
 public class StorageConnectionStringProvider {
 
     public static String getConnectionString(StorageAccount storageAccount, AzureEnvironment environment,
-        boolean isSecureTransfer) {
+            boolean isSecureTransfer) {
         return buildConnectionString(storageAccount, environment, isSecureTransfer);
     }
 
@@ -22,14 +22,15 @@ public static String getConnectionString(String storageAccount, String accessKey
     }
 
     public static String getConnectionString(String storageAccount, String accessKey, AzureEnvironment environment,
-        boolean isSecureTransfer) {
+            boolean isSecureTransfer) {
         return StorageConnectionStringBuilder.build(storageAccount, accessKey, environment, isSecureTransfer);
     }
 
     private static String buildConnectionString(StorageAccount storageAccount, AzureEnvironment environment,
-        boolean isSecureTransfer) {
-        return storageAccount.getKeys().stream().findFirst().map(key -> StorageConnectionStringBuilder
-            .build(storageAccount.name(), key.value(), environment, isSecureTransfer))
-            .orElseThrow(() -> new RuntimeException("Storage account key is empty."));
+            boolean isSecureTransfer) {
+        return storageAccount
+                .getKeys().stream().findFirst().map(key -> StorageConnectionStringBuilder.build(storageAccount.name(),
+                        key.value(), environment, isSecureTransfer))
+                .orElseThrow(() -> new RuntimeException("Storage account key is empty."));
     }
 }

sdk/spring/azure-spring-cloud-eventhubs-stream-binder/src/main/java/com/microsoft/azure/eventhub/stream/binder/config/EventHubBinderConfiguration.java

@@ -3,6 +3,16 @@
 
 package com.microsoft.azure.eventhub.stream.binder.config;
 
+import javax.annotation.PostConstruct;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+
 import com.microsoft.azure.eventhub.stream.binder.EventHubMessageChannelBinder;
 import com.microsoft.azure.eventhub.stream.binder.properties.EventHubExtendedBindingProperties;
 import com.microsoft.azure.eventhub.stream.binder.provisioning.EventHubChannelProvisioner;
@@ -11,18 +21,11 @@
 import com.microsoft.azure.spring.cloud.autoconfigure.eventhub.AzureEventHubAutoConfiguration;
 import com.microsoft.azure.spring.cloud.autoconfigure.eventhub.AzureEventHubProperties;
 import com.microsoft.azure.spring.cloud.autoconfigure.eventhub.EventHubUtils;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
+import com.microsoft.azure.spring.cloud.context.core.impl.EventHubConsumerGroupManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.EventHubManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.EventHubNamespaceManager;
 import com.microsoft.azure.spring.cloud.telemetry.TelemetryCollector;
 import com.microsoft.azure.spring.integration.eventhub.api.EventHubOperation;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.cloud.stream.binder.Binder;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Import;
-
-import javax.annotation.PostConstruct;
 
 /**
  * @author Warren Zhu
@@ -36,8 +39,16 @@ public class EventHubBinderConfiguration {
     private static final String EVENT_HUB_BINDER = "EventHubBinder";
     private static final String NAMESPACE = "Namespace";
 
+
     @Autowired(required = false)
-    private ResourceManagerProvider resourceManagerProvider;
+    private EventHubNamespaceManager eventHubNamespaceManager;
+
+    @Autowired(required=false)
+    private EventHubManager eventHubManager;
+
+    @Autowired(required=false)
+    private EventHubConsumerGroupManager eventHubConsumerGroupManager;
+
 
     @PostConstruct
     public void collectTelemetry() {
@@ -47,8 +58,8 @@ public void collectTelemetry() {
     @Bean
     @ConditionalOnMissingBean
     public EventHubChannelProvisioner eventHubChannelProvisioner(AzureEventHubProperties eventHubProperties) {
-        if (resourceManagerProvider != null) {
-            return new EventHubChannelResourceManagerProvisioner(resourceManagerProvider,
+        if (eventHubNamespaceManager != null && eventHubManager != null && eventHubConsumerGroupManager!=null) {
+            return new EventHubChannelResourceManagerProvisioner(eventHubNamespaceManager, eventHubManager, eventHubConsumerGroupManager,
                     eventHubProperties.getNamespace());
         } else {
             TelemetryCollector.getInstance().addProperty(EVENT_HUB_BINDER, NAMESPACE,

sdk/spring/azure-spring-cloud-eventhubs-stream-binder/src/main/java/com/microsoft/azure/eventhub/stream/binder/provisioning/EventHubChannelResourceManagerProvisioner.java

@@ -3,48 +3,51 @@
 
 package com.microsoft.azure.eventhub.stream.binder.provisioning;
 
-import com.microsoft.azure.management.eventhub.EventHub;
-import com.microsoft.azure.management.eventhub.EventHubNamespace;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
-import com.microsoft.azure.spring.cloud.context.core.util.Tuple;
 import org.springframework.cloud.stream.provisioning.ProvisioningException;
 import org.springframework.lang.NonNull;
 import org.springframework.util.Assert;
 
+import com.microsoft.azure.management.eventhub.EventHub;
+import com.microsoft.azure.management.eventhub.EventHubNamespace;
+import com.microsoft.azure.spring.cloud.context.core.impl.EventHubConsumerGroupManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.EventHubManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.EventHubNamespaceManager;
+import com.microsoft.azure.spring.cloud.context.core.util.Tuple;
+
 /**
  * @author Warren Zhu
  */
 public class EventHubChannelResourceManagerProvisioner extends EventHubChannelProvisioner {
     private final String namespace;
-    private final ResourceManagerProvider resourceManagerProvider;
+    private final EventHubNamespaceManager eventHubNamespaceManager;
+    private final EventHubManager eventHubManager;
+    private final EventHubConsumerGroupManager eventHubConsumerGroupManager;
 
-    public EventHubChannelResourceManagerProvisioner(@NonNull ResourceManagerProvider resourceManagerProvider,
-            @NonNull String namespace) {
+    public EventHubChannelResourceManagerProvisioner(@NonNull EventHubNamespaceManager eventHubNamespaceManager,
+            @NonNull EventHubManager eventHubManager,
+            @NonNull EventHubConsumerGroupManager eventHubConsumerGroupManager, @NonNull String namespace) {
         Assert.hasText(namespace, "The namespace can't be null or empty");
         this.namespace = namespace;
-        this.resourceManagerProvider = resourceManagerProvider;
+        this.eventHubNamespaceManager = eventHubNamespaceManager;
+        this.eventHubManager = eventHubManager;
+        this.eventHubConsumerGroupManager = eventHubConsumerGroupManager;
     }
 
     @Override
     protected void validateOrCreateForConsumer(String name, String group) {
-        EventHubNamespace eventHubNamespace =
-                this.resourceManagerProvider.getEventHubNamespaceManager().getOrCreate(namespace);
-        EventHub eventHub = this.resourceManagerProvider.getEventHubManager().get(Tuple.of(eventHubNamespace, name));
+        EventHubNamespace eventHubNamespace = eventHubNamespaceManager.getOrCreate(namespace);
+        EventHub eventHub = eventHubManager.get(Tuple.of(eventHubNamespace, name));
         if (eventHub == null) {
             throw new ProvisioningException(
                     String.format("Event hub with name '%s' in namespace '%s' not existed", name, namespace));
         }
 
-        this.resourceManagerProvider.getEventHubConsumerGroupManager().getOrCreate(Tuple.of(eventHub, group));
+        eventHubConsumerGroupManager.getOrCreate(Tuple.of(eventHub, group));
     }
 
     @Override
     protected void validateOrCreateForProducer(String name) {
-        if (resourceManagerProvider == null) {
-            return;
-        }
-        EventHubNamespace eventHubNamespace =
-                this.resourceManagerProvider.getEventHubNamespaceManager().getOrCreate(namespace);
-        this.resourceManagerProvider.getEventHubManager().getOrCreate(Tuple.of(eventHubNamespace, name));
+        EventHubNamespace eventHubNamespace = eventHubNamespaceManager.getOrCreate(namespace);
+        eventHubManager.getOrCreate(Tuple.of(eventHubNamespace, name));
     }
 }

sdk/spring/azure-spring-cloud-servicebus-queue-stream-binder/src/main/java/com/microsoft/azure/servicebus/stream/binder/config/ServiceBusQueueBinderConfiguration.java

@@ -3,6 +3,17 @@
 
 package com.microsoft.azure.servicebus.stream.binder.config;
 
+import javax.annotation.PostConstruct;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+
 import com.microsoft.azure.servicebus.stream.binder.ServiceBusQueueMessageChannelBinder;
 import com.microsoft.azure.servicebus.stream.binder.properties.ServiceBusQueueExtendedBindingProperties;
 import com.microsoft.azure.servicebus.stream.binder.provisioning.ServiceBusChannelProvisioner;
@@ -11,47 +22,41 @@
 import com.microsoft.azure.spring.cloud.autoconfigure.servicebus.AzureServiceBusProperties;
 import com.microsoft.azure.spring.cloud.autoconfigure.servicebus.AzureServiceBusQueueAutoConfiguration;
 import com.microsoft.azure.spring.cloud.autoconfigure.servicebus.ServiceBusUtils;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusNamespaceManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusQueueManager;
 import com.microsoft.azure.spring.cloud.telemetry.TelemetryCollector;
 import com.microsoft.azure.spring.integration.servicebus.queue.ServiceBusQueueOperation;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.cloud.stream.binder.Binder;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Import;
-
-import javax.annotation.PostConstruct;
 
 /**
  * @author Warren Zhu
  */
 @Configuration
 @ConditionalOnMissingBean(Binder.class)
-@Import({AzureServiceBusQueueAutoConfiguration.class, AzureEnvironmentAutoConfiguration.class})
-@EnableConfigurationProperties({AzureServiceBusProperties.class, ServiceBusQueueExtendedBindingProperties.class})
+@Import({ AzureServiceBusQueueAutoConfiguration.class, AzureEnvironmentAutoConfiguration.class })
+@EnableConfigurationProperties({ AzureServiceBusProperties.class, ServiceBusQueueExtendedBindingProperties.class })
 public class ServiceBusQueueBinderConfiguration {
 
     private static final String SERVICE_BUS_QUEUE_BINDER = "ServiceBusQueueBinder";
     private static final String NAMESPACE = "Namespace";
 
     @Autowired(required = false)
-    private ResourceManagerProvider resourceManagerProvider;
+    private ServiceBusNamespaceManager serviceBusNamespaceManager;
+
+    @Autowired(required = false)
+    private ServiceBusQueueManager serviceBusQueueManager;
 
     @PostConstruct
     public void collectTelemetry() {
         TelemetryCollector.getInstance().addService(SERVICE_BUS_QUEUE_BINDER);
     }
 
     @Bean
-    @ConditionalOnBean(ResourceManagerProvider.class)
+    @ConditionalOnBean({ ServiceBusNamespaceManager.class, ServiceBusQueueManager.class })
     @ConditionalOnMissingBean
     public ServiceBusChannelProvisioner serviceBusChannelProvisioner(AzureServiceBusProperties serviceBusProperties) {
-        if (this.resourceManagerProvider != null) {
-            return new ServiceBusQueueChannelResourceManagerProvisioner(resourceManagerProvider,
-                    serviceBusProperties.getNamespace());
+        if (this.serviceBusNamespaceManager != null && serviceBusQueueManager != null) {
+            return new ServiceBusQueueChannelResourceManagerProvisioner(serviceBusNamespaceManager,
+                    serviceBusQueueManager, serviceBusProperties.getNamespace());
         } else {
             TelemetryCollector.getInstance().addProperty(SERVICE_BUS_QUEUE_BINDER, NAMESPACE,
                     ServiceBusUtils.getNamespace(serviceBusProperties.getConnectionString()));
@@ -60,7 +65,7 @@ public ServiceBusChannelProvisioner serviceBusChannelProvisioner(AzureServiceBus
     }
 
     @Bean
-    @ConditionalOnMissingBean({ResourceManagerProvider.class, ServiceBusChannelProvisioner.class})
+    @ConditionalOnMissingBean(ServiceBusChannelProvisioner.class)
     public ServiceBusChannelProvisioner serviceBusChannelProvisionerWithResourceManagerProvider() {
         return new ServiceBusChannelProvisioner();
     }
@@ -69,8 +74,8 @@ public ServiceBusChannelProvisioner serviceBusChannelProvisionerWithResourceMana
     public ServiceBusQueueMessageChannelBinder serviceBusQueueBinder(
             ServiceBusChannelProvisioner queueChannelProvisioner, ServiceBusQueueOperation serviceBusQueueOperation,
             ServiceBusQueueExtendedBindingProperties bindingProperties) {
-        ServiceBusQueueMessageChannelBinder binder =
-                new ServiceBusQueueMessageChannelBinder(null, queueChannelProvisioner, serviceBusQueueOperation);
+        ServiceBusQueueMessageChannelBinder binder = new ServiceBusQueueMessageChannelBinder(null,
+                queueChannelProvisioner, serviceBusQueueOperation);
         binder.setBindingProperties(bindingProperties);
         return binder;
     }

sdk/spring/azure-spring-cloud-servicebus-queue-stream-binder/src/main/java/com/microsoft/azure/servicebus/stream/binder/provisioning/ServiceBusQueueChannelResourceManagerProvisioner.java

@@ -3,34 +3,38 @@
 
 package com.microsoft.azure.servicebus.stream.binder.provisioning;
 
-import com.microsoft.azure.management.servicebus.Queue;
-import com.microsoft.azure.management.servicebus.ServiceBusNamespace;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
-import com.microsoft.azure.spring.cloud.context.core.util.Tuple;
 import org.springframework.cloud.stream.provisioning.ProvisioningException;
 import org.springframework.lang.NonNull;
 import org.springframework.util.Assert;
 
+import com.microsoft.azure.management.servicebus.Queue;
+import com.microsoft.azure.management.servicebus.ServiceBusNamespace;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusNamespaceManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusQueueManager;
+import com.microsoft.azure.spring.cloud.context.core.util.Tuple;
+
 /**
  * @author Warren Zhu
  */
 public class ServiceBusQueueChannelResourceManagerProvisioner extends ServiceBusChannelProvisioner {
 
-    private final ResourceManagerProvider resourceManagerProvider;
     private final String namespace;
+    private final ServiceBusNamespaceManager serviceBusNamespaceManager;
+    private final ServiceBusQueueManager serviceBusQueueManager;
 
-    public ServiceBusQueueChannelResourceManagerProvisioner(@NonNull ResourceManagerProvider resourceManagerProvider,
-            @NonNull String namespace) {
+    public ServiceBusQueueChannelResourceManagerProvisioner(
+            @NonNull ServiceBusNamespaceManager serviceBusNamespaceManager,
+            @NonNull ServiceBusQueueManager serviceBusQueueManager, @NonNull String namespace) {
         Assert.hasText(namespace, "The namespace can't be null or empty");
-        this.resourceManagerProvider = resourceManagerProvider;
+        this.serviceBusNamespaceManager = serviceBusNamespaceManager;
+        this.serviceBusQueueManager = serviceBusQueueManager;
         this.namespace = namespace;
     }
 
     @Override
     protected void validateOrCreateForConsumer(String name, String group) {
-        ServiceBusNamespace namespace =
-                this.resourceManagerProvider.getServiceBusNamespaceManager().getOrCreate(this.namespace);
-        Queue queue = this.resourceManagerProvider.getServiceBusQueueManager().getOrCreate(Tuple.of(namespace, name));
+        ServiceBusNamespace namespace = serviceBusNamespaceManager.getOrCreate(this.namespace);
+        Queue queue = serviceBusQueueManager.getOrCreate(Tuple.of(namespace, name));
         if (queue == null) {
             throw new ProvisioningException(
                     String.format("Event hub with name '%s' in namespace '%s' not existed", name, namespace));
@@ -39,8 +43,7 @@ protected void validateOrCreateForConsumer(String name, String group) {
 
     @Override
     protected void validateOrCreateForProducer(String name) {
-        ServiceBusNamespace namespace =
-                this.resourceManagerProvider.getServiceBusNamespaceManager().getOrCreate(this.namespace);
-        this.resourceManagerProvider.getServiceBusQueueManager().getOrCreate(Tuple.of(namespace, name));
+        ServiceBusNamespace namespace = serviceBusNamespaceManager.getOrCreate(this.namespace);
+        serviceBusQueueManager.getOrCreate(Tuple.of(namespace, name));
     }
 }

sdk/spring/azure-spring-cloud-servicebus-topic-stream-binder/src/main/java/com/microsoft/azure/servicebus/stream/binder/config/ServiceBusTopicBinderConfiguration.java

@@ -3,6 +3,17 @@
 
 package com.microsoft.azure.servicebus.stream.binder.config;
 
+import javax.annotation.PostConstruct;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+
 import com.microsoft.azure.servicebus.stream.binder.ServiceBusTopicMessageChannelBinder;
 import com.microsoft.azure.servicebus.stream.binder.properties.ServiceBusTopicExtendedBindingProperties;
 import com.microsoft.azure.servicebus.stream.binder.provisioning.ServiceBusChannelProvisioner;
@@ -11,47 +22,47 @@
 import com.microsoft.azure.spring.cloud.autoconfigure.servicebus.AzureServiceBusProperties;
 import com.microsoft.azure.spring.cloud.autoconfigure.servicebus.AzureServiceBusTopicAutoConfiguration;
 import com.microsoft.azure.spring.cloud.autoconfigure.servicebus.ServiceBusUtils;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusNamespaceManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusTopicManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusTopicSubscriptionManager;
 import com.microsoft.azure.spring.cloud.telemetry.TelemetryCollector;
 import com.microsoft.azure.spring.integration.servicebus.topic.ServiceBusTopicOperation;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.cloud.stream.binder.Binder;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Import;
-
-import javax.annotation.PostConstruct;
 
 /**
  * @author Warren Zhu
  */
 @Configuration
 @ConditionalOnMissingBean(Binder.class)
-@Import({AzureServiceBusTopicAutoConfiguration.class, AzureEnvironmentAutoConfiguration.class})
-@EnableConfigurationProperties({AzureServiceBusProperties.class, ServiceBusTopicExtendedBindingProperties.class})
+@Import({ AzureServiceBusTopicAutoConfiguration.class, AzureEnvironmentAutoConfiguration.class })
+@EnableConfigurationProperties({ AzureServiceBusProperties.class, ServiceBusTopicExtendedBindingProperties.class })
 public class ServiceBusTopicBinderConfiguration {
 
     private static final String SERVICE_BUS_TOPIC_BINDER = "ServiceBusTopicBinder";
     private static final String NAMESPACE = "Namespace";
 
     @Autowired(required = false)
-    private ResourceManagerProvider resourceManagerProvider;
+    private ServiceBusNamespaceManager serviceBusNamespaceManager;
+
+    @Autowired(required = false)
+    private ServiceBusTopicManager serviceBusTopicManager;
+
+    @Autowired(required = false)
+    private ServiceBusTopicSubscriptionManager serviceBusTopicSubscriptionManager;
 
     @PostConstruct
     public void collectTelemetry() {
         TelemetryCollector.getInstance().addService(SERVICE_BUS_TOPIC_BINDER);
     }
 
     @Bean
-    @ConditionalOnBean(ResourceManagerProvider.class)
+    @ConditionalOnBean({ ServiceBusNamespaceManager.class, ServiceBusTopicManager.class,
+            ServiceBusTopicSubscriptionManager.class })
     @ConditionalOnMissingBean
     public ServiceBusChannelProvisioner serviceBusChannelProvisioner(AzureServiceBusProperties serviceBusProperties) {
-        if (this.resourceManagerProvider != null) {
-            return new ServiceBusTopicChannelResourceManagerProvisioner(resourceManagerProvider,
-                    serviceBusProperties.getNamespace());
+        if (this.serviceBusNamespaceManager != null && this.serviceBusTopicManager != null
+                && this.serviceBusTopicSubscriptionManager != null) {
+            return new ServiceBusTopicChannelResourceManagerProvisioner(serviceBusNamespaceManager,
+                    serviceBusTopicManager, serviceBusTopicSubscriptionManager, serviceBusProperties.getNamespace());
         } else {
             TelemetryCollector.getInstance().addProperty(SERVICE_BUS_TOPIC_BINDER, NAMESPACE,
                     ServiceBusUtils.getNamespace(serviceBusProperties.getConnectionString()));
@@ -60,7 +71,8 @@ public ServiceBusChannelProvisioner serviceBusChannelProvisioner(AzureServiceBus
     }
 
     @Bean
-    @ConditionalOnMissingBean({ResourceManagerProvider.class, ServiceBusChannelProvisioner.class})
+    @ConditionalOnMissingBean({ ServiceBusNamespaceManager.class, ServiceBusTopicManager.class,
+            ServiceBusTopicSubscriptionManager.class, ServiceBusChannelProvisioner.class })
     public ServiceBusChannelProvisioner serviceBusChannelProvisionerWithResourceManagerProvider() {
         return new ServiceBusChannelProvisioner();
     }
@@ -69,8 +81,8 @@ public ServiceBusChannelProvisioner serviceBusChannelProvisionerWithResourceMana
     public ServiceBusTopicMessageChannelBinder serviceBusTopicBinder(
             ServiceBusChannelProvisioner topicChannelProvisioner, ServiceBusTopicOperation serviceBusTopicOperation,
             ServiceBusTopicExtendedBindingProperties bindingProperties) {
-        ServiceBusTopicMessageChannelBinder binder =
-                new ServiceBusTopicMessageChannelBinder(null, topicChannelProvisioner, serviceBusTopicOperation);
+        ServiceBusTopicMessageChannelBinder binder = new ServiceBusTopicMessageChannelBinder(null,
+                topicChannelProvisioner, serviceBusTopicOperation);
         binder.setBindingProperties(bindingProperties);
         return binder;
     }

sdk/spring/azure-spring-cloud-servicebus-topic-stream-binder/src/main/java/com/microsoft/azure/servicebus/stream/binder/provisioning/ServiceBusTopicChannelResourceManagerProvisioner.java

@@ -3,46 +3,53 @@
 
 package com.microsoft.azure.servicebus.stream.binder.provisioning;
 
-import com.microsoft.azure.management.servicebus.ServiceBusNamespace;
-import com.microsoft.azure.management.servicebus.Topic;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
-import com.microsoft.azure.spring.cloud.context.core.util.Tuple;
 import org.springframework.cloud.stream.provisioning.ProvisioningException;
 import org.springframework.lang.NonNull;
 import org.springframework.util.Assert;
 
+import com.microsoft.azure.management.servicebus.ServiceBusNamespace;
+import com.microsoft.azure.management.servicebus.Topic;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusNamespaceManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusTopicManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusTopicSubscriptionManager;
+import com.microsoft.azure.spring.cloud.context.core.util.Tuple;
+
 /**
  * @author Warren Zhu
  */
 public class ServiceBusTopicChannelResourceManagerProvisioner extends ServiceBusChannelProvisioner {
 
-    private final ResourceManagerProvider resourceManagerProvider;
+    private final ServiceBusNamespaceManager serviceBusNamespaceManager;
+    private final ServiceBusTopicManager serviceBusTopicManager;
+    private final ServiceBusTopicSubscriptionManager serviceBusTopicSubscriptionManager;
     private final String namespace;
 
-    public ServiceBusTopicChannelResourceManagerProvisioner(@NonNull ResourceManagerProvider resourceManagerProvider,
-            @NonNull String namespace) {
+    public ServiceBusTopicChannelResourceManagerProvisioner(
+            @NonNull ServiceBusNamespaceManager serviceBusNamespaceManager,
+            @NonNull ServiceBusTopicManager serviceBusTopicManager,
+            @NonNull ServiceBusTopicSubscriptionManager serviceBusTopicSubscriptionManager, @NonNull String namespace) {
         Assert.hasText(namespace, "The namespace can't be null or empty");
-        this.resourceManagerProvider = resourceManagerProvider;
+        this.serviceBusNamespaceManager = serviceBusNamespaceManager;
+        this.serviceBusTopicManager = serviceBusTopicManager;
+        this.serviceBusTopicSubscriptionManager = serviceBusTopicSubscriptionManager;
         this.namespace = namespace;
     }
 
     @Override
     protected void validateOrCreateForConsumer(String name, String group) {
-        ServiceBusNamespace namespace =
-                this.resourceManagerProvider.getServiceBusNamespaceManager().getOrCreate(this.namespace);
-        Topic topic = this.resourceManagerProvider.getServiceBusTopicManager().getOrCreate(Tuple.of(namespace, name));
+        ServiceBusNamespace namespace = serviceBusNamespaceManager.getOrCreate(this.namespace);
+        Topic topic = serviceBusTopicManager.getOrCreate(Tuple.of(namespace, name));
         if (topic == null) {
             throw new ProvisioningException(
                     String.format("Event hub with name '%s' in namespace '%s' not existed", name, namespace));
         }
 
-        this.resourceManagerProvider.getServiceBusTopicSubscriptionManager().getOrCreate(Tuple.of(topic, group));
+        this.serviceBusTopicSubscriptionManager.getOrCreate(Tuple.of(topic, group));
     }
 
     @Override
     protected void validateOrCreateForProducer(String name) {
-        ServiceBusNamespace namespace =
-                this.resourceManagerProvider.getServiceBusNamespaceManager().getOrCreate(this.namespace);
-        this.resourceManagerProvider.getServiceBusTopicManager().getOrCreate(Tuple.of(namespace, name));
+        ServiceBusNamespace namespace = serviceBusNamespaceManager.getOrCreate(this.namespace);
+        serviceBusTopicManager.getOrCreate(Tuple.of(namespace, name));
     }
 }

sdk/spring/azure-spring-cloud-storage/src/main/java/com/microsoft/azure/spring/cloud/autoconfigure/storage/AzureStorageAutoConfiguration.java

@@ -29,18 +29,15 @@
 import com.azure.core.http.HttpPipeline;
 import com.azure.core.http.policy.HttpLogOptions;
 import com.azure.core.util.logging.ClientLogger;
-import com.azure.resourcemanager.storage.StorageManagementClient;
-import com.azure.resourcemanager.storage.StorageManagementClientBuilder;
-import com.azure.resourcemanager.storage.fluent.StorageAccountsClient;
+import com.azure.resourcemanager.storage.models.StorageAccount;
 import com.azure.storage.blob.BlobServiceClientBuilder;
 import com.azure.storage.blob.implementation.util.BuilderHelper;
 import com.azure.storage.common.policy.RequestRetryOptions;
 import com.azure.storage.file.share.ShareServiceClientBuilder;
 import com.microsoft.azure.identity.spring.SpringEnvironmentTokenBuilder;
-import com.microsoft.azure.management.storage.StorageAccount;
 import com.microsoft.azure.spring.cloud.autoconfigure.context.AzureContextAutoConfiguration;
 import com.microsoft.azure.spring.cloud.context.core.api.EnvironmentProvider;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
+import com.microsoft.azure.spring.cloud.context.core.impl.StorageAccountManager;
 import com.microsoft.azure.spring.cloud.context.core.storage.StorageConnectionStringProvider;
 import com.microsoft.azure.spring.cloud.context.core.storage.StorageEndpointStringBuilder;
 import com.microsoft.azure.spring.cloud.storage.AzureStorageProtocolResolver;
@@ -62,7 +59,7 @@ public class AzureStorageAutoConfiguration {
     private static final String ACCOUNT_NAME = "AccountName";
 
     @Autowired(required = false)
-    private ResourceManagerProvider resourceManagerProvider;
+    private StorageAccountManager storageAccountManager;
 
     @PostConstruct
     public void collectTelemetry() {
@@ -82,9 +79,8 @@ public BlobServiceClientBuilder blobServiceClientBuilder(AzureStorageProperties
         // Use storage credentials where provided, default identity otherwise.
         if (StringUtils.isNotBlank(storageProperties.getAccessKey())) {
             String connectionString = null;
-            if (resourceManagerProvider != null) {
-                StorageAccount storageAccount = resourceManagerProvider.getStorageAccountManager()
-                        .getOrCreate(storageProperties.getAccount());
+            if (storageAccountManager != null) {
+                StorageAccount storageAccount = storageAccountManager.getOrCreate(storageProperties.getAccount());
                 connectionString = StorageConnectionStringProvider.getConnectionString(storageAccount,
                         environmentProvider.getEnvironment(), storageProperties.isSecureTransfer());
             } else {
@@ -116,11 +112,10 @@ public ShareServiceClientBuilder shareServiceClientBuilder(AzureStoragePropertie
         // Use storage credentials where provided, default identity otherwise.
         if (StringUtils.isNotBlank(storageProperties.getAccessKey())) {
             String connectionString;
-            if (resourceManagerProvider != null) {
+            if (storageAccountManager != null) {
                 String accountName = storageProperties.getAccount();
 
-                StorageAccount storageAccount = resourceManagerProvider.getStorageAccountManager()
-                        .getOrCreate(accountName);
+                StorageAccount storageAccount = storageAccountManager.getOrCreate(accountName);
                 connectionString = StorageConnectionStringProvider.getConnectionString(storageAccount,
                         environmentProvider.getEnvironment(), storageProperties.isSecureTransfer());
             } else {
@@ -143,7 +138,7 @@ public ShareServiceClientBuilder shareServiceClientBuilder(AzureStoragePropertie
                     new ClientLogger(this.getClass()));
 
             authenticatedClientBuilder = new ShareServiceClientBuilder().pipeline(pipeline);
-            
+
         }
 
         return authenticatedClientBuilder

sdk/spring/azure-spring-cloud-storage/src/main/java/com/microsoft/azure/spring/cloud/autoconfigure/storage/AzureStorageQueueAutoConfiguration.java

@@ -3,32 +3,33 @@
 
 package com.microsoft.azure.spring.cloud.autoconfigure.storage;
 
+import javax.annotation.PostConstruct;
+
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import com.azure.resourcemanager.storage.models.StorageAccount;
 import com.azure.storage.queue.QueueServiceClient;
-import com.microsoft.azure.management.storage.StorageAccount;
 import com.microsoft.azure.spring.cloud.autoconfigure.context.AzureContextAutoConfiguration;
 import com.microsoft.azure.spring.cloud.autoconfigure.context.AzureEnvironmentAutoConfiguration;
 import com.microsoft.azure.spring.cloud.context.core.api.EnvironmentProvider;
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
+import com.microsoft.azure.spring.cloud.context.core.impl.StorageAccountManager;
 import com.microsoft.azure.spring.cloud.context.core.storage.StorageConnectionStringProvider;
 import com.microsoft.azure.spring.cloud.telemetry.TelemetryCollector;
 import com.microsoft.azure.spring.integration.storage.queue.StorageQueueOperation;
 import com.microsoft.azure.spring.integration.storage.queue.StorageQueueTemplate;
 import com.microsoft.azure.spring.integration.storage.queue.factory.DefaultStorageQueueClientFactory;
 import com.microsoft.azure.spring.integration.storage.queue.factory.StorageQueueClientFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.AutoConfigureAfter;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-import javax.annotation.PostConstruct;
 
 @Configuration
-@AutoConfigureAfter({AzureContextAutoConfiguration.class, AzureEnvironmentAutoConfiguration.class})
-@ConditionalOnClass({QueueServiceClient.class, StorageQueueClientFactory.class})
+@AutoConfigureAfter({ AzureContextAutoConfiguration.class, AzureEnvironmentAutoConfiguration.class,
+        AzureStorageAutoConfiguration.class })
+@ConditionalOnClass({ QueueServiceClient.class, StorageQueueClientFactory.class })
 @ConditionalOnProperty(name = "spring.cloud.azure.storage.account")
 @EnableConfigurationProperties(AzureStorageProperties.class)
 public class AzureStorageQueueAutoConfiguration {
@@ -37,9 +38,6 @@ public class AzureStorageQueueAutoConfiguration {
     private static final String STORAGE = "Storage";
     private static final String ACCOUNT_NAME = "AccountName";
 
-    @Autowired(required = false)
-    private ResourceManagerProvider resourceManagerProvider;
-
     @PostConstruct
     public void collectTelemetry() {
         TelemetryCollector.getInstance().addService(STORAGE_QUEUE);
@@ -48,23 +46,16 @@ public void collectTelemetry() {
     @Bean
     @ConditionalOnMissingBean
     StorageQueueClientFactory storageQueueClientFactory(AzureStorageProperties storageProperties,
-                                                        EnvironmentProvider environmentProvider) {
+            StorageAccountManager storageAccountManager, EnvironmentProvider environmentProvider) {
 
         String connectionString;
 
-        if (resourceManagerProvider != null) {
-            String accountName = storageProperties.getAccount();
+        String accountName = storageProperties.getAccount();
 
-            StorageAccount storageAccount = resourceManagerProvider.getStorageAccountManager().getOrCreate(accountName);
+        StorageAccount storageAccount = storageAccountManager.getOrCreate(accountName);
 
-            connectionString = StorageConnectionStringProvider
-                .getConnectionString(storageAccount, environmentProvider.getEnvironment(),
-                    storageProperties.isSecureTransfer());
-        } else {
-            connectionString = StorageConnectionStringProvider
-                .getConnectionString(storageProperties.getAccount(), storageProperties.getAccessKey(),
-                    environmentProvider.getEnvironment());
-        }
+        connectionString = StorageConnectionStringProvider.getConnectionString(storageAccount,
+                environmentProvider.getEnvironment(), storageProperties.isSecureTransfer());
 
         return new DefaultStorageQueueClientFactory(connectionString);
     }

sdk/spring/azure-spring-integration-servicebus/src/main/java/com/microsoft/azure/spring/integration/servicebus/factory/AbstractServiceBusSenderFactory.java

@@ -3,7 +3,10 @@
 
 package com.microsoft.azure.spring.integration.servicebus.factory;
 
-import com.microsoft.azure.spring.cloud.context.core.api.ResourceManagerProvider;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusNamespaceManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusQueueManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusTopicManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusTopicSubscriptionManager;
 
 /**
  * Base class of service bus client factory to provide connection string
@@ -12,15 +15,31 @@
  */
 abstract class AbstractServiceBusSenderFactory implements ServiceBusSenderFactory {
     protected final String connectionString;
+    protected ServiceBusNamespaceManager serviceBusNamespaceManager;
+    protected ServiceBusQueueManager serviceBusQueueManager;
+    protected ServiceBusTopicManager serviceBusTopicManager;
+    protected ServiceBusTopicSubscriptionManager serviceBusTopicSubscriptionManager;
     protected String namespace;
-    protected ResourceManagerProvider resourceManagerProvider;
 
     AbstractServiceBusSenderFactory(String connectionString) {
         this.connectionString = connectionString;
     }
 
-    public void setResourceManagerProvider(ResourceManagerProvider resourceManagerProvider) {
-        this.resourceManagerProvider = resourceManagerProvider;
+    public void setServiceBusNamespaceManager(ServiceBusNamespaceManager serviceBusNamespaceManager) {
+        this.serviceBusNamespaceManager = serviceBusNamespaceManager;
+    }
+
+    public void setServiceBusQueueManager(ServiceBusQueueManager serviceBusQueueManager) {
+        this.serviceBusQueueManager = serviceBusQueueManager;
+    }
+
+    public void setServiceBusTopicManager(ServiceBusTopicManager serviceBusTopicManager) {
+        this.serviceBusTopicManager = serviceBusTopicManager;
+    }
+
+    public void setServiceBusTopicSubscriptionManager(
+            ServiceBusTopicSubscriptionManager serviceBusTopicSubscriptionManager) {
+        this.serviceBusTopicSubscriptionManager = serviceBusTopicSubscriptionManager;
     }
 
     public void setNamespace(String namespace) {

sdk/spring/azure-spring-integration-servicebus/src/main/java/com/microsoft/azure/spring/integration/servicebus/factory/DefaultServiceBusQueueClientFactory.java

@@ -33,10 +33,10 @@ public DefaultServiceBusQueueClientFactory(String connectionString) {
     }
 
     private IQueueClient createQueueClient(String destination) {
-        if (resourceManagerProvider != null && StringUtils.hasText(namespace)) {
+        if (StringUtils.hasText(namespace)) {
             ServiceBusNamespace serviceBusNamespace =
-                resourceManagerProvider.getServiceBusNamespaceManager().getOrCreate(namespace);
-            resourceManagerProvider.getServiceBusQueueManager().getOrCreate(Tuple.of(serviceBusNamespace, destination));
+                serviceBusNamespaceManager.getOrCreate(namespace);
+            serviceBusQueueManager.getOrCreate(Tuple.of(serviceBusNamespace, destination));
         }
 
         try {

sdk/spring/azure-spring-integration-servicebus/src/main/java/com/microsoft/azure/spring/integration/servicebus/factory/DefaultServiceBusTopicClientFactory.java

@@ -12,6 +12,8 @@
 import com.microsoft.azure.servicebus.TopicClient;
 import com.microsoft.azure.servicebus.primitives.ConnectionStringBuilder;
 import com.microsoft.azure.servicebus.primitives.ServiceBusException;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusNamespaceManager;
+import com.microsoft.azure.spring.cloud.context.core.impl.ServiceBusTopicSubscriptionManager;
 import com.microsoft.azure.spring.cloud.context.core.util.Memoizer;
 import com.microsoft.azure.spring.cloud.context.core.util.Tuple;
 import com.microsoft.azure.spring.integration.servicebus.ServiceBusRuntimeException;
@@ -21,46 +23,44 @@
 import java.util.function.Function;
 
 /**
- * Default implementation of {@link ServiceBusTopicClientFactory}.
- * Client will be cached to improve performance
+ * Default implementation of {@link ServiceBusTopicClientFactory}. Client will
+ * be cached to improve performance
  *
  * @author Warren Zhu
  */
 public class DefaultServiceBusTopicClientFactory extends AbstractServiceBusSenderFactory
-    implements ServiceBusTopicClientFactory {
+        implements ServiceBusTopicClientFactory {
     private static final String SUBSCRIPTION_PATH = "%s/subscriptions/%s";
-    private final BiFunction<String, String, ISubscriptionClient> subscriptionClientCreator =
-        Memoizer.memoize(this::createSubscriptionClient);
+    private final BiFunction<String, String, ISubscriptionClient> subscriptionClientCreator = Memoizer
+            .memoize(this::createSubscriptionClient);
     private final Function<String, ? extends IMessageSender> sendCreator = Memoizer.memoize(this::createTopicClient);
 
+
     public DefaultServiceBusTopicClientFactory(String connectionString) {
         super(connectionString);
     }
 
     private ISubscriptionClient createSubscriptionClient(String topicName, String subscription) {
 
-        if (resourceManagerProvider != null && StringUtils.hasText(namespace)) {
-            ServiceBusNamespace serviceBusNamespace =
-                resourceManagerProvider.getServiceBusNamespaceManager().getOrCreate(namespace);
-            Topic topic = resourceManagerProvider.getServiceBusTopicManager()
-                .getOrCreate(Tuple.of(serviceBusNamespace, topicName));
-            resourceManagerProvider.getServiceBusTopicSubscriptionManager().getOrCreate(Tuple.of(topic, subscription));
+        if (StringUtils.hasText(namespace)) {
+            ServiceBusNamespace serviceBusNamespace = serviceBusNamespaceManager.getOrCreate(namespace);
+            Topic topic = serviceBusTopicManager.getOrCreate(Tuple.of(serviceBusNamespace, topicName));
+            serviceBusTopicSubscriptionManager.getOrCreate(Tuple.of(topic, subscription));
         }
 
         String subscriptionPath = String.format(SUBSCRIPTION_PATH, topicName, subscription);
         try {
             return new SubscriptionClient(new ConnectionStringBuilder(connectionString, subscriptionPath),
-                ReceiveMode.PEEKLOCK);
+                    ReceiveMode.PEEKLOCK);
         } catch (InterruptedException | ServiceBusException e) {
             throw new ServiceBusRuntimeException("Failed to create service bus subscription client", e);
         }
     }
 
     private IMessageSender createTopicClient(String topicName) {
-        if (resourceManagerProvider != null && StringUtils.hasText(namespace)) {
-            ServiceBusNamespace serviceBusNamespace =
-                resourceManagerProvider.getServiceBusNamespaceManager().getOrCreate(namespace);
-            resourceManagerProvider.getServiceBusTopicManager().getOrCreate(Tuple.of(serviceBusNamespace, topicName));
+        if (serviceBusNamespaceManager != null && serviceBusTopicManager != null && StringUtils.hasText(namespace)) {
+            ServiceBusNamespace serviceBusNamespace = serviceBusNamespaceManager.getOrCreate(namespace);
+            serviceBusTopicManager.getOrCreate(Tuple.of(serviceBusNamespace, topicName));
         }
 
         try {