Make storage account optional in the eventhub binder configuration. (A…

…zure#23640)

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

@@ -55,6 +55,7 @@ public EventHubNamespaceManager eventHubNamespaceManager(AzureResourceManager az
     @Bean
     @ConditionalOnMissingBean
     @ConditionalOnBean(AzureResourceManager.class)
+    @ConditionalOnProperty(value = "spring.cloud.azure.eventhub.checkpoint-storage-account")
     public StorageAccountManager storageAccountManager(AzureResourceManager azureResourceManager,
                                                        AzureProperties azureProperties) {
         return new StorageAccountManager(azureResourceManager, azureProperties);
@@ -118,7 +119,6 @@ public EventHubOperation eventHubOperation(EventHubClientFactory clientFactory)
         return new EventHubTemplate(clientFactory);
     }
 
-
     private String getStorageConnectionString(AzureEventHubProperties properties,
                                               StorageAccountManager storageAccountManager,
                                               AzureEnvironment azureEnvironment) {
@@ -127,6 +127,10 @@ private String getStorageConnectionString(AzureEventHubProperties properties,
         final String accountKey = properties.getCheckpointAccessKey();
         final StorageConnectionStringProvider provider;
 
+        if (accountName == null) {
+            return null;
+        }
+
         if (storageAccountManager != null) {
             provider = new StorageConnectionStringProvider(storageAccountManager.getOrCreate(accountName));
         } else {

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

@@ -28,10 +28,10 @@
 
 import static java.util.Collections.singletonList;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
-import static org.junit.jupiter.api.Assertions.assertThrows;
 
 public class AzureEventHubAutoConfigurationTest {
 
@@ -105,6 +105,40 @@ public void testResourceManagerProvided() {
                           });
     }
 
+    @Test
+    public void testEventHubOperationProvidedNotStorageUnderSP() {
+        this.contextRunner.withUserConfiguration(
+                TestConfigWithAzureResourceManagerAndConnectionProvider.class,
+                AzureEventHubAutoConfiguration.class)
+                          .withPropertyValues(
+                              AZURE_PROPERTY_PREFIX + "resource-group=rg1",
+                              EVENT_HUB_PROPERTY_PREFIX + "namespace=ns1"
+                          )
+                          .run(context -> {
+                              assertThat(context).hasSingleBean(EventHubNamespaceManager.class);
+                              assertThat(context).hasSingleBean(EventHubOperation.class);
+                              assertThat(context).doesNotHaveBean(StorageAccountManager.class);
+                          });
+    }
+
+    @Test
+    public void testEventHubOperationProvidedNotStorageUnderMSI() {
+        this.contextRunner.withUserConfiguration(
+                TestConfigWithAzureResourceManagerAndConnectionProvider.class,
+                AzureEventHubAutoConfiguration.class)
+                          .withPropertyValues(
+                              AZURE_PROPERTY_PREFIX + "resource-group=rg1",
+                              AZURE_PROPERTY_PREFIX + "msi-enabled=true",
+                              EVENT_HUB_PROPERTY_PREFIX + "namespace=ns1",
+                              AZURE_PROPERTY_PREFIX + "subscription-id=sub"
+                          )
+                          .run(context -> {
+                              assertThat(context).hasSingleBean(EventHubNamespaceManager.class);
+                              assertThat(context).hasSingleBean(EventHubOperation.class);
+                              assertThat(context).doesNotHaveBean(StorageAccountManager.class);
+                          });
+    }
+
     @Configuration
     @EnableConfigurationProperties(AzureProperties.class)
     public static class TestConfigWithAzureResourceManagerAndConnectionProvider {

sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/java/com/azure/spring/eventhub/stream/binder/EventHubBinderConfigurationTest.java

@@ -0,0 +1,67 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.spring.eventhub.stream.binder;
+
+import com.azure.spring.cloud.autoconfigure.context.AzureContextAutoConfiguration;
+import com.azure.spring.cloud.autoconfigure.context.AzureEnvironmentAutoConfiguration;
+import com.azure.spring.cloud.autoconfigure.eventhub.AzureEventHubAutoConfiguration;
+import com.azure.spring.cloud.context.core.impl.EventHubNamespaceManager;
+import com.azure.spring.cloud.context.core.impl.StorageAccountManager;
+import com.azure.spring.eventhub.stream.binder.config.EventHubBinderConfiguration;
+import com.azure.spring.integration.eventhub.api.EventHubClientFactory;
+import com.azure.spring.integration.eventhub.api.EventHubOperation;
+import com.azure.spring.integration.eventhub.factory.EventHubConnectionStringProvider;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class EventHubBinderConfigurationTest {
+
+    private static final String EVENT_HUB_PROPERTY_PREFIX = "spring.cloud.azure.eventhub.";
+    private static final String AZURE_PROPERTY_PREFIX = "spring.cloud.azure.";
+
+    private String connectionString = "connection-string=Endpoint=sb://eventhub-test-1\"\n"
+        + "                + \".servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;\"\n"
+        + "                + \"SharedAccessKey=ByyyxxxUw=";
+
+    private ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+        .withPropertyValues(AZURE_PROPERTY_PREFIX + "stream.function.definition=supply")
+        .withPropertyValues(AZURE_PROPERTY_PREFIX + "stream.bindings.supply-out-0.destination=eventhub1")
+        .withConfiguration(AutoConfigurations.of(AzureEnvironmentAutoConfiguration.class))
+        .withConfiguration(AutoConfigurations.of(AzureContextAutoConfiguration.class))
+        .withConfiguration(AutoConfigurations.of(AzureEventHubAutoConfiguration.class))
+        .withConfiguration(AutoConfigurations.of(EventHubBinderConfiguration.class));
+
+    @Test
+    public void testStorageNotConfiguredToGetClientFactoryBeanOnConnectionString() {
+        contextRunner
+            .withPropertyValues(EVENT_HUB_PROPERTY_PREFIX + connectionString)
+            .run(context -> {
+                assertThat(context).hasSingleBean(EventHubClientFactory.class);
+                assertThat(context).hasSingleBean(EventHubOperation.class);
+                assertThat(context).doesNotHaveBean(StorageAccountManager.class);
+            });
+    }
+
+    @Test
+    public void testStorageNotConfiguredToGetClientFactoryBeanOnMSI() {
+        contextRunner
+            .withPropertyValues(
+                AZURE_PROPERTY_PREFIX + "msi-enabled=true",
+                AZURE_PROPERTY_PREFIX + "client-id=fake-client-id",
+                AZURE_PROPERTY_PREFIX + "resource-group=fake-res-group",
+                AZURE_PROPERTY_PREFIX + "subscription-id=fake-sub"
+            )
+            .withBean(EventHubConnectionStringProvider.class, connectionString)
+            .run(context -> {
+                assertThat(context).hasSingleBean(EventHubClientFactory.class);
+                assertThat(context).hasSingleBean(EventHubNamespaceManager.class);
+                assertThat(context).hasSingleBean(EventHubOperation.class);
+                assertThat(context).doesNotHaveBean(StorageAccountManager.class);
+            });
+    }
+
+}

sdk/spring/azure-spring-integration-eventhubs/src/main/java/com/azure/spring/integration/eventhub/factory/DefaultEventHubClientFactory.java

@@ -62,7 +62,6 @@ public class DefaultEventHubClientFactory implements EventHubClientFactory, Disp
     public DefaultEventHubClientFactory(@NonNull String eventHubConnectionString,
                                         String checkpointConnectionString,
                                         String checkpointStorageContainer) {
-        Assert.hasText(checkpointConnectionString, "checkpointConnectionString can't be null or empty");
         this.eventHubConnectionString = eventHubConnectionString;
         this.checkpointStorageConnectionString = checkpointConnectionString;
         this.checkpointStorageContainer = checkpointStorageContainer;
@@ -85,7 +84,8 @@ private EventHubProducerAsyncClient createProducerClient(String eventHubName) {
 
     private EventProcessorClient createEventProcessorClientInternal(String eventHubName, String consumerGroup,
                                                                     EventHubProcessor eventHubProcessor) {
-
+        Assert.hasText(checkpointStorageConnectionString, "checkpointConnectionString can't be null or empty, check "
+            + "whether checkpoint-storage-account is configured in the configuration file.");
         // We set eventHubName as the container name when we use track1 library, and the EventHubProcessor will create
         // the container automatically if not exists
         String containerName = checkpointStorageContainer == null ? eventHubName : checkpointStorageContainer;