Use IterableStream<T>(Iterable) ctor in EventHubProperties

sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubClientBuilder.java

@@ -343,6 +343,21 @@ public EventHubClientBuilder prefetchCount(int prefetchCount) {
         return this;
     }
 
+    /**
+     * Package-private method that sets the scheduler for the created Event Hub client.
+     *
+     * TODO (conniey): Currently, the default is to use an elastic scheduler if none is specified to facilitate the
+     * possibility of legacy blocking code. However, we should consider if we should give consumers an option to use a
+     * parallel Scheduler. https://github.com/Azure/azure-sdk-for-java/issues/5466
+     *
+     * @param scheduler Scheduler to set.
+     * @return The updated {@link EventHubClientBuilder} object.
+     */
+    EventHubClientBuilder scheduler(Scheduler scheduler) {
+        this.scheduler = scheduler;
+        return this;
+    }
+
     /**
      * Creates a new {@link EventHubConsumerAsyncClient} based on the options set on this builder. Every time
      * {@code buildAsyncConsumer()} is invoked, a new instance of {@link EventHubConsumerAsyncClient} is created.

sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProperties.java

@@ -6,15 +6,14 @@
 import com.azure.core.annotation.Immutable;
 import com.azure.core.util.IterableStream;
 import com.azure.messaging.eventhubs.models.EventPosition;
-import reactor.core.publisher.Flux;
 
 import java.time.Instant;
 import java.util.Arrays;
+import java.util.Objects;
 
 /**
- * Holds information about an Event Hub which can come handy while performing operations like
- * {@link EventHubConsumerAsyncClient#receiveFromPartition(String, EventPosition) receiving events from a specific
- * partition}.
+ * Holds information about an Event Hub which can come handy while performing operations like {@link
+ * EventHubConsumerAsyncClient#receiveFromPartition(String, EventPosition) receiving events from a specific partition}.
  *
  * @see EventHubConsumerAsyncClient
  * @see EventHubConsumerClient
@@ -25,15 +24,20 @@ public final class EventHubProperties {
     private final Instant createdAt;
     private final IterableStream<String> partitionIds;
 
-    EventHubProperties(
-        final String name,
-        final Instant createdAt,
-        final String[] partitionIds) {
-        this.name = name;
-        this.createdAt = createdAt;
-        this.partitionIds = partitionIds != null
-            ? new IterableStream<>(Flux.fromArray(Arrays.copyOf(partitionIds, partitionIds.length)))
-            : new IterableStream<>(Flux.empty());
+    /**
+     * Creates an instance of {@link EventHubProperties}.
+     *
+     * @param name Name of the Event Hub.
+     * @param createdAt Datetime the Event Hub was created, in UTC.
+     * @param partitionIds The partitions ids in the Event Hub.
+     *
+     * @throws NullPointerException if {@code name}, {@code createdAt}, or {@code partitionIds} is {@code null}.
+     */
+    EventHubProperties(final String name, final Instant createdAt, final String[] partitionIds) {
+        this.name = Objects.requireNonNull(name, "'name' cannot be null.");
+        this.createdAt = Objects.requireNonNull(createdAt, "'createdAt' cannot be null.");
+        this.partitionIds = new IterableStream<>(Arrays.asList(
+            Objects.requireNonNull(partitionIds, "'partitionIds' cannot be null.")));
     }
 
     /**

sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubClientMetadataIntegrationTest.java

@@ -15,9 +15,6 @@
 import reactor.core.publisher.Flux;
 import reactor.test.StepVerifier;
 
-import java.security.InvalidKeyException;
-import java.security.NoSuchAlgorithmException;
-
 /**
  * Tests the metadata operations such as fetching partition properties and event hub properties.
  */
