From 14979a9ea009c92d80a1feaee7c62e3dedfe92f6 Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 12:42:11 -0800 Subject: [PATCH 1/7] Using Iterable overload in EventHubProperties. --- .../com/azure/messaging/eventhubs/EventHubProperties.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProperties.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProperties.java index 0e3a89e52f43..4f380c9f6b2e 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProperties.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProperties.java @@ -6,10 +6,10 @@ 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.Collections; /** * Holds information about an Event Hub which can come handy while performing operations like @@ -31,9 +31,10 @@ public final class EventHubProperties { 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()); + ? new IterableStream<>(Arrays.asList(partitionIds)) + : new IterableStream<>(Collections.emptyList()); } /** From 5e5d62fbb3aa66eb28f60a4956f5cae5c620fbfc Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 12:43:24 -0800 Subject: [PATCH 2/7] Adding package-private ctor for Scheduler. --- .../eventhubs/EventHubClientBuilder.java | 15 +++++++++++++++ 1 file changed, 15 insertions(+) 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 index b944fe34cded..65fff011ad83 100644 --- 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 @@ -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. + * + * @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. From fde7abde39ea1e11618ca1ff751a54d476ba8e42 Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 12:43:38 -0800 Subject: [PATCH 3/7] Update Integration tests to use parallel scheduler. --- .../eventhubs/EventHubClientMetadataIntegrationTest.java | 7 ++----- .../com/azure/messaging/eventhubs/IntegrationTestBase.java | 4 +++- 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubClientMetadataIntegrationTest.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubClientMetadataIntegrationTest.java index a039b0a7fa39..0dd3db73b4bc 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubClientMetadataIntegrationTest.java +++ b/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( diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/IntegrationTestBase.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/IntegrationTestBase.java index 63270cb18124..e6f1fc55ec33 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/IntegrationTestBase.java +++ b/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(); From eded62c088d4c19265d657aa33d3f970e6a07ffd Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 12:48:59 -0800 Subject: [PATCH 4/7] Add link to GitHub issue. --- .../com/azure/messaging/eventhubs/EventHubClientBuilder.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 index 65fff011ad83..0d1e0f5c561d 100644 --- 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 @@ -348,7 +348,7 @@ public EventHubClientBuilder prefetchCount(int prefetchCount) { * * 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. + * parallel Scheduler. https://github.com/Azure/azure-sdk-for-java/issues/5466 * * @param scheduler Scheduler to set. * @return The updated {@link EventHubClientBuilder} object. From 5bc4e60b8120e88ee677c0e1fdef7d2fa2b48e09 Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 13:00:50 -0800 Subject: [PATCH 5/7] Adding try/catch around producer credential integration tests. --- ...HubProducerAsyncClientIntegrationTest.java | 26 +++++++++++-------- ...EventHubProducerClientIntegrationTest.java | 19 ++++++++++++++ 2 files changed, 34 insertions(+), 11 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientIntegrationTest.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientIntegrationTest.java index 32ada50cd853..de95c04f17e9 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientIntegrationTest.java +++ b/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); + } } } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientIntegrationTest.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientIntegrationTest.java index da15f3d36686..4d622d2decf4 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientIntegrationTest.java +++ b/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); + } + } } From ffc1feef6301ef312a5cde7b6689553a0bb43741 Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 13:26:28 -0800 Subject: [PATCH 6/7] Update EventHubProperties to throw if is negative. --- .../eventhubs/EventHubProperties.java | 31 ++++++++++--------- 1 file changed, 17 insertions(+), 14 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProperties.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProperties.java index 4f380c9f6b2e..97c64513a32c 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProperties.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProperties.java @@ -9,12 +9,11 @@ import java.time.Instant; import java.util.Arrays; -import java.util.Collections; +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,16 +24,20 @@ public final class EventHubProperties { private final Instant createdAt; private final IterableStream partitionIds; - EventHubProperties( - final String name, - final Instant createdAt, - final String[] partitionIds) { - this.name = name; - this.createdAt = createdAt; - - this.partitionIds = partitionIds != null - ? new IterableStream<>(Arrays.asList(partitionIds)) - : new IterableStream<>(Collections.emptyList()); + /** + * 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."))); } /** From 7c30b845de46001ef5d9d898b73a6b1ce2165342 Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 13:31:15 -0800 Subject: [PATCH 7/7] Adding tests. --- .../eventhubs/EventHubPropertiesTest.java | 39 ++++++++++++++----- 1 file changed, 29 insertions(+), 10 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubPropertiesTest.java b/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubPropertiesTest.java index c5bee70d1e72..7d9740d18de9 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubPropertiesTest.java +++ b/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)); + } + }