From 0535a430f88913fd71ac5f3a82c30de1ee1b6e95 Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 23:56:06 -0800 Subject: [PATCH 01/21] Update documentation for EventPosition and ReceiveOperation. --- .../eventhubs/models/EventPosition.java | 4 +-- .../eventhubs/models/ReceiveOptions.java | 28 +++++++++---------- 2 files changed, 15 insertions(+), 17 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/EventPosition.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/EventPosition.java index 3a04ce67bf4d..b4567ca5e821 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/EventPosition.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/EventPosition.java @@ -62,8 +62,8 @@ public static EventPosition earliest() { /** * Corresponds to the end of the partition, where no more events are currently enqueued. Use this position to begin - * receiving from the next event to be enqueued in the partition after an {@link EventHubConsumerAsyncClient} is - * created with this position. + * receiving from the next event to be enqueued in the partition when + * {@link EventHubConsumerAsyncClient#receiveFromPartition(String, EventPosition) receiveFromPartition()} invoked. * * @return An {@link EventPosition} set to the end of an Event Hubs stream and listens for new events. */ diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/ReceiveOptions.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/ReceiveOptions.java index 68f175c9d4d4..c07acb714391 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/ReceiveOptions.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/ReceiveOptions.java @@ -30,20 +30,18 @@ public Long getOwnerLevel() { } /** - * Sets the {@code ownerLevel} value on this consumer. When populated, the level indicates that a consumer is - * intended to be the only reader of events for the requested partition and an associated consumer group. To do so, - * this consumer will attempt to assert ownership over the partition; in the case where more than one exclusive - * consumer attempts to assert ownership for the same partition/consumer group pair, the one having a larger {@link - * ReceiveOptions#getOwnerLevel()} value will "win". + * Sets the {@code ownerLevel} value on this recieve operation. When populated, the level indicates that the receive + * operation is intended to be the only reader of events for the requested partition and associated consumer group. + * To do so, this receive operation will attempt to assert ownership over the partition; in the case where + * there is more than one exclusive receive operation for the same partition/consumer group pair, the one having a + * larger {@link ReceiveOptions#getOwnerLevel()} value will "win". * - *

- * When an exclusive consumer is used, those consumers which are not exclusive or which have a lower priority will - * either not be allowed to be created, if they already exist, will encounter an exception during the next attempted - * operation. - *

+ *

When an exclusive receive operation is used, those receive operations which are not exclusive or which have a + * lower priority will either not be allowed to be created. If they already exist, will encounter an exception + * during the next attempted operation.

* - * @param priority The priority associated with an exclusive consumer; for a non-exclusive consumer, this value - * should be {@code null}. + * @param priority The priority associated with an exclusive receive operation; for a non-exclusive receive + * operation, this value should be {@code null}. * * @return The updated {@link ReceiveOptions} object. * @@ -60,10 +58,10 @@ public ReceiveOptions setOwnerLevel(Long priority) { } /** - * Gets whether or not the consumer should request information on the last enqueued event on its associated + * Gets whether or not the receive operation should request information on the last enqueued event on its associated * partition, and track that information as events are received. * - * @return {@code true} if the resulting consumer will keep track of the last enqueued information for that + * @return {@code true} if the resulting receive operation will keep track of the last enqueued information for that * partition; {@code false} otherwise. */ public boolean getTrackLastEnqueuedEventProperties() { @@ -71,7 +69,7 @@ public boolean getTrackLastEnqueuedEventProperties() { } /** - * Sets whether or not the consumer should request information on the last enqueued event on its associated + * Sets whether or not the receive operation should request information on the last enqueued event on its associated * partition, and track that information as events are received. * *

When information about the partition's last enqueued event is being tracked, each event received from the From 6c77757828095a61c0df416d31bd263b58726a06 Mon Sep 17 00:00:00 2001 From: Connie Date: Mon, 25 Nov 2019 23:59:17 -0800 Subject: [PATCH 02/21] Update sendOptions. --- .../messaging/eventhubs/models/SendOptions.java | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/SendOptions.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/SendOptions.java index a265d14e60bd..631f286cf741 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/SendOptions.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/SendOptions.java @@ -16,17 +16,17 @@ public class SendOptions { private String partitionId; /** - * Sets a hashing key to be provided for the batch of events, which instructs the Event Hubs service map this key to - * a specific partition but allowing the service to choose an arbitrary, partition for this batch of events and any - * other batches using the same partition hashing key. + * Sets a hashing key to be provided for the batch of events, which instructs the Event Hubs service to map this key + * to a specific partition. * - * The selection of a partition is stable for a given partition hashing key. Should any other batches of events be + *

The selection of a partition is stable for a given partition hashing key. Should any other batches of events be * sent using the same exact partition hashing key, the Event Hubs service will route them all to the same - * partition. + * partition.

* - * This should be specified only when there is a need to group events by partition, but there is flexibility into + *

This should be specified only when there is a need to group events by partition, but there is flexibility into * which partition they are routed. If ensuring that a batch of events is sent only to a specific partition, it is - * recommended that the identifier of the position be specified directly when sending the batch. + * recommended that the {@link #setPartitionId(String) identifier of the position be specified directly} when + * sending the batch.

* * @param partitionKey The partition hashing key to associate with the event or batch of events. * @@ -38,7 +38,7 @@ public SendOptions setPartitionKey(String partitionKey) { } /** - * Gets the partition routing key on an event batch. If specified, tells the Event Hubs service that these events + * Gets the hashing key on an event batch. If specified, tells the Event Hubs service that these events * belong to the same group and should belong to the same partition. * * @return The partition hashing key to associate with the event or batch of events. From cfd5cb6c2df6bc735999fc9fcd54faffa57aef1f Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 00:03:49 -0800 Subject: [PATCH 03/21] Update docs in EventData. --- .../com/azure/messaging/eventhubs/EventData.java | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventData.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventData.java index 6b710b63996d..2f1c27770118 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventData.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventData.java @@ -117,9 +117,9 @@ public EventData(String body) { } /** - * The set of free-form event properties which may be used for passing metadata associated with the event with the - * event body during Event Hubs operations. A common use-case for {@code properties()} is to associate serialization - * hints for the {@link #getBody()} as an aid to consumers who wish to deserialize the binary data. + * Gets the set of free-form event properties which may be used for passing metadata associated with the event with + * the event body during Event Hubs operations. A common use-case for {@code properties()} is to associate + * serialization hints for the {@link #getBody()} as an aid to consumers who wish to deserialize the binary data. * *

Adding serialization hint using {@code getProperties()}

*

In the sample, the type of telemetry is indicated by adding an application property with key "eventType".

@@ -152,7 +152,7 @@ public Map getSystemProperties() { * wish to deserialize the binary data. *

* - * @return ByteBuffer representing the data. + * @return A byte array representing the data. */ public byte[] getBody() { return Arrays.copyOf(body, body.length); @@ -178,8 +178,8 @@ public Long getOffset() { } /** - * Gets a partition key used for message partitioning. If it exists, this value was used to compute a hash to select - * a partition to send the message to. + * Gets the partition hashing key if it was set when originally publishing the event. If it exists, this value was + * used to compute a hash to select a partition to send the message to. * * @return A partition key for this Event Data. {@code null} if the {@link EventData} was not received from Event * Hubs service or there was no partition key set when the event was sent to the Event Hub. From 7c2adcfe80da72cd7cd4e20fba5308e25fa2229e Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 00:05:26 -0800 Subject: [PATCH 04/21] Update PartitionProperties. --- .../com/azure/messaging/eventhubs/PartitionProperties.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/PartitionProperties.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/PartitionProperties.java index c64add724b31..998ff71ed6a9 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/PartitionProperties.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/PartitionProperties.java @@ -67,7 +67,7 @@ public long getBeginningSequenceNumber() { /** * Gets the last sequence number of the partition's message stream. * - * @return the last sequence number of the partition's message stream. + * @return The last sequence number of the partition's message stream. */ public long getLastEnqueuedSequenceNumber() { return this.lastEnqueuedSequenceNumber; @@ -82,7 +82,7 @@ public long getLastEnqueuedSequenceNumber() { * are no longer visible within the stream. *

* - * @return the offset of the last enqueued message in the partition's stream. + * @return The offset of the last enqueued message in the partition's stream. */ public String getLastEnqueuedOffset() { return this.lastEnqueuedOffset; @@ -91,7 +91,7 @@ public String getLastEnqueuedOffset() { /** * Gets the instant, in UTC, of the last enqueued message in the partition's stream. * - * @return the instant, in UTC, of the last enqueued message in the partition's stream. + * @return The instant, in UTC, of the last enqueued message in the partition's stream. */ public Instant getLastEnqueuedTime() { return this.lastEnqueuedTime; From 3eb06ef87dc4d4a2af1d67106ae78e5c214ad2a3 Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 00:21:55 -0800 Subject: [PATCH 05/21] Update documentation on EventData. --- .../com/azure/messaging/eventhubs/EventData.java | 16 ++++++++++------ .../messaging/eventhubs/models/SendOptions.java | 4 ++-- 2 files changed, 12 insertions(+), 8 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventData.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventData.java index 2f1c27770118..3483c51ae712 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventData.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventData.java @@ -136,8 +136,8 @@ public Map getProperties() { * Properties that are populated by Event Hubs service. As these are populated by the Event Hubs service, they are * only present on a received {@link EventData}. * - * @return an encapsulation of all SystemProperties appended by EventHubs service into EventData. {@code null} if - * the {@link EventData} is not received and is created by the public constructors. + * @return An encapsulation of all system properties appended by EventHubs service into {@link EventData}. + * {@code null} if the {@link EventData} is not received from the Event Hubs service. */ public Map getSystemProperties() { return systemProperties; @@ -168,7 +168,8 @@ public String getBodyAsString() { } /** - * Gets the offset of the event when it was received from the associated Event Hub partition. + * Gets the offset of the event when it was received from the associated Event Hub partition. This is only present + * on a received {@link EventData}. * * @return The offset within the Event Hub partition of the received event. {@code null} if the {@link EventData} * was not received from Event Hubs service. @@ -179,7 +180,8 @@ public Long getOffset() { /** * Gets the partition hashing key if it was set when originally publishing the event. If it exists, this value was - * used to compute a hash to select a partition to send the message to. + * used to compute a hash to select a partition to send the message to. This is only present on a received + * {@link EventData}. * * @return A partition key for this Event Data. {@code null} if the {@link EventData} was not received from Event * Hubs service or there was no partition key set when the event was sent to the Event Hub. @@ -189,7 +191,8 @@ public String getPartitionKey() { } /** - * Gets the instant, in UTC, of when the event was enqueued in the Event Hub partition. + * Gets the instant, in UTC, of when the event was enqueued in the Event Hub partition. This is only present on a + * received {@link EventData}. * * @return The instant, in UTC, this was enqueued in the Event Hub partition. {@code null} if the {@link EventData} * was not received from Event Hubs service. @@ -200,7 +203,8 @@ public Instant getEnqueuedTime() { /** * Gets the sequence number assigned to the event when it was enqueued in the associated Event Hub partition. This - * is unique for every message received in the Event Hub partition. + * is unique for every message received in the Event Hub partition. This is only present on a received + * {@link EventData}. * * @return The sequence number for this event. {@code null} if the {@link EventData} was not received from Event * Hubs service. diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/SendOptions.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/SendOptions.java index 631f286cf741..b4e22d028378 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/SendOptions.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/SendOptions.java @@ -19,8 +19,8 @@ public class SendOptions { * Sets a hashing key to be provided for the batch of events, which instructs the Event Hubs service to map this key * to a specific partition. * - *

The selection of a partition is stable for a given partition hashing key. Should any other batches of events be - * sent using the same exact partition hashing key, the Event Hubs service will route them all to the same + *

The selection of a partition is stable for a given partition hashing key. Should any other batches of events + * be sent using the same exact partition hashing key, the Event Hubs service will route them all to the same * partition.

* *

This should be specified only when there is a need to group events by partition, but there is flexibility into From dea9f17f19713fc50cdab9a6b092d1e051753627 Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 01:03:21 -0800 Subject: [PATCH 06/21] Fixing samples and documentation. --- .../eventhubs/EventHubClientBuilder.java | 2 +- .../EventHubProducerAsyncClient.java | 51 +++++---- .../eventhubs/EventHubProducerClient.java | 12 +-- .../EventHubConsumerJavaDocCodeSamples.java | 2 +- ...ProducerAsyncClientJavaDocCodeSamples.java | 100 +++++++++++------- ...ntHubProducerClientJavaDocCodeSamples.java | 17 +-- 6 files changed, 104 insertions(+), 80 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubClientBuilder.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubClientBuilder.java index 0d1e0f5c561d..741c435cb82b 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubClientBuilder.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubClientBuilder.java @@ -73,7 +73,7 @@ *

In the sample, the namespace connection string is used to create a synchronous Event Hub consumer. Notice that * {@code "EntityPath"} is in the connection string.

* - * {@codesnippet com.azure.messaging.eventhubs.eventhubconsumerasyncclient.instantiation} + * {@codesnippet com.azure.messaging.eventhubs.eventhubconsumerclient.instantiation} * *

Creating producers and consumers that share the same connection

*

By default, a dedicated connection is created for each producer and consumer created from the builder. If users 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 f68e4eee65f1..ac5135f1d68f 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 @@ -51,8 +51,8 @@ /** * 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. + * together in batches. Depending on the {@link CreateBatchOptions options} specified when creating an + * {@link EventDataBatch}, the events may be automatically routed to an available partition or specific to a partition. * *

* Allowing automatic routing of partitions is recommended when: @@ -71,38 +71,35 @@ * *

* - *

Create a producer that routes events to any partition

- * To allow automatic routing of messages to available partition, do not specify the {@link - * CreateBatchOptions#getPartitionId() partitionId} when creating the {@link EventHubProducerAsyncClient}. - * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.instantiation} + *

Create a producer and publish events to any partition

+ *

To allow automatic routing of messages to available partition, do not specify the {@link + * CreateBatchOptions#getPartitionId() partitionId} when creating the {@link EventHubProducerAsyncClient}.

+ * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch} * - *

Create a producer that publishes events to partition "foo" with a timeout of 45 seconds.

- * Developers can push events to a single partition by specifying the + *

Publish events to partition "foo"

+ *

Developers can push events to a single partition by specifying the * {@link CreateBatchOptions#setPartitionId(String) partitionId} when creating an {@link EventHubProducerAsyncClient}. + *

* - * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.instantiation#partitionId} + * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionId} * - *

Publish events to the same partition, grouped together using {@link SendOptions#setPartitionKey(String)} - * .

- * If developers want to push similar events to end up at the same partition, but do not require them to go to a - * specific partition, they can use {@link SendOptions#setPartitionKey(String)}. - *

+ *

Publish events to the same partition, grouped together using partition key

+ *

If developers want to push similar events to end up at the same partition, but do not require them to go to a + * specific partition, they can use {@link CreateBatchOptions#setPartitionKey(String)}. * In the sample below, all the "sandwiches" end up in the same partition, but it could end up in partition 0, 1, etc. - * of the available partitions. All that matters to the end user is that they are grouped together. - *

- * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#publisher-sendOptions} + * of the available partitions. All that matters to the end user is that they are grouped together.

* - *

Publish events using an {@link EventDataBatch}.

- * Developers can create an {@link EventDataBatch}, add the events they want into it, and publish these - * events together. When creating a {@link EventDataBatch batch}, developers can specify a set of - * {@link CreateBatchOptions options} to configure this batch. - *

- * In the scenario below, the developer is creating a networked video game. They want to receive telemetry about their - * users' gaming systems, but do not want to slow down the network with telemetry. So they limit the size of their + * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey} + * + *

Publish events using a size-limited {@link EventDataBatch}

+ *

In the scenario below, the developer is creating a networked video game. They want to receive telemetry about + * their users' gaming systems, but do not want to slow down the network with telemetry. So they limit the size of their * {@link EventDataBatch batches} to be no larger than 256 bytes. The events within the batch also get hashed to the - * same partition because they all share the same {@link CreateBatchOptions#getPartitionKey()}. - *

- * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#eventDataBatch} + * same partition because they all share the same {@link CreateBatchOptions#getPartitionKey()}.

+ * + * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey-int} + * + * @see EventHubClientBuilder#buildAsyncProducerClient() */ @ServiceClient(builder = EventHubClientBuilder.class, isAsync = true) public class EventHubProducerAsyncClient implements Closeable { 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 76e72d960f5a..400af321ee95 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 @@ -16,8 +16,8 @@ /** * 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. + * together in batches. Depending on the {@link CreateBatchOptions options} specified when creating an + * {@link EventDataBatch}, the events may be automatically routed to an available partition or specific to a partition. * *

* Allowing automatic routing of partitions is recommended when: @@ -28,7 +28,7 @@ *

* *

- * If no {@link SendOptions#getPartitionId() partitionId} is specified, the following rules are used for + * If no {@link CreateBatchOptions#getPartitionId() partitionId} is specified, the following rules are used for * automatically selecting one: * *

    @@ -40,9 +40,9 @@ * *

    Create a producer that routes events to any partition

    * To allow automatic routing of messages to available partition, do not specify the {@link - * SendOptions#getPartitionId() partitionId} when creating the {@link EventHubProducerClient}. + * CreateBatchOptions#getPartitionId() partitionId} when creating the {@link EventDataBatch}. * - * {@codesnippet com.azure.messaging.eventhubs.eventhubproducerclient.instantiation} + * {@codesnippet } * *

    Create a producer that publishes events to partition "foo" with a timeout of 45 seconds.

    * Developers can push events to a single partition by specifying the @@ -75,7 +75,7 @@ *

    * {@codesnippet com.azure.messaging.eventhubs.eventhubproducerclient.send#eventDataBatch} * - * @see EventHubClient#createProducer() + * @see EventHubClientBuilder#buildProducerClient() * @see EventHubProducerAsyncClient To asynchronously generate events to an Event Hub, see EventHubAsyncProducer. */ @ServiceClient(builder = EventHubClientBuilder.class) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerJavaDocCodeSamples.java index 2ed6b4e8c886..8f2e0fcb088b 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerJavaDocCodeSamples.java @@ -20,7 +20,7 @@ public class EventHubConsumerJavaDocCodeSamples { */ public void instantiate() { // BEGIN: com.azure.messaging.eventhubs.eventhubconsumerclient.instantiation - // The required parameters are `consumerGroup` and a way to authenticate with Event Hubs using credentials. + // The required parameters are `consumerGroup`, and a way to authenticate with Event Hubs using credentials. EventHubConsumerClient consumer = new EventHubClientBuilder() .connectionString( "Endpoint={eh-namespace};SharedAccessKeyName={policy-name};SharedAccessKey={key};Entity-Path={hub-name}") diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java index d000eba95126..25342eea0b5f 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java @@ -6,6 +6,7 @@ import com.azure.messaging.eventhubs.models.CreateBatchOptions; import com.azure.messaging.eventhubs.models.SendOptions; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import java.util.concurrent.atomic.AtomicReference; @@ -18,9 +19,7 @@ public class EventHubProducerAsyncClientJavaDocCodeSamples { private final EventHubClientBuilder builder = new EventHubClientBuilder(); /** - * Code snippet demonstrating how to create an {@link EventHubProducerAsyncClient} that automatically routes events to any - * partition. - * + * Code snippet demonstrating how to create an {@link EventHubProducerAsyncClient}. */ public void instantiate() { // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.instantiation @@ -36,20 +35,46 @@ public void instantiate() { producer.close(); } + /** + * Code snippet demonstrating how to send a batch that automatically routes events to any partition. + */ + public void batchAutomaticRouting() { + // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch + // The required parameter is a way to authenticate with Event Hubs using credentials. + // The connectionString provides a way to authenticate with Event Hub. + EventHubProducerAsyncClient producer = new EventHubClientBuilder() + .connectionString( + "Endpoint={fully-qualified-namespace};SharedAccessKeyName={policy-name};SharedAccessKey={key}", + "event-hub-name") + .buildAsyncProducerClient(); + + // Creating a batch without options set, will allow for automatic routing of events to any partition. + producer.createBatch().flatMap(batch -> { + batch.tryAdd(new EventData("test-event-1")); + batch.tryAdd(new EventData("test-event-2")); + return producer.send(batch); + }).block(); + // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch + + producer.close(); + } + /** * Code snippet demonstrating how to create an {@link EventHubProducerAsyncClient} that routes events to a single * partition. - * */ public void instantiatePartitionProducer() { - // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.instantiation#partitionId - EventData eventData = new EventData("data-to-partition-foo"); - SendOptions options = new SendOptions() - .setPartitionId("foo"); - EventHubProducerAsyncClient producer = builder.buildAsyncProducerClient(); - producer.send(eventData, options); - // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.instantiation#partitionId + + // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionId + // Creating a batch with partitionId set will route all events in that batch to partition `foo`. + CreateBatchOptions options = new CreateBatchOptions().setPartitionId("foo"); + producer.createBatch(options).flatMap(batch -> { + batch.tryAdd(new EventData("test-event-1")); + batch.tryAdd(new EventData("test-event-2")); + return producer.send(batch); + }).block(); + // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionId producer.close(); } @@ -58,36 +83,31 @@ public void instantiatePartitionProducer() { * Code snippet demonstrating how to send events with a partition key. */ public void sendEventsFluxSendOptions() { - // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#publisher-sendOptions - Flux events = Flux.just( - new EventData("sourdough".getBytes(UTF_8)), - new EventData("rye".getBytes(UTF_8)), - new EventData("wheat".getBytes(UTF_8)) - ); - EventHubProducerAsyncClient producer = builder.buildAsyncProducerClient(); - SendOptions options = new SendOptions() - .setPartitionKey("bread"); - producer.send(events, options).subscribe(ignored -> System.out.println("sent"), - error -> System.err.println("Error received:" + error), - () -> System.out.println("Send complete.")); - // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#publisher-sendOptions + // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey + // Creating a batch with partitionKey set will tell the service to hash the partitionKey and decide which + // partition to send the events to. Events with the same partitionKey are always routed to the same partition. + CreateBatchOptions options = new CreateBatchOptions().setPartitionKey("sandwiches"); + producer.createBatch(options).flatMap(batch -> { + batch.tryAdd(new EventData("sourdough")); + batch.tryAdd(new EventData("rye")); + return producer.send(batch); + }).block(); + // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey } /** - * Code snippet demonstrating how to create an {@link EventDataBatch} and send it. + * Code snippet demonstrating how to create a size-limited {@link EventDataBatch} and send it. */ public void sendEventDataBatch() { final EventHubProducerAsyncClient producer = builder.buildAsyncProducerClient(); - // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#eventDataBatch + // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey-int final EventData firstEvent = new EventData("92".getBytes(UTF_8)); firstEvent.getProperties().put("telemetry", "latency"); - final EventData secondEvent = new EventData("98".getBytes(UTF_8)); secondEvent.getProperties().put("telemetry", "cpu-temperature"); - final EventData thirdEvent = new EventData("120".getBytes(UTF_8)); thirdEvent.getProperties().put("telemetry", "fps"); @@ -100,21 +120,27 @@ public void sendEventDataBatch() { producer.createBatch(options).block()); // The sample Flux contains three events, but it could be an infinite stream of telemetry events. - telemetryEvents.subscribe(event -> { + telemetryEvents.flatMap(event -> { final EventDataBatch batch = currentBatch.get(); - if (!batch.tryAdd(event)) { + if (batch.tryAdd(event)) { + return Mono.empty(); + } + + 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 the event that did not fit in the previous batch. + newBatch.tryAdd(event); + return newBatch; + })); + }).then() + .doFinally(signal -> { final EventDataBatch batch = currentBatch.getAndSet(null); - if (batch != null) { + if (batch != null && batch.getCount() > 0) { producer.send(batch).block(); } }); - // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#eventDataBatch + // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey-int } } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerClientJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerClientJavaDocCodeSamples.java index d8119b32ea58..18c5542f68ac 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerClientJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerClientJavaDocCodeSamples.java @@ -18,13 +18,14 @@ public class EventHubProducerClientJavaDocCodeSamples { private final EventHubClientBuilder builder = new EventHubClientBuilder() .connectionString("fake-string"); + /** - * Code snippet demonstrating how to create an {@link EventHubProducerClient} that automatically routes events to any - * partition. - * + * Code snippet demonstrating how to create an {@link EventHubProducerClient}. */ public void instantiate() { // BEGIN: com.azure.messaging.eventhubs.eventhubproducerclient.instantiation + // The required parameter is a way to authenticate with Event Hubs using credentials. + // The connectionString provides a way to authenticate with Event Hub. EventHubProducerClient producer = new EventHubClientBuilder() .connectionString("event-hubs-namespace-connection-string", "event-hub-name") .buildProducerClient(); @@ -36,16 +37,16 @@ public void instantiate() { /** * Code snippet demonstrating how to send events to a single partition. * - * @throws IOException if the producer cannot be disposed. */ - public void instantiatePartitionProducer() throws IOException { + public void instantiatePartitionProducer() { // BEGIN: com.azure.messaging.eventhubs.eventhubproducerclient.instantiation#partitionId - EventData eventData = new EventData("data-to-partition-foo"); - SendOptions options = new SendOptions() + CreateBatchOptions options = new CreateBatchOptions() .setPartitionId("foo"); EventHubProducerClient producer = builder.buildProducerClient(); - producer.send(eventData, options); + EventDataBatch batch = producer.createBatch(options); + batch.tryAdd(new EventData("data-to-partition-foo")); + producer.send(batch); // END: com.azure.messaging.eventhubs.eventhubproducerclient.instantiation#partitionId producer.close(); From b5b37e0a0b49af76e10d50805c0000cca16536d4 Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 10:01:30 -0800 Subject: [PATCH 07/21] Fix checkstyle issues. --- .../eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java | 1 - .../eventhubs/EventHubProducerClientJavaDocCodeSamples.java | 1 - 2 files changed, 2 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java index 25342eea0b5f..42f32b8dd6a9 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java @@ -4,7 +4,6 @@ package com.azure.messaging.eventhubs; import com.azure.messaging.eventhubs.models.CreateBatchOptions; -import com.azure.messaging.eventhubs.models.SendOptions; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerClientJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerClientJavaDocCodeSamples.java index 18c5542f68ac..d2901a581a8e 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerClientJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerClientJavaDocCodeSamples.java @@ -6,7 +6,6 @@ import com.azure.messaging.eventhubs.models.CreateBatchOptions; import com.azure.messaging.eventhubs.models.SendOptions; -import java.io.IOException; import java.util.Arrays; import java.util.List; From 54dce1e2dd706629df496b969a463cdcbda53854 Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 10:08:50 -0800 Subject: [PATCH 08/21] Update documentation in EventHubProducerAsyncClient and samples. --- .../eventhubs/EventHubProducerAsyncClient.java | 16 ---------------- ...HubProducerAsyncClientJavaDocCodeSamples.java | 14 +++++++------- 2 files changed, 7 insertions(+), 23 deletions(-) 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 ac5135f1d68f..02f2a68673ef 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 @@ -72,31 +72,15 @@ *

    * *

    Create a producer and publish events to any partition

    - *

    To allow automatic routing of messages to available partition, do not specify the {@link - * CreateBatchOptions#getPartitionId() partitionId} when creating the {@link EventHubProducerAsyncClient}.

    * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch} * *

    Publish events to partition "foo"

    - *

    Developers can push events to a single partition by specifying the - * {@link CreateBatchOptions#setPartitionId(String) partitionId} when creating an {@link EventHubProducerAsyncClient}. - *

    - * * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionId} * *

    Publish events to the same partition, grouped together using partition key

    - *

    If developers want to push similar events to end up at the same partition, but do not require them to go to a - * specific partition, they can use {@link CreateBatchOptions#setPartitionKey(String)}. - * In the sample below, all the "sandwiches" end up in the same partition, but it could end up in partition 0, 1, etc. - * of the available partitions. All that matters to the end user is that they are grouped together.

    - * * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey} * *

    Publish events using a size-limited {@link EventDataBatch}

    - *

    In the scenario below, the developer is creating a networked video game. They want to receive telemetry about - * their users' gaming systems, but do not want to slow down the network with telemetry. So they limit the size of their - * {@link EventDataBatch batches} to be no larger than 256 bytes. The events within the batch also get hashed to the - * same partition because they all share the same {@link CreateBatchOptions#getPartitionKey()}.

    - * * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey-int} * * @see EventHubClientBuilder#buildAsyncProducerClient() diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java index 42f32b8dd6a9..ca3127a15614 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java @@ -62,7 +62,7 @@ public void batchAutomaticRouting() { * Code snippet demonstrating how to create an {@link EventHubProducerAsyncClient} that routes events to a single * partition. */ - public void instantiatePartitionProducer() { + public void batchPartitionId() { EventHubProducerAsyncClient producer = builder.buildAsyncProducerClient(); // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionId @@ -81,7 +81,7 @@ public void instantiatePartitionProducer() { /** * Code snippet demonstrating how to send events with a partition key. */ - public void sendEventsFluxSendOptions() { + public void batchPartitionKey() { EventHubProducerAsyncClient producer = builder.buildAsyncProducerClient(); // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey @@ -99,7 +99,7 @@ public void sendEventsFluxSendOptions() { /** * Code snippet demonstrating how to create a size-limited {@link EventDataBatch} and send it. */ - public void sendEventDataBatch() { + public void batchSizeLimited() { final EventHubProducerAsyncClient producer = builder.buildAsyncProducerClient(); // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey-int @@ -107,18 +107,18 @@ public void sendEventDataBatch() { firstEvent.getProperties().put("telemetry", "latency"); final EventData secondEvent = new EventData("98".getBytes(UTF_8)); secondEvent.getProperties().put("telemetry", "cpu-temperature"); - final EventData thirdEvent = new EventData("120".getBytes(UTF_8)); - thirdEvent.getProperties().put("telemetry", "fps"); - final Flux telemetryEvents = Flux.just(firstEvent, secondEvent, thirdEvent); + final Flux telemetryEvents = Flux.just(firstEvent, secondEvent); + // Setting `setMaximumSizeInBytes` when creating a batch, limits the size of that batch. + // In this case, all the batches created with these options are limited to 256 bytes. final CreateBatchOptions options = new CreateBatchOptions() .setPartitionKey("telemetry") .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. + // The sample Flux contains two events, but it could be an infinite stream of telemetry events. telemetryEvents.flatMap(event -> { final EventDataBatch batch = currentBatch.get(); if (batch.tryAdd(event)) { From 0f09277d6867e514f2a10187f22f190490b62833 Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 10:16:20 -0800 Subject: [PATCH 09/21] Update documentation in EventHubProducerAsycnClient. --- .../messaging/eventhubs/EventHubProducerAsyncClient.java | 9 +++++++-- .../EventHubProducerAsyncClientJavaDocCodeSamples.java | 4 +--- 2 files changed, 8 insertions(+), 5 deletions(-) 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 02f2a68673ef..f8c0e7e694aa 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 @@ -170,6 +170,8 @@ public Flux getPartitionIds() { * @param partitionId The unique identifier of a partition associated with the Event Hub. * * @return The set of information for the requested partition under the Event Hub this client is associated with. + * + * @throws NullPointerException if {@code partitionId} is null. */ @ServiceMethod(returns = ReturnType.SINGLE) public Mono getPartitionProperties(String partitionId) { @@ -186,11 +188,13 @@ public Mono createBatch() { } /** - * Creates an {@link EventDataBatch} that can fit as many events as the transport allows. + * Creates an {@link EventDataBatch} configured with the options specified. * * @param options A set of options used to configure the {@link EventDataBatch}. * * @return A new {@link EventDataBatch} that can fit as many events as the transport allows. + * + * @throws NullPointerException if {@code options} is null. */ public Mono createBatch(CreateBatchOptions options) { if (options == null) { @@ -488,7 +492,8 @@ private Mono getSendLink(String partitionId) { } /** - * Disposes of the {@link EventHubProducerAsyncClient} by closing the underlying connection to the service. + * Disposes of the {@link EventHubProducerAsyncClient}. If the client had a dedicated connection, the underlying + * connection is also closed. */ @Override public void close() { diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java index ca3127a15614..f1d221739118 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java @@ -101,19 +101,17 @@ public void batchPartitionKey() { */ public void batchSizeLimited() { final EventHubProducerAsyncClient producer = builder.buildAsyncProducerClient(); - - // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey-int final EventData firstEvent = new EventData("92".getBytes(UTF_8)); firstEvent.getProperties().put("telemetry", "latency"); final EventData secondEvent = new EventData("98".getBytes(UTF_8)); secondEvent.getProperties().put("telemetry", "cpu-temperature"); + // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey-int final Flux telemetryEvents = Flux.just(firstEvent, secondEvent); // Setting `setMaximumSizeInBytes` when creating a batch, limits the size of that batch. // In this case, all the batches created with these options are limited to 256 bytes. final CreateBatchOptions options = new CreateBatchOptions() - .setPartitionKey("telemetry") .setMaximumSizeInBytes(256); final AtomicReference currentBatch = new AtomicReference<>( producer.createBatch(options).block()); From ee49a21cd9f74b3d773266a9e0d5c5d8010db1a1 Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 14:06:47 -0800 Subject: [PATCH 10/21] Update javadoc snippets. --- .../EventHubProducerAsyncClient.java | 2 +- .../eventhubs/EventHubProducerClient.java | 6 +- ...ProducerAsyncClientJavaDocCodeSamples.java | 9 +- ...ntHubProducerClientJavaDocCodeSamples.java | 102 ++++++++++++------ 4 files changed, 78 insertions(+), 41 deletions(-) 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 f8c0e7e694aa..263b20fb0f48 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 @@ -81,7 +81,7 @@ * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey} * *

    Publish events using a size-limited {@link EventDataBatch}

    - * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey-int} + * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-int} * * @see EventHubClientBuilder#buildAsyncProducerClient() */ 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 400af321ee95..b18790d77113 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 @@ -48,7 +48,7 @@ * Developers can push events to a single partition by specifying the * {@link SendOptions#setPartitionId(String) partitionId} when creating an {@link EventHubProducerClient}. * - * {@codesnippet com.azure.messaging.eventhubs.eventhubproducerclient.instantiation#partitionId} + * {@codesnippet com.azure.messaging.eventhubs.eventhubproducerclient.createBatch#CreateBatchOptions-partitionId} * *

    Publish events to the same partition, grouped together using {@link SendOptions#setPartitionKey(String)} *

    @@ -60,7 +60,7 @@ * of the available partitions. All that matters to the end user is that they are grouped together. *

    * - * {@codesnippet com.azure.messaging.eventhubs.eventhubproducerclient.send#publisher-sendOptions} + * {@codesnippet com.azure.messaging.eventhubs.eventhubproducerclient.createBatch#CreateBatchOptions-partitionKey} * *

    Publish events using an {@link EventDataBatch}

    * Developers can create an {@link EventDataBatch}, add the events they want into it, and publish these events together. @@ -73,7 +73,7 @@ * {@link EventDataBatch batches} to be no larger than 256 bytes. The events within the batch also get hashed to the * same partition because they all share the same {@link CreateBatchOptions#getPartitionKey()}. *

    - * {@codesnippet com.azure.messaging.eventhubs.eventhubproducerclient.send#eventDataBatch} + * {@codesnippet com.azure.messaging.eventhubs.eventhubproducerclient.createBatch#CreateBatchOptions-int} * * @see EventHubClientBuilder#buildProducerClient() * @see EventHubProducerAsyncClient To asynchronously generate events to an Event Hub, see EventHubAsyncProducer. diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java index f1d221739118..8813da500ae5 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java @@ -59,8 +59,7 @@ public void batchAutomaticRouting() { } /** - * Code snippet demonstrating how to create an {@link EventHubProducerAsyncClient} that routes events to a single - * partition. + * Code snippet demonstrating how to create an EventDataBatch at routes events to a single partition. */ public void batchPartitionId() { EventHubProducerAsyncClient producer = builder.buildAsyncProducerClient(); @@ -87,7 +86,7 @@ public void batchPartitionKey() { // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey // Creating a batch with partitionKey set will tell the service to hash the partitionKey and decide which // partition to send the events to. Events with the same partitionKey are always routed to the same partition. - CreateBatchOptions options = new CreateBatchOptions().setPartitionKey("sandwiches"); + CreateBatchOptions options = new CreateBatchOptions().setPartitionKey("bread"); producer.createBatch(options).flatMap(batch -> { batch.tryAdd(new EventData("sourdough")); batch.tryAdd(new EventData("rye")); @@ -106,7 +105,7 @@ public void batchSizeLimited() { final EventData secondEvent = new EventData("98".getBytes(UTF_8)); secondEvent.getProperties().put("telemetry", "cpu-temperature"); - // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey-int + // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-int final Flux telemetryEvents = Flux.just(firstEvent, secondEvent); // Setting `setMaximumSizeInBytes` when creating a batch, limits the size of that batch. @@ -138,6 +137,6 @@ public void batchSizeLimited() { producer.send(batch).block(); } }); - // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey-int + // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-int } } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerClientJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerClientJavaDocCodeSamples.java index d2901a581a8e..912c503eaa29 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerClientJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerClientJavaDocCodeSamples.java @@ -4,7 +4,6 @@ package com.azure.messaging.eventhubs; import com.azure.messaging.eventhubs.models.CreateBatchOptions; -import com.azure.messaging.eventhubs.models.SendOptions; import java.util.Arrays; import java.util.List; @@ -26,7 +25,9 @@ public void instantiate() { // The required parameter is a way to authenticate with Event Hubs using credentials. // The connectionString provides a way to authenticate with Event Hub. EventHubProducerClient producer = new EventHubClientBuilder() - .connectionString("event-hubs-namespace-connection-string", "event-hub-name") + .connectionString( + "Endpoint={fully-qualified-namespace};SharedAccessKeyName={policy-name};SharedAccessKey={key}", + "event-hub-name") .buildProducerClient(); // END: com.azure.messaging.eventhubs.eventhubproducerclient.instantiation @@ -34,61 +35,93 @@ public void instantiate() { } /** - * Code snippet demonstrating how to send events to a single partition. - * + * Code snippet demonstrating how to send a batch that automatically routes events to any partition. */ - public void instantiatePartitionProducer() { - // BEGIN: com.azure.messaging.eventhubs.eventhubproducerclient.instantiation#partitionId - CreateBatchOptions options = new CreateBatchOptions() - .setPartitionId("foo"); + public void batchAutomaticRouting() { + // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch + // The required parameter is a way to authenticate with Event Hubs using credentials. + // The connectionString provides a way to authenticate with Event Hub. + EventHubProducerClient producer = new EventHubClientBuilder() + .connectionString( + "Endpoint={fully-qualified-namespace};SharedAccessKeyName={policy-name};SharedAccessKey={key}", + "event-hub-name") + .buildProducerClient(); + List events = Arrays.asList(new EventData("test-event-1"), new EventData("test-event-2")); + + // Creating a batch without options set, will allow for automatic routing of events to any partition. + EventDataBatch batch = producer.createBatch(); + for (EventData event : events) { + if (batch.tryAdd(event)) { + continue; + } + + producer.send(batch); + batch = producer.createBatch(); + if (!batch.tryAdd(event)) { + throw new IllegalArgumentException("Event is too large for an empty batch."); + } + } + // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch + + producer.close(); + } + + /** + * Code snippet demonstrating how to create an EventDataBatch at routes events to a single partition. + */ + public void batchPartitionId() { + final EventHubProducerClient producer = builder.buildProducerClient(); + + // BEGIN: com.azure.messaging.eventhubs.eventhubproducerclient.createBatch#CreateBatchOptions-partitionId + // Creating a batch with partitionId set will route all events in that batch to partition `foo`. + CreateBatchOptions options = new CreateBatchOptions().setPartitionId("foo"); - EventHubProducerClient producer = builder.buildProducerClient(); EventDataBatch batch = producer.createBatch(options); batch.tryAdd(new EventData("data-to-partition-foo")); producer.send(batch); - // END: com.azure.messaging.eventhubs.eventhubproducerclient.instantiation#partitionId - - producer.close(); + // END: com.azure.messaging.eventhubs.eventhubproducerclient.createBatch#CreateBatchOptions-partitionId } /** * Code snippet demonstrating how to send events with a partition key. */ - public void sendEventsSendOptions() { - // BEGIN: com.azure.messaging.eventhubs.eventhubproducerclient.send#publisher-sendOptions - final List events = Arrays.asList( - new EventData("sourdough".getBytes(UTF_8)), - new EventData("rye".getBytes(UTF_8)), - new EventData("wheat".getBytes(UTF_8)) - ); - + public void batchPartitionKey() { final EventHubProducerClient producer = builder.buildProducerClient(); - final SendOptions options = new SendOptions() - .setPartitionKey("bread"); - producer.send(events, options); - // END: com.azure.messaging.eventhubs.eventhubproducerclient.send#publisher-sendOptions + // BEGIN: com.azure.messaging.eventhubs.eventhubproducerclient.createBatch#CreateBatchOptions-partitionKey + List events = Arrays.asList(new EventData("sourdough"), new EventData("rye"), + new EventData("wheat")); + + // Creating a batch with partitionKey set will tell the service to hash the partitionKey and decide which + // partition to send the events to. Events with the same partitionKey are always routed to the same partition. + CreateBatchOptions options = new CreateBatchOptions().setPartitionKey("bread"); + EventDataBatch batch = producer.createBatch(options); + + events.forEach(event -> batch.tryAdd(event)); + producer.send(batch); + // END: com.azure.messaging.eventhubs.eventhubproducerclient.createBatch#CreateBatchOptions-partitionKey } /** - * Code snippet demonstrating how to create an {@link EventDataBatch} and send it. + * Code snippet demonstrating how to create a size-limited {@link EventDataBatch} and send it. */ - public void sendEventDataBatch() { + public void batchSizeLimited() { final EventHubProducerClient producer = builder.buildProducerClient(); - - // BEGIN: com.azure.messaging.eventhubs.eventhubproducerclient.send#eventDataBatch final EventData firstEvent = new EventData("92".getBytes(UTF_8)); firstEvent.getProperties().put("telemetry", "latency"); - final EventData secondEvent = new EventData("98".getBytes(UTF_8)); secondEvent.getProperties().put("telemetry", "cpu-temperature"); - final EventData thirdEvent = new EventData("120".getBytes(UTF_8)); thirdEvent.getProperties().put("telemetry", "fps"); + // BEGIN: com.azure.messaging.eventhubs.eventhubproducerclient.createBatch#CreateBatchOptions-int + + final List telemetryEvents = Arrays.asList(firstEvent, secondEvent, thirdEvent); + + // Setting `setMaximumSizeInBytes` when creating a batch, limits the size of that batch. + // In this case, all the batches created with these options are limited to 256 bytes. final CreateBatchOptions options = new CreateBatchOptions() - .setPartitionKey("telemetry") .setMaximumSizeInBytes(256); EventDataBatch currentBatch = producer.createBatch(options); @@ -99,8 +132,13 @@ public void sendEventDataBatch() { if (!currentBatch.tryAdd(event)) { producer.send(currentBatch); currentBatch = producer.createBatch(options); + + // Add the event we couldn't before. + if (!currentBatch.tryAdd(event)) { + throw new IllegalArgumentException("Event is too large for an empty batch."); + } } } - // END: com.azure.messaging.eventhubs.eventhubproducerclient.send#eventDataBatch + // END: com.azure.messaging.eventhubs.eventhubproducerclient.createBatch#CreateBatchOptions-int } } From 292e2d0a251ee767df253ee1c74b6d3e9264f25a Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 14:12:43 -0800 Subject: [PATCH 11/21] Fix documentation for EventHubProducer*Clients. --- .../EventHubProducerAsyncClient.java | 9 +-- .../eventhubs/EventHubProducerClient.java | 56 ++++++------------- 2 files changed, 22 insertions(+), 43 deletions(-) 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 263b20fb0f48..0b981cd94d63 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 @@ -63,7 +63,7 @@ *

    * *

    - * If no partition is specified, the following rules are used for automatically selecting one: + * If no partition id is specified, the following rules are used for automatically selecting one: *

      *
    1. Distribute the events equally amongst all available partitions using a round-robin approach.
      2. *
    2. If a partition becomes unavailable, the Event Hubs service will automatically detect it and forward the @@ -84,6 +84,7 @@ * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-int} * * @see EventHubClientBuilder#buildAsyncProducerClient() + * @see EventHubProducerClient To synchronously generate events to an Event Hub, see EventHubProducerClient. */ @ServiceClient(builder = EventHubClientBuilder.class, isAsync = true) public class EventHubProducerAsyncClient implements Closeable { @@ -129,7 +130,7 @@ public class EventHubProducerAsyncClient implements Closeable { * Gets the fully qualified Event Hubs namespace that the connection is associated with. This is likely similar to * {@code {yournamespace}.servicebus.windows.net}. * - * @return The fully qualified Event Hubs namespace that the connection is associated with + * @return The fully qualified Event Hubs namespace that the connection is associated with. */ public String getFullyQualifiedNamespace() { return fullyQualifiedNamespace; @@ -247,7 +248,7 @@ public Mono createBatch(CreateBatchOptions options) { * *

      * For more information regarding the maximum event size allowed, see - * Azure Event Hubs Quotas and + * Azure Event Hubs Quotas and * Limits. *

      * @@ -269,7 +270,7 @@ Mono send(EventData event) { * *

      * For more information regarding the maximum event size allowed, see - * Azure Event Hubs Quotas and + * Azure Event Hubs Quotas and * Limits. *

      * 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 b18790d77113..d3a34481e14d 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 @@ -28,8 +28,7 @@ *

      * *

      - * If no {@link CreateBatchOptions#getPartitionId() partitionId} is specified, the following rules are used for - * automatically selecting one: + * If no partition id is specified, the following rules are used for automatically selecting one: * *

        *
      1. Distribute the events equally amongst all available partitions using a round-robin approach.
        2. @@ -38,45 +37,20 @@ *
      *

      * - *

      Create a producer that routes events to any partition

      - * To allow automatic routing of messages to available partition, do not specify the {@link - * CreateBatchOptions#getPartitionId() partitionId} when creating the {@link EventDataBatch}. - * - * {@codesnippet } - * - *

      Create a producer that publishes events to partition "foo" with a timeout of 45 seconds.

      - * Developers can push events to a single partition by specifying the - * {@link SendOptions#setPartitionId(String) partitionId} when creating an {@link EventHubProducerClient}. + *

      Create a producer and publish events to any partition

      + * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch} * + *

      Publish events to partition "foo"

      * {@codesnippet com.azure.messaging.eventhubs.eventhubproducerclient.createBatch#CreateBatchOptions-partitionId} * - *

      Publish events to the same partition, grouped together using {@link SendOptions#setPartitionKey(String)} - *

      - * If developers want to push similar events to end up at the same partition, but do not require them to go to a - * specific partition, they can use {@link SendOptions#setPartitionKey(String)}. - * - *

      - * In the sample below, all the "sandwiches" end up in the same partition, but it could end up in partition 0, 1, etc. - * of the available partitions. All that matters to the end user is that they are grouped together. - *

      - * + *

      Publish events to the same partition, grouped together using partition key

      * {@codesnippet com.azure.messaging.eventhubs.eventhubproducerclient.createBatch#CreateBatchOptions-partitionKey} * - *

      Publish events using an {@link EventDataBatch}

      - * Developers can create an {@link EventDataBatch}, add the events they want into it, and publish these events together. - * When creating a {@link EventDataBatch batch}, developers can specify a set of {@link CreateBatchOptions options} to - * configure this batch. - * - *

      - * In the scenario below, the developer is creating a networked video game. They want to receive telemetry about their - * users' gaming systems, but do not want to slow down the network with telemetry. So they limit the size of their - * {@link EventDataBatch batches} to be no larger than 256 bytes. The events within the batch also get hashed to the - * same partition because they all share the same {@link CreateBatchOptions#getPartitionKey()}. - *

      + *

      Publish events using a size-limited {@link EventDataBatch}

      * {@codesnippet com.azure.messaging.eventhubs.eventhubproducerclient.createBatch#CreateBatchOptions-int} * * @see EventHubClientBuilder#buildProducerClient() - * @see EventHubProducerAsyncClient To asynchronously generate events to an Event Hub, see EventHubAsyncProducer. + * @see EventHubProducerAsyncClient To asynchronously generate events to an Event Hub, see EventHubProducerAsyncClient. */ @ServiceClient(builder = EventHubClientBuilder.class) public class EventHubProducerClient implements Closeable { @@ -106,7 +80,7 @@ public String getEventHubName() { * Gets the fully qualified Event Hubs namespace that the connection is associated with. This is likely similar to * {@code {yournamespace}.servicebus.windows.net}. * - * @return The fully qualified Event Hubs namespace that the connection is associated with + * @return The fully qualified Event Hubs namespace that the connection is associated with. */ public String getFullyQualifiedNamespace() { return producer.getFullyQualifiedNamespace(); @@ -137,6 +111,7 @@ public IterableStream getPartitionIds() { * * @param partitionId The unique identifier of a partition associated with the Event Hub. * @return The set of information for the requested partition under the Event Hub this client is associated with. + * @throws NullPointerException if {@code partitionId} is null. */ @ServiceMethod(returns = ReturnType.SINGLE) public PartitionProperties getPartitionProperties(String partitionId) { @@ -153,10 +128,13 @@ public EventDataBatch createBatch() { } /** - * Creates an {@link EventDataBatch} that can fit as many events as the transport allows. + * Creates an {@link EventDataBatch} configured with the options specified. * * @param options A set of options used to configure the {@link EventDataBatch}. + * * @return A new {@link EventDataBatch} that can fit as many events as the transport allows. + * + * @throws NullPointerException if {@code options} is null. */ public EventDataBatch createBatch(CreateBatchOptions options) { return producer.createBatch(options).block(tryTimeout); @@ -168,7 +146,7 @@ public EventDataBatch createBatch(CreateBatchOptions options) { * *

      * For more information regarding the maximum event size allowed, see - * Azure Event Hubs Quotas and + * Azure Event Hubs Quotas and * Limits. *

      * @@ -184,7 +162,7 @@ void send(EventData event) { * *

      * For more information regarding the maximum event size allowed, see - * Azure Event Hubs Quotas and + * Azure Event Hubs Quotas and * Limits. *

      * @@ -202,7 +180,7 @@ void send(EventData event, SendOptions options) { * *

      * For more information regarding the maximum event size allowed, see - * Azure Event Hubs Quotas and + * Azure Event Hubs Quotas and * Limits. *

      * @@ -219,7 +197,7 @@ void send(Iterable events) { * *

      * For more information regarding the maximum event size allowed, see - * Azure Event Hubs Quotas and + * Azure Event Hubs Quotas and * Limits. *

      * From 0003c89974ff1806bd8c5f5b06cc6397e9f7e39f Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 14:27:12 -0800 Subject: [PATCH 12/21] Fix spotbug issues. --- .../messaging/eventhubs/EventHubConsumerAsyncClient.java | 3 +-- .../eventhubs/EventHubProducerClientJavaDocCodeSamples.java | 4 ++++ 2 files changed, 5 insertions(+), 2 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 30014b962d0d..595b9b6a650e 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 @@ -33,8 +33,7 @@ * 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 - * creating a consumer.

      {@codesnippet com.azure.messaging.eventhubs.eventhubconsumerasyncclient.instantiation} + * {@codesnippet com.azure.messaging.eventhubs.eventhubconsumerasyncclient.instantiation} * *

      Consuming events a single partition from Event Hub

      * {@codesnippet com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receive#string-eventposition} diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerClientJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerClientJavaDocCodeSamples.java index 912c503eaa29..265e56c7373a 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerClientJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerClientJavaDocCodeSamples.java @@ -36,6 +36,8 @@ public void instantiate() { /** * Code snippet demonstrating how to send a batch that automatically routes events to any partition. + * + * @throws IllegalArgumentException if an event is too large for an empty batch. */ public void batchAutomaticRouting() { // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch @@ -104,6 +106,8 @@ public void batchPartitionKey() { /** * Code snippet demonstrating how to create a size-limited {@link EventDataBatch} and send it. + * + * @throws IllegalArgumentException if an event is too large for an empty batch. */ public void batchSizeLimited() { final EventHubProducerClient producer = builder.buildProducerClient(); From 0897b7444cd2bb9fdc1376831bbb5c6d73614f91 Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 14:33:41 -0800 Subject: [PATCH 13/21] Fixing snippets. --- .../EventHubClientBuilderJavaDocCodeSamples.java | 1 - .../EventHubConsumerAsyncClientJavaDocCodeSamples.java | 10 ++++++---- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubClientBuilderJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubClientBuilderJavaDocCodeSamples.java index a4a2e0e330c8..29ce00fbf27e 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubClientBuilderJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubClientBuilderJavaDocCodeSamples.java @@ -20,7 +20,6 @@ public void sharingConnection() { // Both the producer and consumer created share the same underlying connection. EventHubProducerAsyncClient producer = builder.buildAsyncProducerClient(); - EventHubConsumerAsyncClient consumer = builder .consumerGroup("my-consumer-group") .buildAsyncConsumerClient(); 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 8485f7088c90..8273f5b0368a 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 @@ -18,6 +18,10 @@ * Code snippets demonstrating various {@link EventHubConsumerAsyncClient} scenarios. */ public class EventHubConsumerAsyncClientJavaDocCodeSamples { + private final EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() + .connectionString("fake-string") + .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .buildAsyncConsumerClient(); public void initialization() { // BEGIN: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.instantiation @@ -28,16 +32,14 @@ public void initialization() { .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) .buildAsyncConsumerClient(); // END: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.instantiation + + consumer.close(); } /** * Receives event data from a single partition. */ public void receive() { - EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() - .connectionString("fake-string") - .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) - .buildAsyncConsumerClient(); // BEGIN: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receive#string-eventposition // Obtain partitionId from EventHubConsumerAsyncClient.getPartitionIds() From 766dabe0474c6dac78dfc865a8a940e2a3f8e0d9 Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 15:16:04 -0800 Subject: [PATCH 14/21] Update documentation for EHConsumerAClient. --- .../EventHubConsumerAsyncClient.java | 66 ++++++++++--------- ...ConsumerAsyncClientJavaDocCodeSamples.java | 18 ++--- 2 files changed, 45 insertions(+), 39 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 595b9b6a650e..ff779be0e4df 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 @@ -12,7 +12,6 @@ import com.azure.core.util.logging.ClientLogger; import com.azure.messaging.eventhubs.implementation.EventHubManagementNode; import com.azure.messaging.eventhubs.models.EventPosition; -import com.azure.messaging.eventhubs.models.PartitionContext; import com.azure.messaging.eventhubs.models.PartitionEvent; import com.azure.messaging.eventhubs.models.ReceiveOptions; import reactor.core.publisher.BaseSubscriber; @@ -38,15 +37,16 @@ *

      Consuming events a single partition from Event Hub

      * {@codesnippet com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receive#string-eventposition} * - *

      Rate limiting consumption of events from Event Hub

      - *

      For event consumers that need to limit the number of events they receive at a given time, they can use {@link - * BaseSubscriber#request(long)}.

      {@codesnippet com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receive#string-eventposition-basesubscriber} - * *

      Viewing latest partition information

      *

      Latest partition information as events are received can by setting - * {@link ReceiveOptions#setTrackLastEnqueuedEventProperties(boolean) setTrackLastEnqueuedEventProperties} to {@code - * true}. As events come in, explore the {@link PartitionContext} object. {@codesnippet - * com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receive#boolean-receiveoptions} + * {@link ReceiveOptions#setTrackLastEnqueuedEventProperties(boolean) setTrackLastEnqueuedEventProperties} to + * {@code true}. As events come in, explore the {@link PartitionEvent} object. + * {@codesnippet com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receiveFromPartition#string-eventposition-receiveoptions} + * + *

      Rate limiting consumption of events from Event Hub

      + *

      For event consumers that need to limit the number of events they receive at a given time, they can use + * {@link BaseSubscriber#request(long)}.

      + * {@codesnippet com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receive#string-eventposition-basesubscriber} * *

      Receiving from all partitions

      * {@codesnippet com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receive#boolean} @@ -170,10 +170,10 @@ public Flux receiveFromPartition(String partitionId, EventPositi * *
        *
      • If receive is invoked where {@link ReceiveOptions#getOwnerLevel()} has a value, then Event Hubs service will - * guarantee only one active consumer exists per partitionId and consumer group combination. This consumer is - * sometimes referred to as an "Epoch Consumer."
        • + * guarantee only one active consumer exists per partitionId and consumer group combination. This receive operation + * is sometimes referred to as an "Epoch Consumer". *
      • Multiple consumers per partitionId and consumer group combination can be created by not setting - * {@link ReceiveOptions#getOwnerLevel()} when creating consumers. This non-exclusive consumer is sometimes + * {@link ReceiveOptions#getOwnerLevel()} when invoking receive operations. This non-exclusive consumer is sometimes * referred to as a "Non-Epoch Consumer."
        • *
      * @@ -207,17 +207,16 @@ public Flux receiveFromPartition(String partitionId, EventPositi /** * Consumes events from all partitions starting from the beginning of each partition. * - *

      - * This method is not recommended for production use; the {@link EventProcessorClient} should be used for reading - * events from all partitions in a production scenario, as it offers a much more robust experience with higher - * throughput. + *

      This method is not recommended for production use; the {@link EventProcessorClient} should be used for + * reading events from all partitions in a production scenario, as it offers a much more robust experience with + * higher throughput. * * It is important to note that this method does not guarantee fairness amongst the partitions. Depending on service * communication, there may be a clustering of events per partition and/or there may be a noticeable bias for a - * given partition or subset of partitions. - *

      + * given partition or subset of partitions.

      + * * - * @return A stream of events for every partition in the Event Hub starting from {@code startingPosition}. + * @return A stream of events for every partition in the Event Hub starting from the beginning of each partition. */ public Flux receive() { return receive(true, defaultReceiveOptions); @@ -226,15 +225,13 @@ public Flux receive() { /** * Consumes events from all partitions. * - *

      - * This method is not recommended for production use; the {@link EventProcessorClient} should be used for reading - * events from all partitions in a production scenario, as it offers a much more robust experience with higher - * throughput. + *

      This method is not recommended for production use; the {@link EventProcessorClient} should be used for + * reading events from all partitions in a production scenario, as it offers a much more robust experience with + * higher throughput. * * It is important to note that this method does not guarantee fairness amongst the partitions. Depending on service * communication, there may be a clustering of events per partition and/or there may be a noticeable bias for a - * given partition or subset of partitions. - *

      + * given partition or subset of partitions.

      * * @param startReadingAtEarliestEvent {@code true} to begin reading at the first events available in each * partition; otherwise, reading will begin at the end of each partition seeing only new events as they are @@ -247,17 +244,24 @@ public Flux receive(boolean startReadingAtEarliestEvent) { } /** - * Consumes events from all partitions. + * Consumes events from all partitions configured with a set of {@code receiveOptions}. * - *

      - * This method is not recommended for production use; the {@link EventProcessorClient} should be used for reading - * events from all partitions in a production scenario, as it offers a much more robust experience with higher - * throughput. + *

      This method is not recommended for production use; the {@link EventProcessorClient} should be used for + * reading events from all partitions in a production scenario, as it offers a much more robust experience with + * higher throughput. * * It is important to note that this method does not guarantee fairness amongst the partitions. Depending on service * communication, there may be a clustering of events per partition and/or there may be a noticeable bias for a - * given partition or subset of partitions. - *

      + * given partition or subset of partitions.

      + * + *
        + *
      • If receive is invoked where {@link ReceiveOptions#getOwnerLevel()} has a value, then Event Hubs service will + * guarantee only one active consumer exists per partitionId and consumer group combination. This receive operation + * is sometimes referred to as an "Epoch Consumer".
        • + *
      • Multiple consumers per partitionId and consumer group combination can be created by not setting + * {@link ReceiveOptions#getOwnerLevel()} when invoking receive operations. This non-exclusive consumer is sometimes + * referred to as a "Non-Epoch Consumer."
        • + *
      * * @param startReadingAtEarliestEvent {@code true} to begin reading at the first events available in each * partition; otherwise, reading will begin at the end of each partition seeing only new events as they are 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 8273f5b0368a..04db34841c56 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 @@ -123,7 +123,7 @@ public void receiveAll() { * Receives from all partitions with last enqueued information. */ public void receiveLastEnqueuedInformation() { - // BEGIN: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receive#boolean-receiveoptions + // BEGIN: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receiveFromPartition#string-eventposition-receiveoptions ReceiveOptions receiveOptions = new ReceiveOptions() .setTrackLastEnqueuedEventProperties(true); EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() @@ -131,12 +131,14 @@ public void receiveLastEnqueuedInformation() { .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) .buildAsyncConsumerClient(); - // Receives events from all partitions as they come in. - consumer.receive(false, receiveOptions).subscribe(partitionEvent -> { - LastEnqueuedEventProperties properties = partitionEvent.getLastEnqueuedEventProperties(); - System.out.printf("Information received at %s. Sequence Id: %s%n", properties.getRetrievalTime(), - properties.getSequenceNumber()); - }); - // END: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receive#boolean-receiveoptions + // Receives events from partition "0" as they come in. + consumer.receiveFromPartition("0", EventPosition.earliest(), receiveOptions) + .subscribe(partitionEvent -> { + LastEnqueuedEventProperties properties = partitionEvent.getLastEnqueuedEventProperties(); + System.out.printf("Information received at %s. Last enqueued sequence number: %s%n", + properties.getRetrievalTime(), + properties.getSequenceNumber()); + }); + // END: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receiveFromPartition#string-eventposition-receiveoptions } } From c4b7c22b1fe826f30a08abc66e3cc474e67c952a Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 15:25:06 -0800 Subject: [PATCH 15/21] Fixing consumer documentation. --- .../EventHubConsumerAsyncClient.java | 2 ++ .../eventhubs/EventHubConsumerClient.java | 16 ++++++++----- .../EventHubConsumerJavaDocCodeSamples.java | 23 ++++++++++--------- 3 files changed, 24 insertions(+), 17 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 ff779be0e4df..2dbd09b4b6c6 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 @@ -137,6 +137,8 @@ public Flux getPartitionIds() { * @param partitionId The unique identifier of a partition associated with the Event Hub. * * @return The set of information for the requested partition under the Event Hub this client is associated with. + * + * @throws NullPointerException if {@code partitionId} is null. */ @ServiceMethod(returns = ReturnType.SINGLE) public Mono getPartitionProperties(String partitionId) { 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 4cb9344faf3d..bc985d7c54be 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 @@ -26,7 +26,6 @@ * a specific consumer group. * *

      Creating a synchronous consumer

      - *

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

      * {@codesnippet com.azure.messaging.eventhubs.eventhubconsumerclient.instantiation} * *

      Consuming events from a single partition

      @@ -56,7 +55,7 @@ public class EventHubConsumerClient implements Closeable { * Gets the fully qualified Event Hubs namespace that the connection is associated with. This is likely similar to * {@code {yournamespace}.servicebus.windows.net}. * - * @return The fully qualified Event Hubs namespace that the connection is associated with + * @return The fully qualified Event Hubs namespace that the connection is associated with. */ public String getFullyQualifiedNamespace() { return consumer.getFullyQualifiedNamespace(); @@ -93,7 +92,7 @@ public EventHubProperties getEventHubProperties() { /** * Retrieves the identifiers for the partitions of an Event Hub. * - * @return A Flux of identifiers for the partitions of an Event Hub. + * @return The set of identifiers for the partitions of an Event Hub. */ public IterableStream getPartitionIds() { return new IterableStream<>(consumer.getPartitionIds()); @@ -106,6 +105,8 @@ public IterableStream getPartitionIds() { * @param partitionId The unique identifier of a partition associated with the Event Hub. * * @return The set of information for the requested partition under the Event Hub this client is associated with. + * + * @throws NullPointerException if {@code partitionId} is null. */ @ServiceMethod(returns = ReturnType.SINGLE) public PartitionProperties getPartitionProperties(String partitionId) { @@ -123,7 +124,9 @@ public PartitionProperties getPartitionProperties(String partitionId) { * {@code maximumMessageCount} events. If a stream for the events was opened before, the same position within * that partition is returned. Otherwise, events are read starting from {@code startingPosition}. * - * @throws IllegalArgumentException if {@code maximumMessageCount} is less than 1. + * @throws NullPointerException if {@code partitionId}, or {@code startingPosition} is null. + * @throws IllegalArgumentException if {@code maximumMessageCount} is less than 1, or if {@code partitionId} is an + * empty string. */ public IterableStream receiveFromPartition(String partitionId, int maximumMessageCount, EventPosition startingPosition) { @@ -142,7 +145,8 @@ public IterableStream receiveFromPartition(String partitionId, i * @return A set of {@link PartitionEvent} that was received. The iterable contains up to * {@code maximumMessageCount} events. * - * @throws NullPointerException if {@code maximumWaitTime} or {@code eventPosition} is null. + * @throws NullPointerException if {@code partitionId}, {@code maximumWaitTime}, or {@code startingPosition} is + * {@code null}. * @throws IllegalArgumentException if {@code maximumMessageCount} is less than 1 or {@code maximumWaitTime} is * zero or a negative duration. */ @@ -188,7 +192,7 @@ public IterableStream receiveFromPartition(String partitionId, i * @return A set of {@link PartitionEvent} that was received. The iterable contains up to * {@code maximumMessageCount} events. * - * @throws NullPointerException if {@code maximumWaitTime}, {@code eventPosition}, {@code partitionId}, or + * @throws NullPointerException if {@code maximumWaitTime}, {@code startingPosition}, {@code partitionId}, or * {@code receiveOptions} is {@code null}. * @throws IllegalArgumentException if {@code maximumMessageCount} is less than 1 or {@code maximumWaitTime} is * zero or a negative duration. diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerJavaDocCodeSamples.java index 8f2e0fcb088b..cd1a1eeada2f 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerJavaDocCodeSamples.java @@ -43,28 +43,29 @@ public void receive() { // BEGIN: com.azure.messaging.eventhubs.eventhubconsumerclient.receive#string-int-eventposition-duration Instant twelveHoursAgo = Instant.now().minus(Duration.ofHours(12)); EventPosition startingPosition = EventPosition.fromEnqueuedTime(twelveHoursAgo); - - // Obtain partitionId from EventHubConsumerClient.getPartitionIds(). String partitionId = "0"; - // Begins reading events from partition '0' and returns the first 100 received or until the wait time of 30 - // seconds has elapsed. + // Reads events from partition '0' and returns the first 100 received or until the 30 seconds has elapsed. IterableStream events = consumer.receiveFromPartition(partitionId, 100, startingPosition, Duration.ofSeconds(30)); + Long lastSequenceNumber = -1L; for (PartitionEvent partitionEvent : events) { // For each event, perform some sort of processing. System.out.print("Event received: " + partitionEvent.getData().getSequenceNumber()); + lastSequenceNumber = partitionEvent.getData().getSequenceNumber(); } - // Gets the next set of events from partition '0' to consume and process. - IterableStream nextEvents = consumer.receiveFromPartition(partitionId, 100, - startingPosition, Duration.ofSeconds(30)); - // END: com.azure.messaging.eventhubs.eventhubconsumerclient.receive#string-int-eventposition-duration + // Figure out what the next EventPosition to receive from is based on last event we processed in the stream. + // If lastSequenceNumber is -1L, then we didn't see any events the first time we fetched events from the + // partition. + if (lastSequenceNumber != -1L) { + EventPosition nextPosition = EventPosition.fromSequenceNumber(lastSequenceNumber, false); - for (PartitionEvent partitionEvent : nextEvents) { - // For each event, perform some sort of processing. - System.out.print("Event received: " + partitionEvent.getData().getSequenceNumber()); + // Gets the next set of events from partition '0' to consume and process. + IterableStream nextEvents = consumer.receiveFromPartition(partitionId, 100, + nextPosition, Duration.ofSeconds(30)); } + // END: com.azure.messaging.eventhubs.eventhubconsumerclient.receive#string-int-eventposition-duration } } From 4f10f626e264a666efabf1b2a3a2917844ed5c51 Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 15:29:59 -0800 Subject: [PATCH 16/21] Fixing documentation in EHCB. --- .../azure/messaging/eventhubs/EventHubClientBuilder.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubClientBuilder.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubClientBuilder.java index 741c435cb82b..ac1d1366f5e2 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubClientBuilder.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubClientBuilder.java @@ -81,8 +81,10 @@ * * {@codesnippet com.azure.messaging.eventhubs.eventhubclientbuilder.instantiation} * - * @see EventHubClient - * @see EventHubAsyncClient + * @see EventHubProducerAsyncClient + * @see EventHubProducerClient + * @see EventHubConsumerAsyncClient + * @see EventHubConsumerClient */ @ServiceClientBuilder(serviceClients = {EventHubProducerAsyncClient.class, EventHubProducerClient.class, EventHubConsumerAsyncClient.class, EventHubConsumerClient.class}) From 66ea171219cad1c3b0e3d1a840f3163ef81843a7 Mon Sep 17 00:00:00 2001 From: Connie Yau Date: Tue, 26 Nov 2019 16:26:38 -0800 Subject: [PATCH 17/21] Update sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/ReceiveOptions.java Co-Authored-By: Srikanta <51379715+srnagar@users.noreply.github.com> --- .../com/azure/messaging/eventhubs/models/ReceiveOptions.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/ReceiveOptions.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/ReceiveOptions.java index c07acb714391..440807574741 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/ReceiveOptions.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/ReceiveOptions.java @@ -30,7 +30,7 @@ public Long getOwnerLevel() { } /** - * Sets the {@code ownerLevel} value on this recieve operation. When populated, the level indicates that the receive + * Sets the {@code ownerLevel} value on this receive operation. When populated, the level indicates that the receive * operation is intended to be the only reader of events for the requested partition and associated consumer group. * To do so, this receive operation will attempt to assert ownership over the partition; in the case where * there is more than one exclusive receive operation for the same partition/consumer group pair, the one having a From 37c1a1d8eaabf5d358d7dc36b5ce96f2292bb147 Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 20:35:20 -0800 Subject: [PATCH 18/21] Fix bug in README sample. --- sdk/eventhubs/azure-messaging-eventhubs/README.md | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/README.md b/sdk/eventhubs/azure-messaging-eventhubs/README.md index 8cd5b45b0ff7..6495f78540e2 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/README.md +++ b/sdk/eventhubs/azure-messaging-eventhubs/README.md @@ -190,15 +190,21 @@ 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 >>") .buildProducerClient(); + EventDataBatch eventDataBatch = producer.createBatch(); for (EventData eventData : eventDataList) { if (!eventDataBatch.tryAdd(eventData)) { producer.send(eventDataBatch); eventDataBatch = producer.createBatch(); + + // Try to add that event that couldn't fit before. + if (!eventDataBatch.tryAdd(eventData)) { + throw new IllegalArgumentException("Event is too large for an empty batch. Max size: " + + eventDataBatch.getMaxSizeInBytes()); + } } } From c921cf5c1fa6fa5b3d1d197964e3b9edf01d2074 Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 21:28:00 -0800 Subject: [PATCH 19/21] Fix consumer group name. --- ...ConsumerAsyncClientJavaDocCodeSamples.java | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) 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 04db34841c56..c39136ee3133 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 @@ -20,7 +20,7 @@ public class EventHubConsumerAsyncClientJavaDocCodeSamples { private final EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() .connectionString("fake-string") - .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .consumerGroup("consumer-group-name") .buildAsyncConsumerClient(); public void initialization() { @@ -29,7 +29,7 @@ public void initialization() { EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() .connectionString("Endpoint={fully-qualified-namespace};SharedAccessKeyName={policy-name};" + "SharedAccessKey={key};EntityPath={eh-name}") - .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .consumerGroup("consumer-group-name") .buildAsyncConsumerClient(); // END: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.instantiation @@ -71,7 +71,7 @@ public void receiveBackpressure() { EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() .connectionString("fake-string") - .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .consumerGroup("consumer-group-name") .buildAsyncConsumerClient(); // BEGIN: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receive#string-eventposition-basesubscriber @@ -106,7 +106,7 @@ protected void hookOnNext(PartitionEvent value) { public void receiveAll() { EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() .connectionString("fake-string") - .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .consumerGroup("consumer-group-name") .buildAsyncConsumerClient(); // BEGIN: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receive#boolean @@ -123,14 +123,17 @@ public void receiveAll() { * Receives from all partitions with last enqueued information. */ public void receiveLastEnqueuedInformation() { - // BEGIN: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receiveFromPartition#string-eventposition-receiveoptions - ReceiveOptions receiveOptions = new ReceiveOptions() - .setTrackLastEnqueuedEventProperties(true); EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() .connectionString("event-hub-instance-connection-string") - .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .consumerGroup("consumer-group-name") .buildAsyncConsumerClient(); + // BEGIN: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receiveFromPartition#string-eventposition-receiveoptions + // Set `setTrackLastEnqueuedEventProperties` to true to get the last enqueued information from the partition for + // each event that is received. + ReceiveOptions receiveOptions = new ReceiveOptions() + .setTrackLastEnqueuedEventProperties(true); + // Receives events from partition "0" as they come in. consumer.receiveFromPartition("0", EventPosition.earliest(), receiveOptions) .subscribe(partitionEvent -> { From 9e28bf7d4e4fcdebb11cd2301f1147f30813766c Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 21:28:15 -0800 Subject: [PATCH 20/21] Fix readme --- sdk/eventhubs/azure-messaging-eventhubs/README.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/README.md b/sdk/eventhubs/azure-messaging-eventhubs/README.md index 6495f78540e2..bd1efca45f11 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/README.md +++ b/sdk/eventhubs/azure-messaging-eventhubs/README.md @@ -189,13 +189,14 @@ The snippet below creates a synchronous producer and sends events to any partiti 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 >>") .buildProducerClient(); +List allEvents = Arrays.asList(new EventData("Foo"), new EventData("Bar")); EventDataBatch eventDataBatch = producer.createBatch(); -for (EventData eventData : eventDataList) { + +for (EventData eventData : allEvents) { if (!eventDataBatch.tryAdd(eventData)) { producer.send(eventDataBatch); eventDataBatch = producer.createBatch(); From 464de74459ea03056c18ebbe02563c86ee887e12 Mon Sep 17 00:00:00 2001 From: Connie Date: Tue, 26 Nov 2019 22:46:06 -0800 Subject: [PATCH 21/21] Fix blocking samples. --- .../EventHubConsumerJavaDocCodeSamples.java | 2 +- ...ProducerAsyncClientJavaDocCodeSamples.java | 23 +++++++++++++++---- .../PublishEventsToSpecificPartition.java | 21 ++++++++++------- .../PublishEventsWithAzureIdentity.java | 17 +++++++++----- .../PublishEventsWithCustomMetadata.java | 17 +++++++++----- 5 files changed, 55 insertions(+), 25 deletions(-) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerJavaDocCodeSamples.java index cd1a1eeada2f..49bb72f61efb 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerJavaDocCodeSamples.java @@ -37,7 +37,7 @@ public void instantiate() { public void receive() { EventHubConsumerClient consumer = new EventHubClientBuilder() .connectionString("event-hub-instance-connection-string") - .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .consumerGroup("consumer-group-name") .buildConsumerClient(); // BEGIN: com.azure.messaging.eventhubs.eventhubconsumerclient.receive#string-int-eventposition-duration diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java index 8813da500ae5..ad5b804b1345 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientJavaDocCodeSamples.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 reactor.core.publisher.Mono; @@ -52,7 +53,10 @@ public void batchAutomaticRouting() { batch.tryAdd(new EventData("test-event-1")); batch.tryAdd(new EventData("test-event-2")); return producer.send(batch); - }).block(); + }).subscribe(unused -> { + }, + error -> System.err.println("Error occurred while sending batch:" + error), + () -> System.out.println("Send complete.")); // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch producer.close(); @@ -71,7 +75,10 @@ public void batchPartitionId() { batch.tryAdd(new EventData("test-event-1")); batch.tryAdd(new EventData("test-event-2")); return producer.send(batch); - }).block(); + }).subscribe(unused -> { + }, + error -> System.err.println("Error occurred while sending batch:" + error), + () -> System.out.println("Send complete.")); // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionId producer.close(); @@ -91,7 +98,10 @@ public void batchPartitionKey() { batch.tryAdd(new EventData("sourdough")); batch.tryAdd(new EventData("rye")); return producer.send(batch); - }).block(); + }).subscribe(unused -> { + }, + error -> System.err.println("Error occurred while sending batch:" + error), + () -> System.out.println("Send complete.")); // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-partitionKey } @@ -126,8 +136,13 @@ public void batchSizeLimited() { producer.send(batch), producer.createBatch(options).map(newBatch -> { currentBatch.set(newBatch); + // Add the event that did not fit in the previous batch. - newBatch.tryAdd(event); + if (!newBatch.tryAdd(event)) { + throw Exceptions.propagate(new IllegalArgumentException( + "Event was too large to fit in an empty batch. Max size: " + newBatch.getMaxSizeInBytes())); + } + return newBatch; })); }).then() 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 19ef651f105b..3acc060d8b7a 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 @@ -8,13 +8,14 @@ import reactor.core.publisher.Mono; import java.time.Duration; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import static java.nio.charset.StandardCharsets.UTF_8; /** - * Sample demonstrates how to sent events to specific event hub by defining partition id using - * {@link CreateBatchOptions#setPartitionId(String)}. + * 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); @@ -59,7 +60,7 @@ 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. - final Mono sendOperation = data.flatMap(event -> { + data.flatMap(event -> { final EventDataBatch batch = currentBatch.get(); if (batch.tryAdd(event)) { return Mono.empty(); @@ -87,13 +88,17 @@ public static void main(String[] args) { if (batch != null) { producer.send(batch).block(OPERATION_TIMEOUT); } - }); + }) + .subscribe(unused -> System.out.println("Complete"), + error -> System.out.println("Error sending events: " + error), + () -> System.out.println("Completed sending events.")); - // 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. + // The .subscribe() creation and assignment is not a blocking call. For the purpose of this example, we sleep + // the thread so the program does not end before the send operation is complete. Using .block() instead of + // .subscribe() will turn this into a synchronous call. try { - sendOperation.block(OPERATION_TIMEOUT); + TimeUnit.SECONDS.sleep(5); + } catch (InterruptedException ignored) { } 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 c89383280c20..775b5b92b295 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 @@ -9,6 +9,7 @@ import reactor.core.publisher.Mono; import java.time.Duration; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import static java.nio.charset.StandardCharsets.UTF_8; @@ -57,7 +58,7 @@ 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. - final Mono sendOperation = telemetryEvents.flatMap(event -> { + telemetryEvents.flatMap(event -> { final EventDataBatch batch = currentBatch.get(); if (batch.tryAdd(event)) { return Mono.empty(); @@ -85,13 +86,17 @@ public static void main(String[] args) { if (batch != null) { producer.send(batch).block(OPERATION_TIMEOUT); } - }); + }) + .subscribe(unused -> System.out.println("Complete"), + error -> System.out.println("Error sending events: " + error), + () -> System.out.println("Completed sending events.")); - // 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. + // The .subscribe() creation and assignment is not a blocking call. For the purpose of this example, we sleep + // the thread so the program does not end before the send operation is complete. Using .block() instead of + // .subscribe() will turn this into a synchronous call. try { - sendOperation.block(OPERATION_TIMEOUT); + TimeUnit.SECONDS.sleep(5); + } catch (InterruptedException ignored) { } 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 3da7d8128637..e5952a455ad7 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 @@ -7,6 +7,7 @@ import reactor.core.publisher.Mono; import java.time.Duration; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import static java.nio.charset.StandardCharsets.UTF_8; @@ -64,7 +65,7 @@ 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. - final Mono sendOperation = data.flatMap(event -> { + data.flatMap(event -> { final EventDataBatch batch = currentBatch.get(); if (batch.tryAdd(event)) { return Mono.empty(); @@ -92,13 +93,17 @@ public static void main(String[] args) { if (batch != null) { producer.send(batch).block(OPERATION_TIMEOUT); } - }); + }) + .subscribe(unused -> System.out.println("Complete"), + error -> System.out.println("Error sending events: " + error), + () -> System.out.println("Completed sending events.")); - // 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. + // The .subscribe() creation and assignment is not a blocking call. For the purpose of this example, we sleep + // the thread so the program does not end before the send operation is complete. Using .block() instead of + // .subscribe() will turn this into a synchronous call. try { - sendOperation.block(OPERATION_TIMEOUT); + TimeUnit.SECONDS.sleep(5); + } catch (InterruptedException ignored) { } finally { // Disposing of our producer. producer.close();