@@ -104,7 +101,7 @@ public void getPartitionPropertiesMultipleCalls() {
      * Verifies that error conditions are handled for fetching Event Hub metadata.
      */
     @Test
-    public void getPartitionPropertiesInvalidToken() throws InvalidKeyException, NoSuchAlgorithmException {
+    public void getPartitionPropertiesInvalidToken() {
         // Arrange
         final ConnectionStringProperties original = getConnectionStringProperties();
         final TokenCredential invalidTokenCredential = new EventHubSharedKeyCredential(
@@ -130,7 +127,7 @@ public void getPartitionPropertiesInvalidToken() throws InvalidKeyException, NoS
      * Verifies that error conditions are handled for fetching partition metadata.
      */
     @Test
-    public void getPartitionPropertiesNonExistentHub() throws InvalidKeyException, NoSuchAlgorithmException {
+    public void getPartitionPropertiesNonExistentHub() {
         // Arrange
         final ConnectionStringProperties original = getConnectionStringProperties();
         final TokenCredential validCredentials = new EventHubSharedKeyCredential(

sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientIntegrationTest.java

@@ -177,16 +177,20 @@ public void sendWithCredentials() {
             .buildAsyncProducerClient();
 
         // Act & Assert
-        StepVerifier.create(client.getEventHubProperties())
-            .assertNext(properties -> {
-                Assertions.assertEquals(getEventHubName(), properties.getName());
-                Assertions.assertEquals(2, properties.getPartitionIds().stream().count());
-            })
-            .expectComplete()
-            .verify(TIMEOUT);
-
-        StepVerifier.create(client.send(event, options))
-            .expectComplete()
-            .verify(TIMEOUT);
+        try {
+            StepVerifier.create(client.getEventHubProperties())
+                .assertNext(properties -> {
+                    Assertions.assertEquals(getEventHubName(), properties.getName());
+                    Assertions.assertEquals(2, properties.getPartitionIds().stream().count());
+                })
+                .expectComplete()
+                .verify(TIMEOUT);
+
+            StepVerifier.create(client.send(event, options))
+                .expectComplete()
+                .verify(TIMEOUT);
+        } finally {
+            dispose(client);
+        }
     }
 }

sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientIntegrationTest.java

@@ -140,4 +140,23 @@ public void sendAllPartitions() {
             producer.send(batch);
         }
     }
+
+    /**
+     * Sending with credentials.
+     */
+    @Test
+    public void sendWithCredentials() {
+        // Arrange
+        final EventData event = new EventData("body");
+        final SendOptions options = new SendOptions().setPartitionId(PARTITION_ID);
+        final EventHubProducerClient client = createBuilder(true)
+            .buildProducerClient();
+
+        // Act & Assert
+        try {
+            client.send(event, options);
+        } finally {
+            dispose(client);
+        }
+    }
 }

sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubPropertiesTest.java

@@ -36,22 +36,41 @@ public void setsProperties() {
     }
 
     /**
-     * Verifies that the {@link EventHubProperties#getPartitionIds()} array is not {@code null} when we pass {@code null}
-     * to the constructor.
+     * Throws when we try to set null partitionIds.
      */
     @Test
-    public void setsPropertiesNoPartitions() {
+    public void requiresPartitions() {
         // Arrange
         final String name = "Some-event-hub-name";
         final Instant instant = Instant.ofEpochSecond(145620);
 
-        // Act
-        final EventHubProperties eventHubProperties = new EventHubProperties(name, instant, null);
+        // Act & Assert
+        Assertions.assertThrows(NullPointerException.class, () -> new EventHubProperties(name, instant, null));
+    }
 
-        // Assert
-        Assertions.assertEquals(name, eventHubProperties.getName());
-        Assertions.assertEquals(instant, eventHubProperties.getCreatedAt());
-        Assertions.assertNotNull(eventHubProperties.getPartitionIds());
-        Assertions.assertEquals(0, eventHubProperties.getPartitionIds().stream().count());
+    /**
+     * Throws when we try to set null createdAt.
+     */
+    @Test
+    public void requiresCreatedAt() {
+        // Arrange
+        final String name = "Some-event-hub-name";
+        final String[] partitionIds = new String[]{"one-partition", "two-partition", "three-partition"};
+
+        // Act & Assert
+        Assertions.assertThrows(NullPointerException.class, () -> new EventHubProperties(name, null, partitionIds));
     }
+
+    /**
+     */
+    @Test
+    public void requiresName() {
+        // Arrange
+        final Instant instant = Instant.ofEpochSecond(145620);
+        final String[] partitionIds = new String[]{"one-partition", "two-partition", "three-partition"};
+
+        // Act & Assert
+        Assertions.assertThrows(NullPointerException.class, () -> new EventHubProperties(null, instant, partitionIds));
+    }
+
 }

sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/IntegrationTestBase.java

@@ -21,6 +21,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.TestInfo;
 import org.mockito.Mockito;
+import reactor.core.scheduler.Schedulers;
 
 import java.io.Closeable;
 import java.io.IOException;
@@ -159,7 +160,8 @@ protected EventHubClientBuilder createBuilder(boolean useCredentials) {
         final EventHubClientBuilder builder = new EventHubClientBuilder()
             .proxyOptions(ProxyOptions.SYSTEM_DEFAULTS)
             .retry(RETRY_OPTIONS)
-            .transportType(AmqpTransportType.AMQP);
+            .transportType(AmqpTransportType.AMQP)
+            .scheduler(Schedulers.newParallel("eh-integration"));
 
         if (useCredentials) {
             final String fqdn = getFullyQualifiedDomainName();