diff --git a/sdk/eventhubs/azure-messaging-eventhubs/README.md b/sdk/eventhubs/azure-messaging-eventhubs/README.md index 8cd5b45b0ff7..bd1efca45f11 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/README.md +++ b/sdk/eventhubs/azure-messaging-eventhubs/README.md @@ -189,16 +189,23 @@ 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(); + + // 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()); + } } } 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..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 @@ -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".

@@ -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; @@ -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); @@ -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. @@ -178,8 +179,9 @@ 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. 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/EventHubClientBuilder.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubClientBuilder.java index 0d1e0f5c561d..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 @@ -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 @@ -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}) 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..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 @@ -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; @@ -33,21 +32,21 @@ * 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} * - *

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} @@ -138,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) { @@ -171,10 +172,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."
    • *
* @@ -208,17 +209,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); @@ -227,15 +227,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 @@ -248,17 +246,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/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/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..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 @@ -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: @@ -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 @@ -71,38 +71,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 EventHubProducerAsyncClient}. - * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.instantiation} + *

Create a producer and publish events to any partition

+ * {@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 - * {@link CreateBatchOptions#setPartitionId(String) partitionId} when creating an {@link EventHubProducerAsyncClient}. + *

Publish events to partition "foo"

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

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

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

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. - *

- * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#publisher-sendOptions} + *

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

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

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()}. - *

- * {@codesnippet com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#eventDataBatch} + * @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 { @@ -148,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; @@ -189,6 +171,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) { @@ -205,11 +189,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) { @@ -262,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. *

* @@ -284,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. *

* @@ -507,7 +493,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/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..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 @@ -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,8 +28,7 @@ *

* *

- * If no {@link SendOptions#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 - * SendOptions#getPartitionId() partitionId} when creating the {@link EventHubProducerClient}. + *

Create a producer and publish events to any partition

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

Publish events to partition "foo"

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

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}. + *

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

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

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

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

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. - *

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

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()}. - *

- * {@codesnippet com.azure.messaging.eventhubs.eventhubproducerclient.send#eventDataBatch} - * - * @see EventHubClient#createProducer() - * @see EventHubProducerAsyncClient To asynchronously generate events to an Event Hub, see EventHubAsyncProducer. + * @see EventHubClientBuilder#buildProducerClient() + * @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. *

* 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; 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..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,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 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 + * 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 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..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 @@ -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 - * sent using the same exact partition hashing key, the Event Hubs service will route them all to the same - * 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 + * 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. 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..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 @@ -18,6 +18,10 @@ * Code snippets demonstrating various {@link EventHubConsumerAsyncClient} scenarios. */ public class EventHubConsumerAsyncClientJavaDocCodeSamples { + private final EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() + .connectionString("fake-string") + .consumerGroup("consumer-group-name") + .buildAsyncConsumerClient(); public void initialization() { // BEGIN: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.instantiation @@ -25,19 +29,17 @@ 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 + + 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() @@ -69,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 @@ -104,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 @@ -121,20 +123,25 @@ public void receiveAll() { * Receives from all partitions with last enqueued information. */ public void receiveLastEnqueuedInformation() { - // BEGIN: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.receive#boolean-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(); - // 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 + // 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 -> { + 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 } } 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..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 @@ -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}") @@ -37,34 +37,35 @@ 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 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 } } 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..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,8 +4,9 @@ package com.azure.messaging.eventhubs; import com.azure.messaging.eventhubs.models.CreateBatchOptions; -import com.azure.messaging.eventhubs.models.SendOptions; +import reactor.core.Exceptions; 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 @@ -37,19 +36,50 @@ public void instantiate() { } /** - * Code snippet demonstrating how to create an {@link EventHubProducerAsyncClient} that routes 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.eventhubasyncproducerclient.instantiation#partitionId - EventData eventData = new EventData("data-to-partition-foo"); - SendOptions options = new SendOptions() - .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. + 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); + }).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(); + } + + /** + * Code snippet demonstrating how to create an EventDataBatch at routes events to a single partition. + */ + public void batchPartitionId() { 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); + }).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(); } @@ -57,64 +87,71 @@ 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)) - ); - + public void batchPartitionKey() { 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), + // 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("bread"); + producer.createBatch(options).flatMap(batch -> { + batch.tryAdd(new EventData("sourdough")); + batch.tryAdd(new EventData("rye")); + return producer.send(batch); + }).subscribe(unused -> { + }, + error -> System.err.println("Error occurred while sending batch:" + error), () -> System.out.println("Send complete.")); - // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#publisher-sendOptions + // 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() { + public void batchSizeLimited() { final EventHubProducerAsyncClient producer = builder.buildAsyncProducerClient(); - - // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.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"); - - final Flux telemetryEvents = Flux.just(firstEvent, secondEvent, thirdEvent); + // 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. + // 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. - telemetryEvents.subscribe(event -> { + // 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)) { + 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. + 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() + .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-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..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 @@ -4,9 +4,7 @@ package com.azure.messaging.eventhubs; 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; @@ -18,15 +16,18 @@ 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") + .connectionString( + "Endpoint={fully-qualified-namespace};SharedAccessKeyName={policy-name};SharedAccessKey={key}", + "event-hub-name") .buildProducerClient(); // END: com.azure.messaging.eventhubs.eventhubproducerclient.instantiation @@ -34,61 +35,97 @@ 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. * - * @throws IOException if the producer cannot be disposed. + * @throws IllegalArgumentException if an event is too large for an empty batch. */ - public void instantiatePartitionProducer() throws IOException { - // BEGIN: com.azure.messaging.eventhubs.eventhubproducerclient.instantiation#partitionId - EventData eventData = new EventData("data-to-partition-foo"); - SendOptions options = new SendOptions() - .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")); - EventHubProducerClient producer = builder.buildProducerClient(); - producer.send(eventData, options); - // END: com.azure.messaging.eventhubs.eventhubproducerclient.instantiation#partitionId + // 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 send events with a partition key. + * Code snippet demonstrating how to create an EventDataBatch at routes events to a single partition. */ - 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 batchPartitionId() { 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-partitionId + // Creating a batch with partitionId set will route all events in that batch to partition `foo`. + CreateBatchOptions options = new CreateBatchOptions().setPartitionId("foo"); + + EventDataBatch batch = producer.createBatch(options); + batch.tryAdd(new EventData("data-to-partition-foo")); + producer.send(batch); + // END: com.azure.messaging.eventhubs.eventhubproducerclient.createBatch#CreateBatchOptions-partitionId } /** - * Code snippet demonstrating how to create an {@link EventDataBatch} and send it. + * Code snippet demonstrating how to send events with a partition key. */ - public void sendEventDataBatch() { + public void batchPartitionKey() { final EventHubProducerClient producer = builder.buildProducerClient(); - // BEGIN: com.azure.messaging.eventhubs.eventhubproducerclient.send#eventDataBatch + // 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 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(); 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 +136,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 } } 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();