Fixes NullPointerException when using connectionString and build client. (Azure#32086)

conniey · web-flow · commit 496458db17bf · 2023-01-03T23:55:15.000Z
* Do not validate eventHubName until client is built.

* Add tests.

* Update changelog.
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md b/sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md
@@ -10,6 +10,7 @@
 
 ### Bugs Fixed
 
+- Fixed ability to pass in namespace connection string in EventHubClientBuilder. ([#29536](https://github.com/Azure/azure-sdk-for-java/issues/29536))
 - Added retry for createBatch API as this API also makes network calls similar to its companion send API.
 
 ### Other Changes
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubClientBuilder.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubClientBuilder.java
@@ -53,6 +53,7 @@
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
 import java.util.regex.Pattern;
 
 import static com.azure.messaging.eventhubs.implementation.ClientConstants.CONNECTION_ID_KEY;
@@ -221,7 +222,7 @@ public EventHubClientBuilder() {
     }
 
     /**
-     * Sets the credential information given a connection string to the Event Hub instance.
+     * Sets the credential information given a connection string to the Event Hub instance or the Event Hubs namespace.
      *
      * <p>
      * If the connection string is copied from the Event Hubs namespace, it will likely not contain the name to the
@@ -234,20 +235,34 @@ public EventHubClientBuilder() {
      * from that Event Hub will result in a connection string that contains the name.
      * </p>
      *
-     * @param connectionString The connection string to use for connecting to the Event Hub instance. It is expected
-     *     that the Event Hub name and the shared access key properties are contained in this connection string.
+     * @param connectionString The connection string to use for connecting to the Event Hub instance or Event Hubs
+     *     instance. It is expected that the Event Hub name and the shared access key properties are contained in this
+     *     connection string.
      *
      * @return The updated {@link EventHubClientBuilder} object.
-     * @throws IllegalArgumentException if {@code connectionString} is null or empty. Or, the {@code
-     *     connectionString} does not contain the "EntityPath" key, which is the name of the Event Hub instance.
+     * @throws IllegalArgumentException if {@code connectionString} is null or empty. If {@code fullyQualifiedNamespace}
+     *     in the connection string is null.
+     * @throws NullPointerException if a credential could not be extracted
      * @throws AzureException If the shared access signature token credential could not be created using the
      *     connection string.
      */
     @Override
     public EventHubClientBuilder connectionString(String connectionString) {
-        ConnectionStringProperties properties = new ConnectionStringProperties(connectionString);
-        TokenCredential tokenCredential = getTokenCredential(properties);
-        return credential(properties.getEndpoint().getHost(), properties.getEntityPath(), tokenCredential);
+        final ConnectionStringProperties properties = new ConnectionStringProperties(connectionString);
+
+        this.fullyQualifiedNamespace = Objects.requireNonNull(properties.getEndpoint().getHost(),
+            "'fullyQualifiedNamespace' cannot be null.");
+        this.credentials = getTokenCredential(properties);
+
+        if (CoreUtils.isNullOrEmpty(fullyQualifiedNamespace)) {
+            throw LOGGER.logExceptionAsError(new IllegalArgumentException("'host' cannot be an empty string."));
+        }
+
+        if (!CoreUtils.isNullOrEmpty(properties.getEntityPath())) {
+            this.eventHubName = properties.getEntityPath();
+        }
+
+        return this;
     }
 
     private TokenCredential getTokenCredential(ConnectionStringProperties properties) {
@@ -405,13 +420,6 @@ public EventHubClientBuilder eventHubName(String eventHubName) {
         return this;
     }
 
-    private String getEventHubName() {
-        if (CoreUtils.isNullOrEmpty(eventHubName)) {
-            throw LOGGER.logExceptionAsError(new IllegalArgumentException("'eventHubName' cannot be an empty string."));
-        }
-        return eventHubName;
-    }
-
     /**
      * Toggles the builder to use the same connection for producers or consumers that are built from this instance. By
      * default, a new connection is constructed and used created for each Event Hub consumer or producer created.
@@ -714,7 +722,7 @@ EventHubClientBuilder verifyMode(SslDomain.VerifyMode verifyMode) {
      * @throws IllegalArgumentException If shared connection is not used and the credentials have not been set using
      *     either {@link #connectionString(String)} or {@link #credential(String, String, TokenCredential)}. Also, if
      *     {@link #consumerGroup(String)} have not been set. And if a proxy is specified but the transport type is not
-     *     {@link AmqpTransportType#AMQP_WEB_SOCKETS web sockets}.
+     *     {@link AmqpTransportType#AMQP_WEB_SOCKETS web sockets}.  Or, if the {@code eventHubName} has not been set.
      */
     public EventHubConsumerAsyncClient buildAsyncConsumerClient() {
         if (CoreUtils.isNullOrEmpty(consumerGroup)) {
@@ -733,7 +741,7 @@ public EventHubConsumerAsyncClient buildAsyncConsumerClient() {
      * @throws IllegalArgumentException If shared connection is not used and the credentials have not been set using
      *     either {@link #connectionString(String)} or {@link #credential(String, String, TokenCredential)}. Also, if
      *     {@link #consumerGroup(String)} have not been set. And if a proxy is specified but the transport type is not
-     *     {@link AmqpTransportType#AMQP_WEB_SOCKETS web sockets}.
+     *     {@link AmqpTransportType#AMQP_WEB_SOCKETS web sockets}.  Or, if the {@code eventHubName} has not been set.
      */
     public EventHubConsumerClient buildConsumerClient() {
         return buildClient().createConsumer(consumerGroup, prefetchCount);
@@ -747,6 +755,7 @@ public EventHubConsumerClient buildConsumerClient() {
      * @throws IllegalArgumentException If shared connection is not used and the credentials have not been set using
      *     either {@link #connectionString(String)} or {@link #credential(String, String, TokenCredential)}. Or, if a
      *     proxy is specified but the transport type is not {@link AmqpTransportType#AMQP_WEB_SOCKETS web sockets}.
+     *     Or, if the {@code eventHubName} has not been set.
      */
     public EventHubProducerAsyncClient buildAsyncProducerClient() {
         return buildAsyncClient().createProducer();
@@ -760,6 +769,7 @@ public EventHubProducerAsyncClient buildAsyncProducerClient() {
      * @throws IllegalArgumentException If shared connection is not used and the credentials have not been set using
      *     either {@link #connectionString(String)} or {@link #credential(String, String, TokenCredential)}. Or, if a
      *     proxy is specified but the transport type is not {@link AmqpTransportType#AMQP_WEB_SOCKETS web sockets}.
+     *     Or, if the {@code eventHubName} has not been set.
      */
     public EventHubProducerClient buildProducerClient() {
         return buildClient().createProducer();
@@ -786,7 +796,8 @@ public EventHubProducerClient buildProducerClient() {
      * @return A new {@link EventHubAsyncClient} instance with all the configured options.
      * @throws IllegalArgumentException if the credentials have not been set using either {@link
      *     #connectionString(String)} or {@link #credential(String, String, TokenCredential)}. Or, if a proxy is
-     *     specified but the transport type is not {@link AmqpTransportType#AMQP_WEB_SOCKETS web sockets}.
+     *     specified but the transport type is not {@link AmqpTransportType#AMQP_WEB_SOCKETS web sockets}. Or, if the
+     *     {@code eventHubName} has not been set.
      */
     EventHubAsyncClient buildAsyncClient() {
         if (retryOptions == null) {
@@ -898,6 +909,13 @@ Meter createMeter() {
 
     private EventHubConnectionProcessor buildConnectionProcessor(MessageSerializer messageSerializer, Meter meter) {
         final ConnectionOptions connectionOptions = getConnectionOptions();
+        final Supplier<String> getEventHubName = () -> {
+            if (CoreUtils.isNullOrEmpty(eventHubName)) {
+                throw LOGGER.logExceptionAsError(new IllegalArgumentException("'eventHubName' cannot be an empty string."));
+            }
+            return eventHubName;
+        };
+
         final Flux<EventHubAmqpConnection> connectionFlux = Flux.create(sink -> {
             sink.onRequest(request -> {
 
@@ -921,15 +939,15 @@ private EventHubConnectionProcessor buildConnectionProcessor(MessageSerializer m
                 final ReactorHandlerProvider handlerProvider = new ReactorHandlerProvider(provider, meter);
 
                 final EventHubAmqpConnection connection = new EventHubReactorAmqpConnection(connectionId,
-                    connectionOptions, getEventHubName(), provider, handlerProvider, tokenManagerProvider,
+                    connectionOptions, getEventHubName.get(), provider, handlerProvider, tokenManagerProvider,
                     messageSerializer);
 
                 sink.next(connection);
             });
         });
 
         return connectionFlux.subscribeWith(new EventHubConnectionProcessor(
-            connectionOptions.getFullyQualifiedNamespace(), getEventHubName(), connectionOptions.getRetry()));
+            connectionOptions.getFullyQualifiedNamespace(), getEventHubName.get(), connectionOptions.getRetry()));
     }
 
     private ConnectionOptions getConnectionOptions() {
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubClientBuilderTest.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubClientBuilderTest.java
@@ -24,8 +24,10 @@
 import java.util.Locale;
 import java.util.stream.Stream;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class EventHubClientBuilderTest {
     private static final String NAMESPACE_NAME = "dummyNamespaceName";
@@ -106,8 +108,6 @@ public void testConnectionStringWithSas() {
             .connectionString(connectionStringWithNoEntityPath, "eh-name"));
         assertNotNull(new EventHubClientBuilder()
             .connectionString(connectionStringWithEntityPath));
-        assertThrows(NullPointerException.class, () -> new EventHubClientBuilder()
-            .connectionString(connectionStringWithNoEntityPath));
         assertThrows(IllegalArgumentException.class, () -> new EventHubClientBuilder()
             .connectionString(connectionStringWithEntityPath, "eh-name-mismatch"));
     }
@@ -181,22 +181,33 @@ public void testConnectionWithAzureSasCredential() {
 
     @Test
     public void testCreatesClientWithTokenCredential() {
-        new EventHubClientBuilder()
+        EventHubClient eventHubClient = new EventHubClientBuilder()
             .credential(TOKEN_CREDENTIAL)
             .fullyQualifiedNamespace(NAMESPACE_NAME)
             .eventHubName(EVENT_HUB_NAME)
             .buildClient();
-        new EventHubClientBuilder()
+        EventHubProducerClient eventHubProducerClient = new EventHubClientBuilder()
             .credential(TOKEN_CREDENTIAL)
             .fullyQualifiedNamespace(NAMESPACE_NAME)
             .eventHubName(EVENT_HUB_NAME)
             .buildProducerClient();
-        new EventHubClientBuilder()
+        EventHubConsumerClient eventHubConsumerClient = new EventHubClientBuilder()
             .credential(TOKEN_CREDENTIAL)
             .fullyQualifiedNamespace(NAMESPACE_NAME)
             .eventHubName(EVENT_HUB_NAME)
             .consumerGroup("foo")
             .buildConsumerClient();
+
+        // Assert
+        assertNotNull(eventHubClient);
+        assertNotNull(eventHubProducerClient);
+        assertNotNull(eventHubConsumerClient);
+
+        assertEquals(EVENT_HUB_NAME, eventHubProducerClient.getEventHubName());
+        assertEquals(NAMESPACE_NAME, eventHubProducerClient.getFullyQualifiedNamespace());
+
+        assertEquals(EVENT_HUB_NAME, eventHubConsumerClient.getEventHubName());
+        assertEquals(NAMESPACE_NAME, eventHubConsumerClient.getFullyQualifiedNamespace());
     }
 
     @Test
@@ -233,6 +244,48 @@ public void testThrowsIfAttemptsToCreateClientWithTokenCredentialWithoutEventHub
             .buildConsumerClient());
     }
 
+    /**
+     * Verifies that we can pass an Event Hub namespace connection string and event hub name to create a client.
+     */
+    @Test
+    public void namespaceConnectionStringAndName() {
+        // Arrange
+        final String namespaceConnectionString = String.format("Endpoint=%s;SharedAccessKeyName=%s;SharedAccessKey=%s",
+            ENDPOINT, SHARED_ACCESS_KEY_NAME, SHARED_ACCESS_KEY);
+        final String fullyQualifiedDomainName = NAMESPACE_NAME + ENDPOINT_SUFFIX;
+
+        // Act
+        final EventHubProducerAsyncClient client = new EventHubClientBuilder()
+            .connectionString(namespaceConnectionString)
+            .eventHubName(EVENT_HUB_NAME)
+            .buildAsyncProducerClient();
+
+        // Assert
+        assertTrue(fullyQualifiedDomainName.equalsIgnoreCase(client.getFullyQualifiedNamespace()),
+            String.format("Expected: %s. Actual: %s%n", fullyQualifiedDomainName,
+                client.getFullyQualifiedNamespace()));
+
+        assertEquals(EVENT_HUB_NAME, client.getEventHubName());
+    }
+
+    /**
+     * Verifies that an exception is thrown when we try to construct a client without setting the event hub name.
+     */
+    @Test
+    public void namespaceConnectionStringThrowsNoEventHubName() {
+        // Arrange
+        final String namespaceConnectionString = String.format("Endpoint=%s;SharedAccessKeyName=%s;SharedAccessKey=%s",
+            ENDPOINT, SHARED_ACCESS_KEY_NAME, SHARED_ACCESS_KEY);
+
+        // Act & Assert
+        assertThrows(IllegalArgumentException.class, () -> new EventHubClientBuilder()
+            .connectionString(namespaceConnectionString)
+            .buildAsyncProducerClient());
+        assertThrows(IllegalArgumentException.class, () -> new EventHubClientBuilder()
+            .connectionString(namespaceConnectionString)
+            .buildAsyncConsumerClient());
+    }
+
     private static Stream<Arguments> getProxyConfigurations() {
         return Stream.of(
             Arguments.of("http://localhost:8080"),
