From 01ceb60e98e479d7fdc2d5056f52157adb60f247 Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 10:22:30 -0800 Subject: [PATCH 01/17] Fix javadoc. --- .../messaging/eventhubs/EventHubConsumerAsyncClient.java | 4 ++-- .../azure/messaging/eventhubs/EventHubConsumerClient.java | 4 ++-- .../messaging/eventhubs/EventHubProducerAsyncClient.java | 6 +++--- .../azure/messaging/eventhubs/EventHubProducerClient.java | 6 +++--- 4 files changed, 10 insertions(+), 10 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClient.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClient.java index 3af7f25ef199..30014b962d0d 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClient.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClient.java @@ -29,8 +29,8 @@ import static com.azure.core.util.FluxUtil.monoError; /** - * A consumer responsible for reading {@link EventData} from a specific Event Hub partition in the context of a specific - * consumer group. + * An asynchronous consumer responsible for reading {@link EventData} from either a specific Event Hub partition + * or all partitions in the context of a specific consumer group. * *

Creating an {@link EventHubConsumerAsyncClient}

*

Required parameters are {@code consumerGroup}, and credentials are required when diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerClient.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerClient.java index 322ba9f0fe8a..f0b07f50feb6 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerClient.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerClient.java @@ -22,8 +22,8 @@ import java.util.concurrent.atomic.AtomicInteger; /** - * A consumer responsible for reading {@link EventData} from either a specific Event Hub partition or all partitions in - * the context of a consumer group. + * A synchronous consumer responsible for reading {@link EventData} from a Event Hub partition in the context of + * a specific consumer group. * *

Creating a synchronous consumer

*

Required parameters are {@code consumerGroup} and credentials when creating a consumer.

diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java index f8559e44c997..f68e4eee65f1 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java @@ -50,9 +50,9 @@ import static com.azure.messaging.eventhubs.implementation.ClientConstants.MAX_MESSAGE_LENGTH_BYTES; /** - * A producer responsible for transmitting {@link EventData} to a specific Event Hub, grouped together in batches. - * Depending on the options specified at creation, the producer may be created to allow event data to be automatically - * routed to an available partition or specific to a partition. + * An asynchronous producer responsible for transmitting {@link EventData} to a specific Event Hub, grouped + * together in batches. Depending on the options specified at creation, the producer may be created to allow event data + * to be automatically routed to an available partition or specific to a partition. * *

* Allowing automatic routing of partitions is recommended when: diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerClient.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerClient.java index 356e93c4a8f6..76e72d960f5a 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerClient.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerClient.java @@ -15,9 +15,9 @@ import java.util.Objects; /** - * A producer responsible for transmitting {@link EventData} to a specific Event Hub, grouped together in batches. - * Depending on the options specified at creation, the producer may be created to allow event data to be automatically - * routed to an available partition or specific to a partition. + * A synchronous producer responsible for transmitting {@link EventData} to a specific Event Hub, grouped + * together in batches. Depending on the options specified at creation, the producer may be created to allow event data + * to be automatically routed to an available partition or specific to a partition. * *

* Allowing automatic routing of partitions is recommended when: From 99ee9e19443a9ba771a459e4ede4d7e8843edb02 Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 13:37:13 -0800 Subject: [PATCH 02/17] Update Metadata sample. --- .../com/azure/messaging/eventhubs/GetEventHubMetadata.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java index ecfb4c6ada5c..72bca134ea9d 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java @@ -39,8 +39,10 @@ public static void main(String[] args) throws InterruptedException { System.out.println("The Event Hub has the following properties:"); System.out.printf( "Event Hub Name: %s; Partition Id: %s; Is partition empty? %s; First Sequence Number: %s; " - + "Last Enqueued Time: %s; Last Enqueued Sequence Number: %s; Last Enqueued Offset: %s", - properties.getEventHubName(), properties.getId(), properties.isEmpty(), + + "Last Enqueued Time: %s; Last Enqueued Sequence Number: %s; Last Enqueued Offset: %s%n", + properties.getEventHubName(), + properties.getId(), + properties.isEmpty(), properties.getBeginningSequenceNumber(), properties.getLastEnqueuedTime(), properties.getLastEnqueuedSequenceNumber(), From 0fb06b3f5eb0cf7bcd13319c39c5f09ac42dfb0c Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 14:40:20 -0800 Subject: [PATCH 03/17] Split string to match docs. --- .../EventHubConsumerAsyncClientJavaDocCodeSamples.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClientJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClientJavaDocCodeSamples.java index d6e2ffc1ce59..8485f7088c90 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClientJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClientJavaDocCodeSamples.java @@ -23,7 +23,8 @@ public void initialization() { // BEGIN: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.instantiation // The required parameters are `consumerGroup` and a way to authenticate with Event Hubs using credentials. EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() - .connectionString("event-hub-instance-connection-string") + .connectionString("Endpoint={fully-qualified-namespace};SharedAccessKeyName={policy-name};" + + "SharedAccessKey={key};EntityPath={eh-name}") .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) .buildAsyncConsumerClient(); // END: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.instantiation From 65b25962a01632a76d9098078195ab4e03fac79c Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 15:07:06 -0800 Subject: [PATCH 04/17] Update sample for publishing events to a partition using EventDataBatch. --- .../PublishEventsToSpecificPartition.java | 39 +++++++++++++------ 1 file changed, 28 insertions(+), 11 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java index d40335198791..1dabb584226e 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java @@ -4,6 +4,7 @@ import com.azure.messaging.eventhubs.models.CreateBatchOptions; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import java.time.Duration; import java.util.concurrent.atomic.AtomicReference; @@ -58,23 +59,39 @@ public static void main(String[] args) { // We try to add as many events as a batch can fit based on the event size and send to Event Hub when // the batch can hold no more events. Create a new batch for next set of events and repeat until all events // are sent. - data.subscribe(event -> { + final Mono sendOperation = data.flatMap(event -> { final EventDataBatch batch = currentBatch.get(); - if (!batch.tryAdd(event)) { + if (batch.tryAdd(event)) { + return Mono.empty(); + } + + // The batch is full, so we create a new batch and send the batch. Mono.when completes when both operations + // have completed. + return Mono.when( + producer.send(batch), producer.createBatch(options).map(newBatch -> { currentBatch.set(newBatch); - return producer.send(batch); - }).block(); - } - }, error -> System.err.println("Error received:" + error), - () -> { + + // Add that event that we couldn't before. + newBatch.tryAdd(event); + return newBatch; + })); + }).then() + .doFinally(signal -> { final EventDataBatch batch = currentBatch.getAndSet(null); if (batch != null) { - producer.send(batch).block(); + producer.send(batch).block(Duration.ofSeconds(60)); } - - // Disposing of our producer. - producer.close(); }); + + // The sendOperation creation and assignment is not a blocking call. It does not get invoked until there is a + // subscriber to that operation. For the purpose of this example, we block so the program does not end before + // the send operation is complete. Any of the `.subscribe` overloads also work to start the Mono asynchronously. + try { + sendOperation.block(Duration.ofSeconds(60)); + } finally { + // Disposing of our producer. + producer.close(); + } } } From 22e2e46d811a4cfaabb59dcca03b21bac60fd60b Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 15:15:51 -0800 Subject: [PATCH 05/17] Update documentation for PublishEventDataBatch. --- .../eventhubs/PublishEventDataBatch.java | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventDataBatch.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventDataBatch.java index 3fb8f293331e..1f35624f4706 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventDataBatch.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventDataBatch.java @@ -4,6 +4,7 @@ import com.azure.messaging.eventhubs.models.CreateBatchOptions; import reactor.core.publisher.Flux; +import reactor.core.scheduler.Schedulers; import java.util.concurrent.atomic.AtomicReference; @@ -30,12 +31,13 @@ public static void main(String[] args) { // 2. Creating an Event Hub instance. // 3. Creating a "Shared access policy" for your Event Hub instance. // 4. Copying the connection string from the policy's properties. - final String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};" + String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};" + "SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; // Create a producer. EventHubProducerAsyncClient producer = new EventHubClientBuilder() .connectionString(connectionString) + .scheduler(Schedulers.parallel()) .buildAsyncProducerClient(); // Creating a batch where we want the events ending up in the same partition by setting the partition key. @@ -51,6 +53,9 @@ public static void main(String[] args) { if (!batch.tryAdd(event)) { producer.createBatch(options).map(newBatch -> { currentBatch.set(newBatch); + + // Adding that event we couldn't add before. + newBatch.tryAdd(event); return producer.send(batch); }).block(); } @@ -64,5 +69,17 @@ public static void main(String[] args) { // Disposing of our producer. producer.close(); }); + + // Sleeping this thread because we want to wait for all the events to send before ending the program. + // `.subscribe` is not a blocking call. It coordinates the callbacks and starts the send operation, but does not + // wait for it to complete. + // Customers can chain together Reactor operators like .then() and `.doFinally` if they want an operation to run + // after all the events have been transmitted. .block() can also be used to turn the call into a synchronous + // operation. + try { + Thread.sleep(5000); + } catch (InterruptedException e) { + e.printStackTrace(); + } } } From 724f1324dc4794b0d20877bb675bf88c4b421421 Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 15:55:06 -0800 Subject: [PATCH 06/17] Update publish samples. --- .../PublishEventsToSpecificPartition.java | 3 +- .../PublishEventsWithAzureIdentity.java | 89 +++++++++++++++++++ .../PublishEventsWithCustomMetadata.java | 82 ++++++++--------- .../PublishEventsWithPartitionKey.java | 5 +- ... PublishEventsWithSizeLimitedBatches.java} | 46 +++++----- 5 files changed, 158 insertions(+), 67 deletions(-) create mode 100644 sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java rename sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/{PublishEventDataBatch.java => PublishEventsWithSizeLimitedBatches.java} (66%) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java index 1dabb584226e..0c77aba0cf66 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java @@ -51,8 +51,7 @@ public static void main(String[] args) { // Create a batch to send the events. final CreateBatchOptions options = new CreateBatchOptions() - .setPartitionId(firstPartition) - .setMaximumSizeInBytes(256); + .setPartitionId(firstPartition); final AtomicReference currentBatch = new AtomicReference<>( producer.createBatch(options).block()); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java new file mode 100644 index 000000000000..2cd2a508f44f --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java @@ -0,0 +1,89 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.azure.messaging.eventhubs; + +import com.azure.core.credential.TokenCredential; +import com.azure.identity.DefaultAzureCredentialBuilder; +import reactor.core.publisher.Flux; + +import java.util.concurrent.atomic.AtomicReference; + +import static java.nio.charset.StandardCharsets.UTF_8; + +/** + * Sample demonstrates how to send an {@link EventDataBatch} to an Azure Event Hub using Azure Identity. + */ +public class PublishEventsWithAzureIdentity { + /** + * Main method to invoke this demo on how to send an {@link EventDataBatch} to an Azure Event Hub. + * + * @param args Unused arguments to the program. + */ + public static void main(String[] args) { + Flux telemetryEvents = Flux.just( + new EventData("Roast beef".getBytes(UTF_8)), + new EventData("Cheese".getBytes(UTF_8)), + new EventData("Tofu".getBytes(UTF_8)), + new EventData("Turkey".getBytes(UTF_8))); + + // The default azure credential checks multiple locations for credentials and determines the best one to use. + // For the purpose of this sample, create a service principal and set the following environment variables. + // See https://docs.microsoft.com/azure/active-directory/develop/howto-create-service-principal-portal for + // information on how to create a service principal. + System.setProperty("AZURE_CLIENT_ID", "<>"); + System.setProperty("AZURE_CLIENT_ID", "<>"); + System.setProperty("AZURE_TENANT_ID", "<>"); + + // DefaultAzureCredentialBuilder exists inside the azure-identity package. + TokenCredential credential = new DefaultAzureCredentialBuilder() + .build(); + + // Create a producer. + // "<>" will look similar to "{your-namespace}.servicebus.windows.net" + // "<>" will be the name of the Event Hub instance you created inside the Event Hubs namespace. + EventHubProducerAsyncClient producer = new EventHubClientBuilder() + .credential( + "<>", + "<>", + credential) + .buildAsyncProducerClient(); + + final AtomicReference currentBatch = new AtomicReference<>( + producer.createBatch().block()); + + // The sample Flux contains three events, but it could be an infinite stream of telemetry events. + telemetryEvents.subscribe(event -> { + final EventDataBatch batch = currentBatch.get(); + if (!batch.tryAdd(event)) { + producer.createBatch().map(newBatch -> { + currentBatch.set(newBatch); + + // Adding that event we couldn't add before. + newBatch.tryAdd(event); + return producer.send(batch); + }).block(); + } + }, error -> System.err.println("Error received:" + error), + () -> { + final EventDataBatch batch = currentBatch.getAndSet(null); + if (batch != null) { + producer.send(batch).block(); + } + + // Disposing of our producer. + producer.close(); + }); + + // Sleeping this thread because we want to wait for all the events to send before ending the program. + // `.subscribe` is not a blocking call. It coordinates the callbacks and starts the send operation, but does not + // wait for it to complete. + // Customers can chain together Reactor operators like .then() and `.doFinally` if they want an operation to run + // after all the events have been transmitted. .block() can also be used to turn the call into a synchronous + // operation. + try { + Thread.sleep(5000); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } +} diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java index fb4e1560b295..52ef3496c3d5 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java @@ -2,8 +2,8 @@ // Licensed under the MIT License. package com.azure.messaging.eventhubs; -import com.azure.messaging.eventhubs.models.CreateBatchOptions; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import java.time.Duration; import java.util.concurrent.atomic.AtomicReference; @@ -11,7 +11,8 @@ import static java.nio.charset.StandardCharsets.UTF_8; /** - * Sample demonstrates how to sent events to a specific event hub by defining partition ID in producer option only. + * Sample demonstrates how to sent events to Event Hubs, allowing the service to load-balance the events between all + * partitions. */ public class PublishEventsWithCustomMetadata { private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); @@ -37,62 +38,63 @@ public static void main(String[] args) { // Because an event consists mainly of an opaque set of bytes, it may be difficult for consumers of those events // to make informed decisions about how to process them. // - // In order to allow event publishers to offer better context for consumers, event data may also contain custom metadata, - // in the form of a set of key/value pairs. This metadata is not used by, or in any way meaningful to, the Event Hubs - // service; it exists only for coordination between event publishers and consumers. + // In order to allow event publishers to offer better context for consumers, event data may also contain custom + // metadata, in the form of a set of key/value pairs. This metadata is not used by, or in any way meaningful to, + // the Event Hubs service; it exists only for coordination between event publishers and consumers. // - // One common scenario for the inclusion of metadata is to provide a hint about the type of data contained by an event, - // so that consumers understand its format and can deserialize it appropriately. - // - // We will publish two events based on simple sentences, but will attach some custom metadata with - // pretend type names and other hints. Note that the set of metadata is unique to an event; there is no need for every - // event in a batch to have the same metadata properties available nor the same data type for those properties. + // One common scenario for the inclusion of metadata is to provide a hint about the type of data contained by an + // event, so that consumers understand its format and can deserialize it appropriately. EventData firstEvent = new EventData("EventData Sample 1".getBytes(UTF_8)); firstEvent.getProperties().put("EventType", "com.microsoft.samples.hello-event"); firstEvent.getProperties().put("priority", 1); firstEvent.getProperties().put("score", 9.0); - EventData secEvent = new EventData("EventData Sample 2".getBytes(UTF_8)); - secEvent.getProperties().put("EventType", "com.microsoft.samples.goodbye-event"); - secEvent.getProperties().put("priority", "17"); - secEvent.getProperties().put("blob", 10); - - final Flux data = Flux.just(firstEvent, secEvent); + EventData secondEvent = new EventData("EventData Sample 2".getBytes(UTF_8)); + secondEvent.getProperties().put("EventType", "com.microsoft.samples.goodbye-event"); + secondEvent.getProperties().put("priority", "17"); + secondEvent.getProperties().put("blob", 10); - // We want to send events to the a specific partition. For the sake of this sample, we take the first partition - // identifier. - // .blockFirst() here is used to synchronously block until the first partition id is emitted. The maximum wait - // time is set by passing in the OPERATION_TIMEOUT value. If no item is emitted before the timeout elapses, a - // TimeoutException is thrown. - String firstPartition = producer.getPartitionIds().blockFirst(OPERATION_TIMEOUT); + final Flux data = Flux.just(firstEvent, secondEvent); - // Create a batch to send the events. - final CreateBatchOptions options = new CreateBatchOptions() - .setPartitionId(firstPartition) - .setMaximumSizeInBytes(256); final AtomicReference currentBatch = new AtomicReference<>( - producer.createBatch(options).block()); + producer.createBatch().block()); // We try to add as many events as a batch can fit based on the event size and send to Event Hub when // the batch can hold no more events. Create a new batch for next set of events and repeat until all events // are sent. - data.subscribe(event -> { + final Mono sendOperation = data.flatMap(event -> { final EventDataBatch batch = currentBatch.get(); - if (!batch.tryAdd(event)) { - producer.createBatch(options).map(newBatch -> { - currentBatch.set(newBatch); - return producer.send(batch); - }).block(); + if (batch.tryAdd(event)) { + return Mono.empty(); } - }, error -> System.err.println("Error received:" + error), - () -> { + + // The batch is full, so we create a new batch and send the batch. Mono.when completes when both operations + // have completed. + return Mono.when( + producer.send(batch), + producer.createBatch().map(newBatch -> { + currentBatch.set(newBatch); + + // Add that event that we couldn't before. + newBatch.tryAdd(event); + return newBatch; + })); + }).then() + .doFinally(signal -> { final EventDataBatch batch = currentBatch.getAndSet(null); if (batch != null) { - producer.send(batch).block(); + producer.send(batch).block(OPERATION_TIMEOUT); } - - // Disposing of our producer. - producer.close(); }); + + // The sendOperation creation and assignment is not a blocking call. It does not get invoked until there is a + // subscriber to that operation. For the purpose of this example, we block so the program does not end before + // the send operation is complete. Any of the `.subscribe` overloads also work to start the Mono asynchronously. + try { + sendOperation.block(OPERATION_TIMEOUT); + } finally { + // Disposing of our producer. + producer.close(); + } } } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java index 23918d80a7b7..006b388329b6 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java @@ -10,7 +10,7 @@ import static java.nio.charset.StandardCharsets.UTF_8; /** - * Send a list of events with send option configured + * Send a Flux of events using a partition key. */ public class PublishEventsWithPartitionKey { @@ -53,8 +53,7 @@ public static void main(String[] args) { // we can only be assured that it will be a consistent choice of partition. If you have a need to understand which // exact partition an event is published to, you will need to use an Event Hub producer associated with that partition. final CreateBatchOptions options = new CreateBatchOptions() - .setPartitionKey("basketball") - .setMaximumSizeInBytes(256); + .setPartitionKey("basketball"); final AtomicReference currentBatch = new AtomicReference<>( producer.createBatch(options).block()); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventDataBatch.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java similarity index 66% rename from sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventDataBatch.java rename to sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java index 1f35624f4706..8a44f0c28a34 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventDataBatch.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java @@ -1,19 +1,19 @@ // Copyright (c) Microsoft Corporation. All rights reserved. // Licensed under the MIT License. + package com.azure.messaging.eventhubs; import com.azure.messaging.eventhubs.models.CreateBatchOptions; import reactor.core.publisher.Flux; -import reactor.core.scheduler.Schedulers; import java.util.concurrent.atomic.AtomicReference; import static java.nio.charset.StandardCharsets.UTF_8; /** - * Sample demonstrates how to send an {@link EventDataBatch} to an Azure Event Hub. + * Demonstrates how to publish events when there is a size constraint on batch size. */ -public class PublishEventDataBatch { +public class PublishEventsWithSizeLimitedBatches { /** * Main method to invoke this demo on how to send an {@link EventDataBatch} to an Azure Event Hub. * @@ -31,43 +31,40 @@ public static void main(String[] args) { // 2. Creating an Event Hub instance. // 3. Creating a "Shared access policy" for your Event Hub instance. // 4. Copying the connection string from the policy's properties. - String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};" - + "SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; + String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; - // Create a producer. + // Instantiate a client that will be used to call the service. EventHubProducerAsyncClient producer = new EventHubClientBuilder() .connectionString(connectionString) - .scheduler(Schedulers.parallel()) .buildAsyncProducerClient(); - // Creating a batch where we want the events ending up in the same partition by setting the partition key. + // In cases where developers need to size limit their batch size, they can use `setMaximumSizeInBytes` to limit + // the size of their EventDataBatch. By default, it will be the max size allowed by the underlying link. + // Since there is no partition id or partition key set, the Event Hubs service will automatically load balance + // the events between all available partitions. final CreateBatchOptions options = new CreateBatchOptions() - .setPartitionKey("sandwiches") .setMaximumSizeInBytes(256); final AtomicReference currentBatch = new AtomicReference<>( producer.createBatch(options).block()); // The sample Flux contains three events, but it could be an infinite stream of telemetry events. telemetryEvents.subscribe(event -> { - final EventDataBatch batch = currentBatch.get(); - if (!batch.tryAdd(event)) { - producer.createBatch(options).map(newBatch -> { - currentBatch.set(newBatch); + final EventDataBatch batch = currentBatch.get(); + if (!batch.tryAdd(event)) { + producer.createBatch(options).map(newBatch -> { + currentBatch.set(newBatch); - // Adding that event we couldn't add before. - newBatch.tryAdd(event); - return producer.send(batch); - }).block(); - } - }, error -> System.err.println("Error received:" + error), + // Adding that event we couldn't add before. + newBatch.tryAdd(event); + return producer.send(batch); + }).block(); + } + }, error -> System.err.println("Error received:" + error), () -> { final EventDataBatch batch = currentBatch.getAndSet(null); if (batch != null) { producer.send(batch).block(); } - - // Disposing of our producer. - producer.close(); }); // Sleeping this thread because we want to wait for all the events to send before ending the program. @@ -80,6 +77,11 @@ public static void main(String[] args) { Thread.sleep(5000); } catch (InterruptedException e) { e.printStackTrace(); + } finally { + // Disposing of our producer. + producer.close(); } } } + + From 7886c8d1a3f18500623adb3e8a6030dd441e4883 Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 15:58:30 -0800 Subject: [PATCH 07/17] Fixing documentation on publish events. --- .../eventhubs/PublishEventsToSpecificPartition.java | 6 +++--- .../eventhubs/PublishEventsWithCustomMetadata.java | 5 +++-- .../eventhubs/PublishEventsWithSizeLimitedBatches.java | 3 ++- 3 files changed, 8 insertions(+), 6 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java index 0c77aba0cf66..550c3929d7c3 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java @@ -12,14 +12,14 @@ import static java.nio.charset.StandardCharsets.UTF_8; /** - * Sample demonstrates how to sent events to specific event hub by define partition ID in producer option only. + * Sample demonstrates how to sent events to specific event hub by defining partition id using + * {@link CreateBatchOptions#setPartitionId(String)}. */ public class PublishEventsToSpecificPartition { private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); /** - * Main method to invoke this demo about how to send a batch of events with partition ID configured in producer - * option to an Azure Event Hub instance. + * Main method to invoke this demo about how to send a batch of events with partition ID configured. * * @param args Unused arguments to the program. */ diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java index 52ef3496c3d5..5e2ddaa02be0 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java @@ -11,8 +11,9 @@ import static java.nio.charset.StandardCharsets.UTF_8; /** - * Sample demonstrates how to sent events to Event Hubs, allowing the service to load-balance the events between all - * partitions. + * Sample demonstrates how to sent events with custom metadata to Event Hubs using {@link EventData#getProperties()}. + * Allows the service to load-balance the events between all partitions by using + * {@link EventHubProducerAsyncClient#createBatch()} which uses the default set of create batch options. */ public class PublishEventsWithCustomMetadata { private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java index 8a44f0c28a34..444a7f80165a 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java @@ -11,7 +11,8 @@ import static java.nio.charset.StandardCharsets.UTF_8; /** - * Demonstrates how to publish events when there is a size constraint on batch size. + * Demonstrates how to publish events when there is a size constraint on batch size using + * {@link CreateBatchOptions#setMaximumSizeInBytes(int)}. */ public class PublishEventsWithSizeLimitedBatches { /** From 9045f9bbbaa9a01e9435a8eb967ee0edf23b92cc Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 16:04:26 -0800 Subject: [PATCH 08/17] Update GetMetadata sample. --- .../eventhubs/GetEventHubMetadata.java | 40 ++++++------------- .../PublishEventsWithAzureIdentity.java | 6 +-- 2 files changed, 15 insertions(+), 31 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java index 72bca134ea9d..39a3a6b4daf4 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java @@ -2,21 +2,16 @@ // Licensed under the MIT License. package com.azure.messaging.eventhubs; -import java.util.concurrent.Semaphore; - /** - * Demonstrates how to fetch metadata from an Event Hub's partitions. + * Demonstrates how to fetch metadata from an Event Hub's partitions using synchronous client. */ public class GetEventHubMetadata { /** * Demonstrates how to get metadata from an Event Hub's partitions. * * @param args Unused arguments to the sample. - * @throws InterruptedException if the semaphore could not be acquired. */ - public static void main(String[] args) throws InterruptedException { - Semaphore semaphore = new Semaphore(1); - + public static void main(String[] args) { // The connection string value can be obtained by: // 1. Going to your Event Hubs namespace in Azure Portal. // 2. Creating an Event Hub instance. @@ -24,19 +19,16 @@ public static void main(String[] args) throws InterruptedException { // 4. Copying the connection string from the policy's properties. String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; - // Instantiate a client that will be used to call the service. - EventHubProducerAsyncClient client = new EventHubClientBuilder() + // Instantiate a client that will be used to call the service. Using a try-resource block, so it disposes of + // the client when we are done. + try (EventHubProducerClient client = new EventHubClientBuilder() .connectionString(connectionString) - .buildAsyncProducerClient(); + .buildProducerClient()) { - // Acquiring the semaphore so that this sample does not end before all the partition properties are fetched. - semaphore.acquire(); - - // Querying the partition identifiers for the Event Hub. Then calling client.getPartitionProperties with the - // identifier to get information about each partition. - client.getPartitionIds().flatMap(partitionId -> client.getPartitionProperties(partitionId)) - .subscribe(properties -> { - System.out.println("The Event Hub has the following properties:"); + // Querying the partition identifiers for the Event Hub. Then calling client.getPartitionProperties with the + // identifier to get information about each partition. + for (String partitionId : client.getPartitionIds()) { + PartitionProperties properties = client.getPartitionProperties(partitionId); System.out.printf( "Event Hub Name: %s; Partition Id: %s; Is partition empty? %s; First Sequence Number: %s; " + "Last Enqueued Time: %s; Last Enqueued Sequence Number: %s; Last Enqueued Offset: %s%n", @@ -47,15 +39,7 @@ public static void main(String[] args) throws InterruptedException { properties.getLastEnqueuedTime(), properties.getLastEnqueuedSequenceNumber(), properties.getLastEnqueuedOffset()); - }, error -> { - System.err.println("Error occurred while fetching partition properties: " + error.toString()); - }, () -> { - // Releasing the semaphore now that we've finished querying for partition properties. - semaphore.release(); - }); - - System.out.println("Waiting for partition properties to complete..."); - semaphore.acquire(); - System.out.println("Finished."); + } + } } } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java index 2cd2a508f44f..11d6a8068131 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java @@ -69,9 +69,6 @@ public static void main(String[] args) { if (batch != null) { producer.send(batch).block(); } - - // Disposing of our producer. - producer.close(); }); // Sleeping this thread because we want to wait for all the events to send before ending the program. @@ -84,6 +81,9 @@ public static void main(String[] args) { Thread.sleep(5000); } catch (InterruptedException e) { e.printStackTrace(); + } finally { + // Disposing of our producer. + producer.close(); } } } From e8ab55e62684223748f26a7fe1322969dd78a8bf Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 16:21:20 -0800 Subject: [PATCH 09/17] Update consume events. --- .../{ConsumeEvent.java => ConsumeEvents.java} | 2 +- ...EventsFromKnownSequenceNumberPosition.java | 112 +++++++----------- 2 files changed, 47 insertions(+), 67 deletions(-) rename sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/{ConsumeEvent.java => ConsumeEvents.java} (99%) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEvent.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEvents.java similarity index 99% rename from sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEvent.java rename to sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEvents.java index 1aaea8b63d86..f89dcab19dfb 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEvent.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEvents.java @@ -16,7 +16,7 @@ /** * Sample demonstrates how to receive events from an Azure Event Hub instance. */ -public class ConsumeEvent { +public class ConsumeEvents { private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); private static final int NUMBER_OF_EVENTS = 10; diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEventsFromKnownSequenceNumberPosition.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEventsFromKnownSequenceNumberPosition.java index 733ef6095470..a2559f15e574 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEventsFromKnownSequenceNumberPosition.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEventsFromKnownSequenceNumberPosition.java @@ -4,30 +4,27 @@ import com.azure.messaging.eventhubs.models.EventPosition; import com.azure.messaging.eventhubs.models.SendOptions; -import reactor.core.Disposable; import java.time.Duration; -import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicBoolean; import static java.nio.charset.StandardCharsets.UTF_8; /** - * Sample demonstrates how to receive events starting from the specific sequence number position in an Event Hub instance. + * Sample demonstrates how to receive events starting from the specific sequence number position in an Event Hub + * instance. It also demonstrates how to publish events to a specific partition. */ public class ConsumeEventsFromKnownSequenceNumberPosition { private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); - private static long lastEnqueuedSequenceNumber = -1; - private static String lastEnqueuedSequencePartitionId = null; /** - * Main method to invoke this demo about how to receive event from a known sequence number position in an Azure Event Hub instance. + * Main method to invoke this demo about how to receive event from a known sequence number position in an Azure + * Event Hub instance. * * @param args Unused arguments to the program. - * @throws InterruptedException The countdown latch was interrupted while waiting for this sample to - * complete. */ - public static void main(String[] args) throws InterruptedException { - Semaphore semaphore = new Semaphore(0); + public static void main(String[] args) { + final AtomicBoolean isRunning = new AtomicBoolean(true); // The connection string value can be obtained by: // 1. Going to your Event Hubs namespace in Azure Portal. @@ -36,76 +33,59 @@ public static void main(String[] args) throws InterruptedException { // 4. Copying the connection string from the policy's properties. String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; - EventHubClientBuilder builder = new EventHubClientBuilder() - .connectionString(connectionString) - .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME); - - EventHubConsumerAsyncClient earliestConsumer = builder.buildAsyncConsumerClient(); - - earliestConsumer.getPartitionIds().flatMap(partitionId -> earliestConsumer.getPartitionProperties(partitionId)) - .subscribe( - properties -> { - if (!properties.isEmpty()) { - lastEnqueuedSequenceNumber = properties.getLastEnqueuedSequenceNumber(); - lastEnqueuedSequencePartitionId = properties.getId(); - } - }, - error -> System.err.println("Error occurred while fetching partition properties: " + error.toString()), - () -> { - // Releasing the semaphore now that we've finished querying for partition properties. - semaphore.release(); - }); - - System.out.println("Waiting for partition properties to complete..."); - // Acquiring the semaphore so that this sample does not end before all the partition properties are fetched. - semaphore.acquire(); - System.out.printf("Last enqueued sequence number: %s%n", lastEnqueuedSequenceNumber); + final EventHubClientBuilder builder = new EventHubClientBuilder() + .connectionString(connectionString); + + // The consumer group is required for consuming events. + final EventHubConsumerAsyncClient consumer = builder + .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .buildAsyncConsumerClient(); + + // Find the first non-empty partition we can start consuming from. + // Block on it because we don't know what partition to start reading from, yet. + final PartitionProperties nonEmptyPartition = consumer.getPartitionIds() + .flatMap(partitionId -> consumer.getPartitionProperties(partitionId)) + .filter(properties -> !properties.isEmpty()) + .blockFirst(OPERATION_TIMEOUT); // Make sure to have at least one non-empty event hub in order to continue the sample execution // if you don't have an non-empty event hub, try with another example 'SendEvent' in the same directory. - if (lastEnqueuedSequenceNumber == -1 || lastEnqueuedSequencePartitionId == null) { - System.err.println("All event hubs are empty"); + if (nonEmptyPartition == null) { + System.err.println("All event hub partitions are empty"); System.exit(0); } - // Create a consumer. - // The "$Default" consumer group is created by default. This value can be found by going to the Event Hub - // instance you are connecting to, and selecting the "Consumer groups" page. EventPosition.latest() tells the - // service we only want events that are sent to the partition after we begin listening. - EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() - .connectionString(connectionString) - .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) - .buildAsyncConsumerClient(); + // ex. The last enqueued sequence number is 99. If isInclusive is true, the received event starting from + // the same event with sequence number of '99'. Otherwise, the event with sequence number of '100' will + // be the first event received. + final EventPosition position = EventPosition.fromSequenceNumber( + nonEmptyPartition.getLastEnqueuedSequenceNumber(), true); - // We start receiving any events that come from `firstPartition`, print out the contents, and decrement the - // countDownLatch. - final EventPosition position = EventPosition.fromSequenceNumber(lastEnqueuedSequenceNumber, false); - Disposable subscription = consumer.receiveFromPartition(lastEnqueuedSequencePartitionId, position).subscribe(partitionEvent -> { - EventData event = partitionEvent.getData(); - String contents = new String(event.getBody(), UTF_8); - // ex. The last enqueued sequence number is 99. If isInclusive is true, the received event starting from the same - // event with sequence number of '99'. Otherwise, the event with sequence number of '100' will be the first - // event received. - System.out.println(String.format("Receiving an event starting from the sequence number: %s. Contents: %s", - event.getSequenceNumber(), contents)); + // We start receiving any events that come from that non-empty partition, print out the contents. + // We keep receiving events while `takeWhile` resolves to true, that is, the program is still running. + consumer.receiveFromPartition(nonEmptyPartition.getId(), position) + .takeWhile(ignored -> isRunning.get()) + .subscribe(partitionEvent -> { + EventData event = partitionEvent.getData(); + String contents = new String(event.getBody(), UTF_8); - semaphore.release(); - }); + System.out.println(String.format("Event sequence number number: %s. Contents: %s%n", + event.getSequenceNumber(), contents)); + }); - EventHubProducerAsyncClient producer = builder.buildAsyncProducerClient(); + // Create a producer. + final EventHubProducerClient producer = builder.buildProducerClient(); - // Because the consumer is only listening to new events, we need to send some events to that partition. - // This sends the events to `lastEnqueuedSequencePartitionId`. - SendOptions sendOptions = new SendOptions().setPartitionId(lastEnqueuedSequencePartitionId); + // Because the consumer is only listening to new events after the last enqueued event was received, we need to + // send some events to that partition. + final SendOptions sendOptions = new SendOptions().setPartitionId(nonEmptyPartition.getId()); + producer.send(new EventData("Hello world!" .getBytes(UTF_8)), sendOptions); - producer.send(new EventData("Hello world!".getBytes(UTF_8)), sendOptions).block(OPERATION_TIMEOUT); - // Acquiring the semaphore so that this sample does not end before all events are fetched. - semaphore.acquire(); + // Set isRunning to false so we stop taking events. + isRunning.set(false); // Dispose and close of all the resources we've created. - subscription.dispose(); producer.close(); consumer.close(); - earliestConsumer.close(); } } From 055b70c7a77fb6726a16095affe613277e834c54 Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 16:36:04 -0800 Subject: [PATCH 10/17] Fix indentation. --- .../PublishEventsWithSizeLimitedBatches.java | 31 +++++++++---------- 1 file changed, 15 insertions(+), 16 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java index 444a7f80165a..1627e6f0715d 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java @@ -50,23 +50,22 @@ public static void main(String[] args) { // The sample Flux contains three events, but it could be an infinite stream of telemetry events. telemetryEvents.subscribe(event -> { - final EventDataBatch batch = currentBatch.get(); - if (!batch.tryAdd(event)) { - producer.createBatch(options).map(newBatch -> { - currentBatch.set(newBatch); + final EventDataBatch batch = currentBatch.get(); + if (!batch.tryAdd(event)) { + producer.createBatch(options).map(newBatch -> { + currentBatch.set(newBatch); - // Adding that event we couldn't add before. - newBatch.tryAdd(event); - return producer.send(batch); - }).block(); - } - }, error -> System.err.println("Error received:" + error), - () -> { - final EventDataBatch batch = currentBatch.getAndSet(null); - if (batch != null) { - producer.send(batch).block(); - } - }); + // Adding that event we couldn't add before. + newBatch.tryAdd(event); + return producer.send(batch); + }).block(); + } + }, error -> System.err.println("Error received:" + error), () -> { + final EventDataBatch batch = currentBatch.getAndSet(null); + if (batch != null) { + producer.send(batch).block(); + } + }); // Sleeping this thread because we want to wait for all the events to send before ending the program. // `.subscribe` is not a blocking call. It coordinates the callbacks and starts the send operation, but does not From 1a53cc905cd45f8b0b1340fbaaf1bcb6be488dc5 Mon Sep 17 00:00:00 2001 From: Connie Yau Date: Mon, 25 Nov 2019 20:22:37 -0800 Subject: [PATCH 11/17] Update sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerClient.java Co-Authored-By: Srikanta <51379715+srnagar@users.noreply.github.com> --- .../com/azure/messaging/eventhubs/EventHubConsumerClient.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerClient.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerClient.java index f0b07f50feb6..4cb9344faf3d 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerClient.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerClient.java @@ -22,7 +22,7 @@ import java.util.concurrent.atomic.AtomicInteger; /** - * A synchronous consumer responsible for reading {@link EventData} from a Event Hub partition in the context of + * A synchronous consumer responsible for reading {@link EventData} from an Event Hub partition in the context of * a specific consumer group. * *

Creating a synchronous consumer

From be0f88e014b4586f9f30dc9f3932bb9d749e4d9f Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 20:43:28 -0800 Subject: [PATCH 12/17] Fix checkstyle issues. --- .../eventhubs/PublishEventsWithSizeLimitedBatches.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java index 1627e6f0715d..38256c7bafce 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java @@ -61,11 +61,11 @@ public static void main(String[] args) { }).block(); } }, error -> System.err.println("Error received:" + error), () -> { - final EventDataBatch batch = currentBatch.getAndSet(null); - if (batch != null) { - producer.send(batch).block(); - } - }); + final EventDataBatch batch = currentBatch.getAndSet(null); + if (batch != null) { + producer.send(batch).block(); + } + }); // Sleeping this thread because we want to wait for all the events to send before ending the program. // `.subscribe` is not a blocking call. It coordinates the callbacks and starts the send operation, but does not From a7d86c4e56916842a2f6696770782a7f419a869e Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 22:18:42 -0800 Subject: [PATCH 13/17] (WIP) --- .../eventhubs/PublishEventsWithCustomMetadata.java | 10 ++++++++-- .../eventhubs/PublishEventsWithSizeLimitedBatches.java | 8 +++++++- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java index 5e2ddaa02be0..3dc588b9583e 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java @@ -2,6 +2,7 @@ // Licensed under the MIT License. package com.azure.messaging.eventhubs; +import reactor.core.Exceptions; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -76,8 +77,13 @@ public static void main(String[] args) { producer.createBatch().map(newBatch -> { currentBatch.set(newBatch); - // Add that event that we couldn't before. - newBatch.tryAdd(event); + // Adding that event we couldn't add before. + if (!newBatch.tryAdd(event)) { + throw Exceptions.propagate(new IllegalArgumentException( + String.format("Could not add event because it is too large. Contents: %s", + event.getBodyAsString()))); + } + return newBatch; })); }).then() diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java index 38256c7bafce..7dd7dc52af44 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java @@ -4,6 +4,7 @@ package com.azure.messaging.eventhubs; import com.azure.messaging.eventhubs.models.CreateBatchOptions; +import reactor.core.Exceptions; import reactor.core.publisher.Flux; import java.util.concurrent.atomic.AtomicReference; @@ -56,7 +57,12 @@ public static void main(String[] args) { currentBatch.set(newBatch); // Adding that event we couldn't add before. - newBatch.tryAdd(event); + if (!newBatch.tryAdd(event)) { + throw Exceptions.propagate(new IllegalArgumentException( + String.format("Could not add event because it is too large. Contents: %s", + event.getBodyAsString()))); + } + return producer.send(batch); }).block(); } From ca3333b8a993e0276984e09f18ea490c9f9ea93f Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 22:31:10 -0800 Subject: [PATCH 14/17] Update samples. --- .../PublishEventsToSpecificPartition.java | 14 ++-- .../PublishEventsWithAzureIdentity.java | 49 ++++++++----- .../PublishEventsWithCustomMetadata.java | 10 +-- .../PublishEventsWithPartitionKey.java | 69 +++++++++++++------ .../PublishEventsWithSizeLimitedBatches.java | 53 ++++++++------ 5 files changed, 124 insertions(+), 71 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java index 550c3929d7c3..0c989564be42 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java @@ -3,6 +3,7 @@ package com.azure.messaging.eventhubs; import com.azure.messaging.eventhubs.models.CreateBatchOptions; +import reactor.core.Exceptions; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -19,7 +20,7 @@ public class PublishEventsToSpecificPartition { private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); /** - * Main method to invoke this demo about how to send a batch of events with partition ID configured. + * Main method to invoke this demo about how to send a batch of events with partition id configured. * * @param args Unused arguments to the program. */ @@ -72,14 +73,19 @@ public static void main(String[] args) { currentBatch.set(newBatch); // Add that event that we couldn't before. - newBatch.tryAdd(event); + if (!newBatch.tryAdd(event)) { + throw Exceptions.propagate(new IllegalArgumentException(String.format( + "Event is too large for an empty batch. Max size: %s. Event: %s", + newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); + }; + return newBatch; })); }).then() .doFinally(signal -> { final EventDataBatch batch = currentBatch.getAndSet(null); if (batch != null) { - producer.send(batch).block(Duration.ofSeconds(60)); + producer.send(batch).block(OPERATION_TIMEOUT); } }); @@ -87,7 +93,7 @@ public static void main(String[] args) { // subscriber to that operation. For the purpose of this example, we block so the program does not end before // the send operation is complete. Any of the `.subscribe` overloads also work to start the Mono asynchronously. try { - sendOperation.block(Duration.ofSeconds(60)); + sendOperation.block(OPERATION_TIMEOUT); } finally { // Disposing of our producer. producer.close(); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java index 11d6a8068131..58edede900fd 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java @@ -4,8 +4,11 @@ import com.azure.core.credential.TokenCredential; import com.azure.identity.DefaultAzureCredentialBuilder; +import reactor.core.Exceptions; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import java.time.Duration; import java.util.concurrent.atomic.AtomicReference; import static java.nio.charset.StandardCharsets.UTF_8; @@ -14,6 +17,8 @@ * Sample demonstrates how to send an {@link EventDataBatch} to an Azure Event Hub using Azure Identity. */ public class PublishEventsWithAzureIdentity { + private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); + /** * Main method to invoke this demo on how to send an {@link EventDataBatch} to an Azure Event Hub. * @@ -52,35 +57,41 @@ public static void main(String[] args) { producer.createBatch().block()); // The sample Flux contains three events, but it could be an infinite stream of telemetry events. - telemetryEvents.subscribe(event -> { + final Mono sendOperation = telemetryEvents.flatMap(event -> { final EventDataBatch batch = currentBatch.get(); - if (!batch.tryAdd(event)) { + if (batch.tryAdd(event)) { + return Mono.empty(); + } + + // The batch is full, so we create a new batch and send the batch. Mono.when completes when both operations + // have completed. + return Mono.when( + producer.send(batch), producer.createBatch().map(newBatch -> { currentBatch.set(newBatch); - // Adding that event we couldn't add before. - newBatch.tryAdd(event); - return producer.send(batch); - }).block(); - } - }, error -> System.err.println("Error received:" + error), - () -> { + // Add that event that we couldn't before. + if (!newBatch.tryAdd(event)) { + throw Exceptions.propagate(new IllegalArgumentException(String.format( + "Event is too large for an empty batch. Max size: %s. Event: %s", + newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); + }; + + return newBatch; + })); + }).then() + .doFinally(signal -> { final EventDataBatch batch = currentBatch.getAndSet(null); if (batch != null) { - producer.send(batch).block(); + producer.send(batch).block(OPERATION_TIMEOUT); } }); - // Sleeping this thread because we want to wait for all the events to send before ending the program. - // `.subscribe` is not a blocking call. It coordinates the callbacks and starts the send operation, but does not - // wait for it to complete. - // Customers can chain together Reactor operators like .then() and `.doFinally` if they want an operation to run - // after all the events have been transmitted. .block() can also be used to turn the call into a synchronous - // operation. + // The sendOperation creation and assignment is not a blocking call. It does not get invoked until there is a + // subscriber to that operation. For the purpose of this example, we block so the program does not end before + // the send operation is complete. Any of the `.subscribe` overloads also work to start the Mono asynchronously. try { - Thread.sleep(5000); - } catch (InterruptedException e) { - e.printStackTrace(); + sendOperation.block(OPERATION_TIMEOUT); } finally { // Disposing of our producer. producer.close(); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java index 3dc588b9583e..fd2e932cc62b 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java @@ -77,12 +77,12 @@ public static void main(String[] args) { producer.createBatch().map(newBatch -> { currentBatch.set(newBatch); - // Adding that event we couldn't add before. + // Add that event that we couldn't before. if (!newBatch.tryAdd(event)) { - throw Exceptions.propagate(new IllegalArgumentException( - String.format("Could not add event because it is too large. Contents: %s", - event.getBodyAsString()))); - } + throw Exceptions.propagate(new IllegalArgumentException(String.format( + "Event is too large for an empty batch. Max size: %s. Event: %s", + newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); + }; return newBatch; })); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java index 006b388329b6..e6b20887d538 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java @@ -3,8 +3,11 @@ package com.azure.messaging.eventhubs; import com.azure.messaging.eventhubs.models.CreateBatchOptions; +import reactor.core.Exceptions; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import java.time.Duration; import java.util.concurrent.atomic.AtomicReference; import static java.nio.charset.StandardCharsets.UTF_8; @@ -13,10 +16,11 @@ * Send a Flux of events using a partition key. */ public class PublishEventsWithPartitionKey { + private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); /** - * Main method to invoke this demo about how to send a list of events with partition ID configured in send option - * to an Azure Event Hub instance. + * Main method to invoke this demo about how to send a list of events with partition key configured in + * CreateBatchOptions to an Azure Event Hub instance. * * @param args Unused arguments to the program. */ @@ -43,15 +47,17 @@ public static void main(String[] args) { // the Event Hubs service keep different batches of events together on the same partition. This can be // accomplished by setting a partition key when publishing the events. // - // The partition key is NOT the identifier of a specific partition. Rather, it is an arbitrary piece of string data - // that Event Hubs uses as the basis to compute a hash value. Event Hubs will associate the hash value with a specific - // partition, ensuring that any events published with the same partition key are rerouted to the same partition. + // The partition key is NOT the identifier of a specific partition. Rather, it is an arbitrary piece of string + // data that Event Hubs uses as the basis to compute a hash value. Event Hubs will associate the hash value with + // a specific partition, ensuring that any events published with the same partition key are rerouted to the same + // partition. // - // All of event data send to the same partition of the partition key 'basketball' associate with. + // All the event data with partition key 'basketball' end up in the same partition. // - // Note that there is no means of accurately predicting which partition will be associated with a given partition key; - // we can only be assured that it will be a consistent choice of partition. If you have a need to understand which - // exact partition an event is published to, you will need to use an Event Hub producer associated with that partition. + // Note that there is no means of accurately predicting which partition will be associated with a given + // partition key; we can only be assured that it will be a consistent choice of partition. If you have a need to + // understand which exact partition an event is published to, you will need to use + // CreateBatchOptions.setPartitionId(String) when creating the EventDataBatch. final CreateBatchOptions options = new CreateBatchOptions() .setPartitionKey("basketball"); final AtomicReference currentBatch = new AtomicReference<>( @@ -60,23 +66,44 @@ public static void main(String[] args) { // We try to add as many events as a batch can fit based on the event size and send to Event Hub when // the batch can hold no more events. Create a new batch for next set of events and repeat until all events // are sent. - data.subscribe(event -> { + final Mono sendOperation = data.flatMap(event -> { final EventDataBatch batch = currentBatch.get(); - if (!batch.tryAdd(event)) { - producer.createBatch(options).map(newBatch -> { - currentBatch.set(newBatch); - return producer.send(batch); - }).block(); + if (batch.tryAdd(event)) { + return Mono.empty(); } - }, error -> System.err.println("Error received:" + error), - () -> { + + // The batch is full, so we create a new batch and send the batch. Mono.when completes when both operations + // have completed. + return Mono.when( + producer.send(batch), + producer.createBatch().map(newBatch -> { + currentBatch.set(newBatch); + + // Add that event that we couldn't before. + if (!newBatch.tryAdd(event)) { + throw Exceptions.propagate(new IllegalArgumentException(String.format( + "Event is too large for an empty batch. Max size: %s. Event: %s", + newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); + }; + + return newBatch; + })); + }).then() + .doFinally(signal -> { final EventDataBatch batch = currentBatch.getAndSet(null); if (batch != null) { - producer.send(batch).block(); + producer.send(batch).block(OPERATION_TIMEOUT); } - - // Disposing of our producer. - producer.close(); }); + + // The sendOperation creation and assignment is not a blocking call. It does not get invoked until there is a + // subscriber to that operation. For the purpose of this example, we block so the program does not end before + // the send operation is complete. Any of the `.subscribe` overloads also work to start the Mono asynchronously. + try { + sendOperation.block(OPERATION_TIMEOUT); + } finally { + // Disposing of our producer. + producer.close(); + } } } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java index 7dd7dc52af44..fb349162594a 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java @@ -6,7 +6,9 @@ import com.azure.messaging.eventhubs.models.CreateBatchOptions; import reactor.core.Exceptions; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import java.time.Duration; import java.util.concurrent.atomic.AtomicReference; import static java.nio.charset.StandardCharsets.UTF_8; @@ -16,6 +18,8 @@ * {@link CreateBatchOptions#setMaximumSizeInBytes(int)}. */ public class PublishEventsWithSizeLimitedBatches { + private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); + /** * Main method to invoke this demo on how to send an {@link EventDataBatch} to an Azure Event Hub. * @@ -50,39 +54,44 @@ public static void main(String[] args) { producer.createBatch(options).block()); // The sample Flux contains three events, but it could be an infinite stream of telemetry events. - telemetryEvents.subscribe(event -> { + // We try to add as many events as a batch can fit based on the event size and send to Event Hub when + // the batch can hold no more events. Create a new batch for next set of events and repeat until all events + // are sent. + final Mono sendOperation = telemetryEvents.flatMap(event -> { final EventDataBatch batch = currentBatch.get(); - if (!batch.tryAdd(event)) { - producer.createBatch(options).map(newBatch -> { + if (batch.tryAdd(event)) { + return Mono.empty(); + } + + // The batch is full, so we create a new batch and send the batch. Mono.when completes when both operations + // have completed. + return Mono.when( + producer.send(batch), + producer.createBatch().map(newBatch -> { currentBatch.set(newBatch); - // Adding that event we couldn't add before. + // Add that event that we couldn't before. if (!newBatch.tryAdd(event)) { - throw Exceptions.propagate(new IllegalArgumentException( - String.format("Could not add event because it is too large. Contents: %s", - event.getBodyAsString()))); - } + throw Exceptions.propagate(new IllegalArgumentException(String.format( + "Event is too large for an empty batch. Max size: %s. Event: %s", + newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); + }; - return producer.send(batch); - }).block(); - } - }, error -> System.err.println("Error received:" + error), () -> { + return newBatch; + })); + }).then() + .doFinally(signal -> { final EventDataBatch batch = currentBatch.getAndSet(null); if (batch != null) { - producer.send(batch).block(); + producer.send(batch).block(OPERATION_TIMEOUT); } }); - // Sleeping this thread because we want to wait for all the events to send before ending the program. - // `.subscribe` is not a blocking call. It coordinates the callbacks and starts the send operation, but does not - // wait for it to complete. - // Customers can chain together Reactor operators like .then() and `.doFinally` if they want an operation to run - // after all the events have been transmitted. .block() can also be used to turn the call into a synchronous - // operation. + // The sendOperation creation and assignment is not a blocking call. It does not get invoked until there is a + // subscriber to that operation. For the purpose of this example, we block so the program does not end before + // the send operation is complete. Any of the `.subscribe` overloads also work to start the Mono asynchronously. try { - Thread.sleep(5000); - } catch (InterruptedException e) { - e.printStackTrace(); + sendOperation.block(OPERATION_TIMEOUT); } finally { // Disposing of our producer. producer.close(); From fdf4f62215bae538c896acd7957d0c8eb4449ade Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 22:41:26 -0800 Subject: [PATCH 15/17] Fix checkstyle issues. --- .../com/azure/messaging/eventhubs/EventHubProperties.java | 4 ++-- .../messaging/eventhubs/PublishEventsToSpecificPartition.java | 2 +- .../messaging/eventhubs/PublishEventsWithAzureIdentity.java | 2 +- .../messaging/eventhubs/PublishEventsWithCustomMetadata.java | 2 +- .../messaging/eventhubs/PublishEventsWithPartitionKey.java | 2 +- .../eventhubs/PublishEventsWithSizeLimitedBatches.java | 2 +- 6 files changed, 7 insertions(+), 7 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 97c64513a32c..15fe0fb2b2e3 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 @@ -41,7 +41,7 @@ public final class EventHubProperties { } /** - * Gets the Event Hub name + * Gets the name of the Event Hub. * * @return Name of the Event Hub. */ @@ -50,7 +50,7 @@ public String getName() { } /** - * Gets the instant, in UTC, at which Event Hub was created at. + * Gets the instant, in UTC, at which Event Hub was created. * * @return The instant, in UTC, at which the Event Hub was created. */ diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java index 0c989564be42..19ef651f105b 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java @@ -77,7 +77,7 @@ public static void main(String[] args) { throw Exceptions.propagate(new IllegalArgumentException(String.format( "Event is too large for an empty batch. Max size: %s. Event: %s", newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); - }; + } return newBatch; })); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java index 58edede900fd..c89383280c20 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java @@ -75,7 +75,7 @@ public static void main(String[] args) { throw Exceptions.propagate(new IllegalArgumentException(String.format( "Event is too large for an empty batch. Max size: %s. Event: %s", newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); - }; + } return newBatch; })); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java index fd2e932cc62b..3da7d8128637 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java @@ -82,7 +82,7 @@ public static void main(String[] args) { throw Exceptions.propagate(new IllegalArgumentException(String.format( "Event is too large for an empty batch. Max size: %s. Event: %s", newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); - }; + } return newBatch; })); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java index e6b20887d538..2496e26e06f3 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java @@ -84,7 +84,7 @@ public static void main(String[] args) { throw Exceptions.propagate(new IllegalArgumentException(String.format( "Event is too large for an empty batch. Max size: %s. Event: %s", newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); - }; + } return newBatch; })); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java index fb349162594a..21544154c737 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java @@ -75,7 +75,7 @@ public static void main(String[] args) { throw Exceptions.propagate(new IllegalArgumentException(String.format( "Event is too large for an empty batch. Max size: %s. Event: %s", newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); - }; + } return newBatch; })); From 88585ff5d86c9ee9fc966ca2f4e873bb16399a79 Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 23:38:16 -0800 Subject: [PATCH 16/17] Update README. --- .../azure-messaging-eventhubs/README.md | 219 +++++++++--------- 1 file changed, 115 insertions(+), 104 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/README.md b/sdk/eventhubs/azure-messaging-eventhubs/README.md index 02dd87c2f5e0..8cd5b45b0ff7 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/README.md +++ b/sdk/eventhubs/azure-messaging-eventhubs/README.md @@ -23,17 +23,17 @@ documentation][event_hubs_product_docs] | [Samples][sample_examples] - [Table of contents](#table-of-contents) - [Getting started](#getting-started) - - [Default SSL](#default-ssl-library) - [Prerequisites](#prerequisites) - [Adding the package to your product](#adding-the-package-to-your-product) + - [Default SSL library](#default-ssl-library) - [Methods to authorize with Event Hubs](#methods-to-authorize-with-event-hubs) - - [Create an Event Hub client using a connection string](#create-an-event-hub-client-using-a-connection-string) + - [Create an Event Hub producer using a connection string](#create-an-event-hub-producer-using-a-connection-string) - [Create an Event Hub client using Microsoft identity platform (formerly Azure Active Directory)](#create-an-event-hub-client-using-microsoft-identity-platform-formerly-azure-active-directory) - [Key concepts](#key-concepts) - [Examples](#examples) - [Publish events to an Event Hub](#publish-events-to-an-event-hub) - [Consume events from an Event Hub partition](#consume-events-from-an-event-hub-partition) - - [Consume events using an Event Processor](#consume-events-using-an-event-processor) + - [Consume events using an EventProcessorClient](#consume-events-using-an-eventprocessorclient) - [Troubleshooting](#troubleshooting) - [Enable client logging](#enable-client-logging) - [Enable AMQP transport logging](#enable-amqp-transport-logging) @@ -44,17 +44,14 @@ documentation][event_hubs_product_docs] | [Samples][sample_examples] ## Getting started -### Default SSL library -All client libraries, by default, use the Tomcat-native Boring SSL library to enable native-level performance for SSL operations. The Boring SSL library is an uber jar containing native libraries for Linux / macOS / Windows, and provides better performance compared to the default SSL implementation within the JDK. For more information, including how to reduce the dependency size, refer to the [performance tuning][performance_tuning] section of the wiki. - ### Prerequisites - Java Development Kit (JDK) with version 8 or above - [Maven][maven] - Microsoft Azure subscription - - You can create a free account at: https://azure.microsoft.com + - You can create a free account at: https://azure.microsoft.com - Azure Event Hubs instance - - Step-by-step guide for [creating an Event Hub using the Azure Portal][event_hubs_create] + - Step-by-step guide for [creating an Event Hub using the Azure Portal][event_hubs_create] ### Adding the package to your product @@ -68,6 +65,13 @@ All client libraries, by default, use the Tomcat-native Boring SSL library to en ``` [//]: # ({x-version-update-end}) +### Default SSL library +All client libraries, by default, use the Tomcat-native Boring SSL library to enable native-level +performance for SSL operations. The Boring SSL library is an uber jar containing native libraries +for Linux/macOS/Windows, and provides better performance compared to the default SSL +implementation within the JDK. For more information, including how to reduce the dependency size, +refer to the [performance tuning][performance_tuning] section of the wiki. + ### Methods to authorize with Event Hubs For the Event Hubs client library to interact with an Event Hub, it will need to understand how to connect and authorize @@ -79,9 +83,10 @@ The easiest means for doing so is to use a connection string, which is created a namespace. If you aren't familiar with shared access policies in Azure, you may wish to follow the step-by-step guide to [get an Event Hubs connection string][event_hubs_connection_string]. -Both the asynchronous and synchronous Event Hub producer and consumer clients can be created using -`EventHubClientBuilder`. Invoking `.buildAsyncProducer()` and `buildProducer` will build the asynchronous and -synchronous producers. Similarly, `.buildAsyncConsumer` and `.buildConsumer` will build the appropriate consumers. +Both the asynchronous and synchronous Event Hub producer and consumer clients can be created using +`EventHubClientBuilder`. Invoking `buildAsyncProducerClient()` or `buildProducerClient()` will build the asynchronous or +synchronous producers. Similarly, `buildAsyncConsumerClient()` or `buildConsumerClient()` will build the appropriate +consumers. The snippet below creates a synchronous Event Hub producer. @@ -90,7 +95,7 @@ String connectionString = "<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>" String eventHubName = "<< NAME OF THE EVENT HUB >>"; EventHubProducerClient producer = new EventHubClientBuilder() .connectionString(connectionString, eventHubName) - .buildProducer(); + .buildProducerClient(); ``` ### Create an Event Hub client using Microsoft identity platform (formerly Azure Active Directory) @@ -127,11 +132,11 @@ ClientSecretCredential credential = new ClientSecretCredentialBuilder() // The fully qualified namespace for the Event Hubs instance. This is likely to be similar to: // {your-namespace}.servicebus.windows.net -String fullyQualifiedNamespace = "<< EVENT HUBS FULLY QUALIFIED NAMESPACE >>" +String fullyQualifiedNamespace = "my-test-eventhubs.servicebus.windows.net"; String eventHubName = "<< NAME OF THE EVENT HUB >>"; EventHubProducerClient client = new EventHubClientBuilder() .credential(fullyQualifiedNamespace, eventHubName, credential) - .buildProducer(); + .buildProducerClient(); ``` ## Key concepts @@ -162,38 +167,33 @@ are well documented in [OASIS Advanced Messaging Queuing Protocol (AMQP) Version ## Examples -- [Inspect Event Hub and partition properties][sample_get_event_hubs_metadata] -- [Publish an EventDataBatch to an Event Hub][sample_publish_eventdatabatch] -- [Consume events from an Event Hub partition][sample_consume_event] -- [Consume events from all Event Hub partitions][sample_event_processor] - ### Publish events to an Event Hub -In order to publish events, you'll need to create an asynchronous -[`EventHubProducerAsyncClient`][source_eventhubasyncproducerclient] or a synchronous -[`EventHubProducerClient`][source_eventHubProducerClient]. -Each producers can send events to either, a specific partition, or allow the Event Hubs service to decide which -partition events should be published to. It is recommended to use automatic routing when the publishing of events needs -to be highly available or when event data should be distributed evenly among the partitions. In the our example, we will -take advantage of automatic routing. +To publish events, you'll need to create an asynchronous [`EventHubProducerAsyncClient`][EventHubProducerAsyncClient] or +a synchronous [`EventHubProducerClient`][EventHubProducerClient]. Each producer can send events to either, a specific +partition, or allow the Event Hubs service to decide which partition events should be published to. It is recommended to +use automatic routing when the publishing of events needs to be highly available or when event data should be +distributed evenly among the partitions. -#### Event Hub producer creation +#### Create an Event Hub producer and publish events -Developers can create a producer by calling `buildProducer()` or `buildAsyncProducer`. If `buildProducer` is invoked, a -synchronous `EventHubProducerClient` is created. If `buildAsyncProducer` is used, an asynchronous -`EventHubProducerAsyncClient` is returned. +Developers can create a producer by calling `buildAsyncProducerClient()` or `buildProducerClient()`. If +`buildProducerClient()` is invoked, a synchronous `EventHubProducerClient` is created. If `buildAsyncProducerClient()` +is used, an asynchronous `EventHubProducerAsyncClient` is returned. -Specifying `batchOptions.partitionId(String)` will send events to a specific partition, and not, will allow for automatic -routing. In addition, specifying `partitionKey(String)` will tell Event Hubs service to hash the events and send them to -the same partition. +Specifying `CreateBatchOptions.setPartitionId(String)` will send events to a specific partition. If not specified, will +allow for automatic partition routing. In addition, specifying `CreateBatchOptions.setPartitionKey(String)` will tell +Event Hubs service to hash the events and send them to the same partition. -The snippet below creates a synchronous producer and sends events to any partition, allowing Event Hubs service to route the event -to an available partition. +The snippet below creates a synchronous producer and sends events to any partition, allowing Event Hubs service to route +the event to an available partition. ```java +List eventDataList = Arrays.asList(new EventData("Foo"), new EventData("Bar")); + EventHubProducerClient producer = new EventHubClientBuilder() .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") - .buildProducer(); + .buildProducerClient(); EventDataBatch eventDataBatch = producer.createBatch(); for (EventData eventData : eventDataList) { if (!eventDataBatch.tryAdd(eventData)) { @@ -203,100 +203,108 @@ for (EventData eventData : eventDataList) { } // send the last batch of remaining events -if (eventDataBatch.getSize() > 0) { +if (eventDataBatch.getCount() > 0) { producer.send(eventDataBatch); } ``` -To send events to a particular partition, set the optional parameter `partitionId` on -[`CreateBatchOptions`][source_CreateBatchOptions]. +To send events to a particular partition, set the optional parameter `setPartitionId(String)` on +[`CreateBatchOptions`][CreateBatchOptions]. -#### Partition identifier +#### Publish events using partition identifier Many Event Hub operations take place within the scope of a specific partition. Because partitions are owned by the Event -Hub, their names are assigned at the time of creation. To understand what partitions are available, You can use the -`getPartitionIds` function to get the ids of all available partitions in your Event Hub instance. +Hub, their names are assigned at the time of creation. To understand what partitions are available, you can use the +`getPartitionIds()` function to get the ids of all available partitions in your Event Hub instance. All clients created +using `EventHubsClientBuilder` can query for metadata about the Event Hub using `getPartitionIds()` or +`getEventHubProperties()`. ```java -IterableStream partitionIds = client.getPartitionIds(); +EventHubProducerClient producer = new EventHubClientBuilder() + .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") + .buildProducerClient(); + +CreateBatchOptions options = new CreateBatchOptions().setPartitionId("0"); +EventDataBatch batch = producer.createBatch(options); + +// Add events to batch and when you want to send the batch, send it using the producer. +producer.send(batch); ``` -#### Partition key +#### Publish events using partition key When an Event Hub producer is not associated with any specific partition, it may be desirable to request that the Event Hubs service keep different events or batches of events together on the same partition. This can be accomplished by setting a `partition key` when publishing the events. ```java -CreateBatchOptions batchOptions = new CreateBatchOptions() - .partitionKey("grouping-key"); +CreateBatchOptions batchOptions = new CreateBatchOptions().setPartitionKey("grouping-key"); EventDataBatch eventDataBatch = producer.createBatch(batchOptions); -// add events to eventDataBatch + +// Add events to batch and when you want to send the batch, send it using the producer. producer.send(eventDataBatch); ``` ### Consume events from an Event Hub partition -In order to consume events, you'll need to create an `EventHubAsyncConsumer` or `EventHubConsumer` for a specific -partition and consumer group combination. When an Event Hub is created, it starts with a default consumer group that can -be used to get started. A consumer also needs to specify where in the event stream to begin receiving events; in our -example, we will focus on reading new events as they are published. +In order to consume events, you'll need to create an [`EventHubConsumerAsyncClient`][EventHubConsumerAsyncClient] or +[`EventHubConsumerClient`][EventHubConsumerClient] for a specific consumer group. When an Event Hub is created, it +starts with a default consumer group that can be used to get started. A consumer also needs to specify where in the +event stream to begin receiving events. -#### Consume events with EventHubAsyncConsumer +#### Consume events with EventHubConsumerAsyncClient In the snippet below, we are creating an asynchronous consumer that receives events from `partitionId` and only listens -to newest events that get pushed to the partition using `receive(String partitionId)`. Developers can begin receiving -events by calling `.receive(String partitionId)` and subscribing to the stream. +to newest events that get pushed to the partition by invoking `receiveFromPartition(String, EventPosition)`. Developers +can begin receiving events from multiple partitions using the same EventHubConsumerAsyncClient by calling +`receiveFromPartition(String, EventPosition)` with another partition id, and subscribing to that Flux. ```java -EventHubAsyncConsumer client = new EventHubClientBuilder() +EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") - .consumerGroup(EventHubAsyncClient.DEFAULT_CONSUMER_GROUP_NAME) - .startingPosition(EventPosition.earliest()) - .buildAsyncConsumer(); + .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .buildAsyncConsumerClient(); -consumer.receive(partitionId).subscribe(event -> { +// Receive events from partition with id "0", only getting events that are newly added to the partition. +consumer.receiveFromPartition("0", EventPosition.latest()).subscribe(event -> { // Process each event as it arrives. }); ``` -#### Consume events with EventHubConsumer +#### Consume events with EventHubConsumerClient -Developers can create a synchronous consumer that returns events in batches using an `EventHubAsyncConsumer`. In the +Developers can create a synchronous consumer that returns events in batches using an `EventHubConsumerClient`. In the snippet below, a consumer is created that starts reading events from the beginning of the partition's event stream. ```java -EventHubConsumer client = new EventHubClientBuilder() +EventHubConsumerClient consumer = new EventHubClientBuilder() .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") - .consumerGroup(EventHubAsyncClient.DEFAULT_CONSUMER_GROUP_NAME) - .startingPosition(EventPosition.earliest()) - .buildConsumer(); + .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .buildConsumerClient(); String partitionId = "<< EVENT HUB PARTITION ID >>"; // Get the first 15 events in the stream, or as many events as can be received within 40 seconds. -IterableStream events = consumer.receive(partitionId, 15, Duration.ofSeconds(40)); -for (EventData event : events) { - // Process each event +IterableStream events = consumer.receiveFromPartition(partitionId, 15, + EventPosition.earliest(), Duration.ofSeconds(40)); +for (PartitionEvent event : events) { + System.out.println("Event: " + event.getData().getBodyAsString()); } - -// Calling receive again returns the next 15 events in the stream, or as many as possible in 40 seconds. -IterableStream nextEvents = consumer.receive(partitionId, 15, Duration.ofSeconds(40)); ``` -### Consume events using an Event Processor Client +### Consume events using an EventProcessorClient -To consume events for all partitions of an Event Hub, you'll create an [`EventProcessorClient`][source_eventprocessorclient] for a -specific consumer group. When an Event Hub is created, it provides a default consumer group that can be used to get -started. +To consume events for all partitions of an Event Hub, you'll create an [`EventProcessorClient`][EventProcessorClient] +for a specific consumer group. When an Event Hub is created, it provides a default consumer group that can be used to +get started. -The [`EventProcessorClient`][source_eventprocessorclient] will delegate processing of events to a callback function that you -provide, allowing you to focus on the logic needed to provide value while the processor holds responsibility for +The [`EventProcessorClient`][EventProcessorClient] will delegate processing of events to a callback function that you +provide, allowing you to focus on the logic needed to provide value while the processor holds responsibility for managing the underlying consumer operations. -In our example, we will focus on building the [`EventProcessorClient`][source_eventprocessorclient], use the -[`InMemoryCheckpointStore`][source_inmemorycheckpointstore] available in samples, and a callback function that -processes events received from the Event Hub and writes to console. +In our example, we will focus on building the [`EventProcessorClient`][EventProcessorClient], use the +[`InMemoryCheckpointStore`][InMemoryCheckpointStore] available in samples, and a callback function that processes events +received from the Event Hub and writes to console. ```java class Program { @@ -316,9 +324,9 @@ class Program { // This will start the processor. It will start processing events from all partitions. eventProcessorClient.start(); - + // (for demo purposes only - adding sleep to wait for receiving events) - TimeUnit.SECONDS.sleep(2); + TimeUnit.SECONDS.sleep(2); // When the user wishes to stop processing events, they can call `stop()`. eventProcessorClient.stop(); @@ -332,7 +340,7 @@ class Program { You can set the `AZURE_LOG_LEVEL` environment variable to view logging statements made in the client library. For example, setting `AZURE_LOG_LEVEL=2` would show all informational, warning, and error log messages. The log levels can -be found here: [log levels][source_loglevels]. +be found here: [log levels][LogLevels]. ### Enable AMQP transport logging @@ -364,17 +372,17 @@ This is a general exception for AMQP related failures, which includes the AMQP e that caused this exception as ErrorContext. 'isTransient' is a boolean indicating if the exception is a transient error or not. If true, then the request can be retried; otherwise not. -The [ErrorCondition][source_errorcondition] contains error conditions common to the AMQP protocol and used by Azure +[`AmqpErrorCondition`][AmqpErrorCondition] contains error conditions common to the AMQP protocol and used by Azure services. When an AMQP exception is thrown, examining the error condition field can inform developers as to why the AMQP exception occurred and if possible, how to mitigate this exception. A list of all the AMQP exceptions can be found in [OASIS AMQP Version 1.0 Transport Errors][oasis_amqp_v1_error]. -The [ErrorContext][source_errorcontext] in the [AmqpException][source_amqpexception] provides information about the AMQP +The [`AmqpErrorContext`][AmqpErrorContext] in the [`AmqpException`][AmqpException] provides information about the AMQP session, link, or connection that the exception occurred in. This is useful to diagnose which level in the transport this exception occurred at and whether it was an issue in one of the producers or consumers. -The recommended way to solve the specific exception the AMQP exception represents is to follow the [Event Hubs Messaging -Exceptions][event_hubs_messaging_exceptions] guidance. +The recommended way to solve the specific exception the AMQP exception represents is to follow the +[Event Hubs Messaging Exceptions][event_hubs_messaging_exceptions] guidance. #### Operation cancelled exception @@ -400,12 +408,14 @@ advantage of the full feature set of the Azure Event Hubs service. In order to h the following set of sample is available: - [Inspect Event Hub and partition properties][sample_get_event_hubs_metadata] -- [Publish an event batch to an Event Hub][sample_publish_eventdatabatch] +- [Publish events using Microsoft identity platform][sample_publish_identity] - [Publish events to a specific Event Hub partition with partition identifier][sample_publish_partitionId] - [Publish events to a specific Event Hub partition with partition key][sample_publish_partitionKey] +- [Publish events to an Event Hub with a size-limited batch][sample_publish_size_limited] - [Publish events with custom metadata][sample_publish_custom_metadata] - [Consume events from an Event Hub partition][sample_consume_event] -- [Save the last read event and resume from that point][sample_sequence_number] +- [Consume events starting from an event sequence number][sample_consume_sequence_number] +- [Consume events from all partitions using EventProcessorClient][sample_event_processor] ## Contributing @@ -429,26 +439,27 @@ Guidelines](./CONTRIBUTING.md) for more information. [oasis_amqp_v1]: http://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-overview-v1.0-os.html [performance_tuning]: https://github.com/Azure/azure-sdk-for-java/wiki/Performance-Tuning [qpid_proton_j_apache]: http://qpid.apache.org/proton/ -[sample_consume_event]: ./src/samples/java/com/azure/messaging/eventhubs/ConsumeEvent.java -[sample_event_processor]: ./src/samples/java/com/azure/messaging/eventhubs/EventProcessorSample.java [sample_examples]: ./src/samples/java/com/azure/messaging/eventhubs/ +[sample_consume_event]: ./src/samples/java/com/azure/messaging/eventhubs/ConsumeEvents.java +[sample_consume_sequence_number]: ./src/samples/java/com/azure/messaging/eventhubs/ConsumeEventsFromKnownSequenceNumberPosition.java +[sample_event_processor]: ./src/samples/java/com/azure/messaging/eventhubs/EventProcessorSample.java [sample_get_event_hubs_metadata]: ./src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java [sample_publish_custom_metadata]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java -[sample_publish_eventdatabatch]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventDataBatch.java +[sample_publish_identity]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java [sample_publish_partitionId]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java [sample_publish_partitionKey]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java -[sample_sequence_number]: ./src/samples/java/com/azure/messaging/eventhubs/ConsumeEventsFromKnownSequenceNumberPosition.java -[source_amqpexception]: ../../core/azure-core-amqp/src/main/java/com/azure/core/amqp/exception/AmqpException.java +[sample_publish_size_limited]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java [source_code]: ./ -[source_errorcondition]: ../../core/azure-core-amqp/src/main/java/com/azure/core/amqp/exception/ErrorCondition.java -[source_errorcontext]: ../../core/azure-core-amqp/src/main/java/com/azure/core/amqp/exception/ErrorContext.java -[source_eventhubasyncclient]: ./src/main/java/com/azure/messaging/eventhubs/EventHubAsyncClient.java -[source_eventhubasyncproducerclient]: ./src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java -[source_eventhubclient]: ./src/main/java/com/azure/messaging/eventhubs/EventHubClient.java -[source_eventHubProducerClient]: ./src/main/java/com/azure/messaging/eventhubs/EventHubProducerClient.java -[source_eventprocessorclient]: ./src/main/java/com/azure/messaging/eventhubs/EventProcessorClient.java -[source_CreateBatchOptions]: ./src/main/java/com/azure/messaging/eventhubs/models/CreateBatchOptions.java -[source_inmemorycheckpointstore]: ./src/samples/java/com/azure/messaging/eventhubs/InMemoryCheckpointStore.java -[source_loglevels]: ../../core/azure-core/src/main/java/com/azure/core/util/logging/ClientLogger.java +[AmqpException]: ../../core/azure-core-amqp/src/main/java/com/azure/core/amqp/exception/AmqpException.java +[AmqpErrorCondition]: ../../core/azure-core-amqp/src/main/java/com/azure/core/amqp/exception/AmqpErrorCondition.java +[AmqpErrorContext]: ../../core/azure-core-amqp/src/main/java/com/azure/core/amqp/exception/AmqpErrorContext.java +[EventHubConsumerAsyncClient]: ./src/main/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClient.java +[EventHubConsumerClient]: ./src/main/java/com/azure/messaging/eventhubs/EventHubConsumerClient.java +[EventHubProducerAsyncClient]: ./src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java +[EventHubProducerClient]: ./src/main/java/com/azure/messaging/eventhubs/EventHubProducerClient.java +[EventProcessorClient]: ./src/main/java/com/azure/messaging/eventhubs/EventProcessorClient.java +[CreateBatchOptions]: ./src/main/java/com/azure/messaging/eventhubs/models/CreateBatchOptions.java +[InMemoryCheckpointStore]: ./src/samples/java/com/azure/messaging/eventhubs/InMemoryCheckpointStore.java +[LogLevels]: ../../core/azure-core/src/main/java/com/azure/core/util/logging/ClientLogger.java ![Impressions](https://azure-sdk-impressions.azurewebsites.net/api/impressions/azure-sdk-for-java%2Fsdk%2Feventhubs%2Fazure-messaging-eventhubs%2FREADME.png) From 934b135271f3e3835a7720646eda6fa0495e6c6b Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 00:07:37 -0800 Subject: [PATCH 17/17] Fix spotbugs. --- .../eventhubs/GetEventHubMetadata.java | 36 ++++++++++--------- 1 file changed, 19 insertions(+), 17 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java index 39a3a6b4daf4..109117f647d4 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java @@ -21,25 +21,27 @@ public static void main(String[] args) { // Instantiate a client that will be used to call the service. Using a try-resource block, so it disposes of // the client when we are done. - try (EventHubProducerClient client = new EventHubClientBuilder() + EventHubProducerClient client = new EventHubClientBuilder() .connectionString(connectionString) - .buildProducerClient()) { + .buildProducerClient(); - // Querying the partition identifiers for the Event Hub. Then calling client.getPartitionProperties with the - // identifier to get information about each partition. - for (String partitionId : client.getPartitionIds()) { - PartitionProperties properties = client.getPartitionProperties(partitionId); - System.out.printf( - "Event Hub Name: %s; Partition Id: %s; Is partition empty? %s; First Sequence Number: %s; " - + "Last Enqueued Time: %s; Last Enqueued Sequence Number: %s; Last Enqueued Offset: %s%n", - properties.getEventHubName(), - properties.getId(), - properties.isEmpty(), - properties.getBeginningSequenceNumber(), - properties.getLastEnqueuedTime(), - properties.getLastEnqueuedSequenceNumber(), - properties.getLastEnqueuedOffset()); - } + // Querying the partition identifiers for the Event Hub. Then calling client.getPartitionProperties with the + // identifier to get information about each partition. + for (String partitionId : client.getPartitionIds()) { + PartitionProperties properties = client.getPartitionProperties(partitionId); + System.out.printf( + "Event Hub Name: %s; Partition Id: %s; Is partition empty? %s; First Sequence Number: %s; " + + "Last Enqueued Time: %s; Last Enqueued Sequence Number: %s; Last Enqueued Offset: %s%n", + properties.getEventHubName(), + properties.getId(), + properties.isEmpty(), + properties.getBeginningSequenceNumber(), + properties.getLastEnqueuedTime(), + properties.getLastEnqueuedSequenceNumber(), + properties.getLastEnqueuedOffset()); } + + // Dispose of the client. + client.close(); } }