diff --git a/sdk/eventhubs/azure-messaging-eventhubs/README.md b/sdk/eventhubs/azure-messaging-eventhubs/README.md index 02dd87c2f5e0..8cd5b45b0ff7 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/README.md +++ b/sdk/eventhubs/azure-messaging-eventhubs/README.md @@ -23,17 +23,17 @@ documentation][event_hubs_product_docs] | [Samples][sample_examples] - [Table of contents](#table-of-contents) - [Getting started](#getting-started) - - [Default SSL](#default-ssl-library) - [Prerequisites](#prerequisites) - [Adding the package to your product](#adding-the-package-to-your-product) + - [Default SSL library](#default-ssl-library) - [Methods to authorize with Event Hubs](#methods-to-authorize-with-event-hubs) - - [Create an Event Hub client using a connection string](#create-an-event-hub-client-using-a-connection-string) + - [Create an Event Hub producer using a connection string](#create-an-event-hub-producer-using-a-connection-string) - [Create an Event Hub client using Microsoft identity platform (formerly Azure Active Directory)](#create-an-event-hub-client-using-microsoft-identity-platform-formerly-azure-active-directory) - [Key concepts](#key-concepts) - [Examples](#examples) - [Publish events to an Event Hub](#publish-events-to-an-event-hub) - [Consume events from an Event Hub partition](#consume-events-from-an-event-hub-partition) - - [Consume events using an Event Processor](#consume-events-using-an-event-processor) + - [Consume events using an EventProcessorClient](#consume-events-using-an-eventprocessorclient) - [Troubleshooting](#troubleshooting) - [Enable client logging](#enable-client-logging) - [Enable AMQP transport logging](#enable-amqp-transport-logging) @@ -44,17 +44,14 @@ documentation][event_hubs_product_docs] | [Samples][sample_examples] ## Getting started -### Default SSL library -All client libraries, by default, use the Tomcat-native Boring SSL library to enable native-level performance for SSL operations. The Boring SSL library is an uber jar containing native libraries for Linux / macOS / Windows, and provides better performance compared to the default SSL implementation within the JDK. For more information, including how to reduce the dependency size, refer to the [performance tuning][performance_tuning] section of the wiki. - ### Prerequisites - Java Development Kit (JDK) with version 8 or above - [Maven][maven] - Microsoft Azure subscription - - You can create a free account at: https://azure.microsoft.com + - You can create a free account at: https://azure.microsoft.com - Azure Event Hubs instance - - Step-by-step guide for [creating an Event Hub using the Azure Portal][event_hubs_create] + - Step-by-step guide for [creating an Event Hub using the Azure Portal][event_hubs_create] ### Adding the package to your product @@ -68,6 +65,13 @@ All client libraries, by default, use the Tomcat-native Boring SSL library to en ``` [//]: # ({x-version-update-end}) +### Default SSL library +All client libraries, by default, use the Tomcat-native Boring SSL library to enable native-level +performance for SSL operations. The Boring SSL library is an uber jar containing native libraries +for Linux/macOS/Windows, and provides better performance compared to the default SSL +implementation within the JDK. For more information, including how to reduce the dependency size, +refer to the [performance tuning][performance_tuning] section of the wiki. + ### Methods to authorize with Event Hubs For the Event Hubs client library to interact with an Event Hub, it will need to understand how to connect and authorize @@ -79,9 +83,10 @@ The easiest means for doing so is to use a connection string, which is created a namespace. If you aren't familiar with shared access policies in Azure, you may wish to follow the step-by-step guide to [get an Event Hubs connection string][event_hubs_connection_string]. -Both the asynchronous and synchronous Event Hub producer and consumer clients can be created using -`EventHubClientBuilder`. Invoking `.buildAsyncProducer()` and `buildProducer` will build the asynchronous and -synchronous producers. Similarly, `.buildAsyncConsumer` and `.buildConsumer` will build the appropriate consumers. +Both the asynchronous and synchronous Event Hub producer and consumer clients can be created using +`EventHubClientBuilder`. Invoking `buildAsyncProducerClient()` or `buildProducerClient()` will build the asynchronous or +synchronous producers. Similarly, `buildAsyncConsumerClient()` or `buildConsumerClient()` will build the appropriate +consumers. The snippet below creates a synchronous Event Hub producer. @@ -90,7 +95,7 @@ String connectionString = "<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>" String eventHubName = "<< NAME OF THE EVENT HUB >>"; EventHubProducerClient producer = new EventHubClientBuilder() .connectionString(connectionString, eventHubName) - .buildProducer(); + .buildProducerClient(); ``` ### Create an Event Hub client using Microsoft identity platform (formerly Azure Active Directory) @@ -127,11 +132,11 @@ ClientSecretCredential credential = new ClientSecretCredentialBuilder() // The fully qualified namespace for the Event Hubs instance. This is likely to be similar to: // {your-namespace}.servicebus.windows.net -String fullyQualifiedNamespace = "<< EVENT HUBS FULLY QUALIFIED NAMESPACE >>" +String fullyQualifiedNamespace = "my-test-eventhubs.servicebus.windows.net"; String eventHubName = "<< NAME OF THE EVENT HUB >>"; EventHubProducerClient client = new EventHubClientBuilder() .credential(fullyQualifiedNamespace, eventHubName, credential) - .buildProducer(); + .buildProducerClient(); ``` ## Key concepts @@ -162,38 +167,33 @@ are well documented in [OASIS Advanced Messaging Queuing Protocol (AMQP) Version ## Examples -- [Inspect Event Hub and partition properties][sample_get_event_hubs_metadata] -- [Publish an EventDataBatch to an Event Hub][sample_publish_eventdatabatch] -- [Consume events from an Event Hub partition][sample_consume_event] -- [Consume events from all Event Hub partitions][sample_event_processor] - ### Publish events to an Event Hub -In order to publish events, you'll need to create an asynchronous -[`EventHubProducerAsyncClient`][source_eventhubasyncproducerclient] or a synchronous -[`EventHubProducerClient`][source_eventHubProducerClient]. -Each producers can send events to either, a specific partition, or allow the Event Hubs service to decide which -partition events should be published to. It is recommended to use automatic routing when the publishing of events needs -to be highly available or when event data should be distributed evenly among the partitions. In the our example, we will -take advantage of automatic routing. +To publish events, you'll need to create an asynchronous [`EventHubProducerAsyncClient`][EventHubProducerAsyncClient] or +a synchronous [`EventHubProducerClient`][EventHubProducerClient]. Each producer can send events to either, a specific +partition, or allow the Event Hubs service to decide which partition events should be published to. It is recommended to +use automatic routing when the publishing of events needs to be highly available or when event data should be +distributed evenly among the partitions. -#### Event Hub producer creation +#### Create an Event Hub producer and publish events -Developers can create a producer by calling `buildProducer()` or `buildAsyncProducer`. If `buildProducer` is invoked, a -synchronous `EventHubProducerClient` is created. If `buildAsyncProducer` is used, an asynchronous -`EventHubProducerAsyncClient` is returned. +Developers can create a producer by calling `buildAsyncProducerClient()` or `buildProducerClient()`. If +`buildProducerClient()` is invoked, a synchronous `EventHubProducerClient` is created. If `buildAsyncProducerClient()` +is used, an asynchronous `EventHubProducerAsyncClient` is returned. -Specifying `batchOptions.partitionId(String)` will send events to a specific partition, and not, will allow for automatic -routing. In addition, specifying `partitionKey(String)` will tell Event Hubs service to hash the events and send them to -the same partition. +Specifying `CreateBatchOptions.setPartitionId(String)` will send events to a specific partition. If not specified, will +allow for automatic partition routing. In addition, specifying `CreateBatchOptions.setPartitionKey(String)` will tell +Event Hubs service to hash the events and send them to the same partition. -The snippet below creates a synchronous producer and sends events to any partition, allowing Event Hubs service to route the event -to an available partition. +The snippet below creates a synchronous producer and sends events to any partition, allowing Event Hubs service to route +the event to an available partition. ```java +List eventDataList = Arrays.asList(new EventData("Foo"), new EventData("Bar")); + EventHubProducerClient producer = new EventHubClientBuilder() .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") - .buildProducer(); + .buildProducerClient(); EventDataBatch eventDataBatch = producer.createBatch(); for (EventData eventData : eventDataList) { if (!eventDataBatch.tryAdd(eventData)) { @@ -203,100 +203,108 @@ for (EventData eventData : eventDataList) { } // send the last batch of remaining events -if (eventDataBatch.getSize() > 0) { +if (eventDataBatch.getCount() > 0) { producer.send(eventDataBatch); } ``` -To send events to a particular partition, set the optional parameter `partitionId` on -[`CreateBatchOptions`][source_CreateBatchOptions]. +To send events to a particular partition, set the optional parameter `setPartitionId(String)` on +[`CreateBatchOptions`][CreateBatchOptions]. -#### Partition identifier +#### Publish events using partition identifier Many Event Hub operations take place within the scope of a specific partition. Because partitions are owned by the Event -Hub, their names are assigned at the time of creation. To understand what partitions are available, You can use the -`getPartitionIds` function to get the ids of all available partitions in your Event Hub instance. +Hub, their names are assigned at the time of creation. To understand what partitions are available, you can use the +`getPartitionIds()` function to get the ids of all available partitions in your Event Hub instance. All clients created +using `EventHubsClientBuilder` can query for metadata about the Event Hub using `getPartitionIds()` or +`getEventHubProperties()`. ```java -IterableStream partitionIds = client.getPartitionIds(); +EventHubProducerClient producer = new EventHubClientBuilder() + .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") + .buildProducerClient(); + +CreateBatchOptions options = new CreateBatchOptions().setPartitionId("0"); +EventDataBatch batch = producer.createBatch(options); + +// Add events to batch and when you want to send the batch, send it using the producer. +producer.send(batch); ``` -#### Partition key +#### Publish events using partition key When an Event Hub producer is not associated with any specific partition, it may be desirable to request that the Event Hubs service keep different events or batches of events together on the same partition. This can be accomplished by setting a `partition key` when publishing the events. ```java -CreateBatchOptions batchOptions = new CreateBatchOptions() - .partitionKey("grouping-key"); +CreateBatchOptions batchOptions = new CreateBatchOptions().setPartitionKey("grouping-key"); EventDataBatch eventDataBatch = producer.createBatch(batchOptions); -// add events to eventDataBatch + +// Add events to batch and when you want to send the batch, send it using the producer. producer.send(eventDataBatch); ``` ### Consume events from an Event Hub partition -In order to consume events, you'll need to create an `EventHubAsyncConsumer` or `EventHubConsumer` for a specific -partition and consumer group combination. When an Event Hub is created, it starts with a default consumer group that can -be used to get started. A consumer also needs to specify where in the event stream to begin receiving events; in our -example, we will focus on reading new events as they are published. +In order to consume events, you'll need to create an [`EventHubConsumerAsyncClient`][EventHubConsumerAsyncClient] or +[`EventHubConsumerClient`][EventHubConsumerClient] for a specific consumer group. When an Event Hub is created, it +starts with a default consumer group that can be used to get started. A consumer also needs to specify where in the +event stream to begin receiving events. -#### Consume events with EventHubAsyncConsumer +#### Consume events with EventHubConsumerAsyncClient In the snippet below, we are creating an asynchronous consumer that receives events from `partitionId` and only listens -to newest events that get pushed to the partition using `receive(String partitionId)`. Developers can begin receiving -events by calling `.receive(String partitionId)` and subscribing to the stream. +to newest events that get pushed to the partition by invoking `receiveFromPartition(String, EventPosition)`. Developers +can begin receiving events from multiple partitions using the same EventHubConsumerAsyncClient by calling +`receiveFromPartition(String, EventPosition)` with another partition id, and subscribing to that Flux. ```java -EventHubAsyncConsumer client = new EventHubClientBuilder() +EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") - .consumerGroup(EventHubAsyncClient.DEFAULT_CONSUMER_GROUP_NAME) - .startingPosition(EventPosition.earliest()) - .buildAsyncConsumer(); + .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .buildAsyncConsumerClient(); -consumer.receive(partitionId).subscribe(event -> { +// Receive events from partition with id "0", only getting events that are newly added to the partition. +consumer.receiveFromPartition("0", EventPosition.latest()).subscribe(event -> { // Process each event as it arrives. }); ``` -#### Consume events with EventHubConsumer +#### Consume events with EventHubConsumerClient -Developers can create a synchronous consumer that returns events in batches using an `EventHubAsyncConsumer`. In the +Developers can create a synchronous consumer that returns events in batches using an `EventHubConsumerClient`. In the snippet below, a consumer is created that starts reading events from the beginning of the partition's event stream. ```java -EventHubConsumer client = new EventHubClientBuilder() +EventHubConsumerClient consumer = new EventHubClientBuilder() .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") - .consumerGroup(EventHubAsyncClient.DEFAULT_CONSUMER_GROUP_NAME) - .startingPosition(EventPosition.earliest()) - .buildConsumer(); + .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .buildConsumerClient(); String partitionId = "<< EVENT HUB PARTITION ID >>"; // Get the first 15 events in the stream, or as many events as can be received within 40 seconds. -IterableStream events = consumer.receive(partitionId, 15, Duration.ofSeconds(40)); -for (EventData event : events) { - // Process each event +IterableStream events = consumer.receiveFromPartition(partitionId, 15, + EventPosition.earliest(), Duration.ofSeconds(40)); +for (PartitionEvent event : events) { + System.out.println("Event: " + event.getData().getBodyAsString()); } - -// Calling receive again returns the next 15 events in the stream, or as many as possible in 40 seconds. -IterableStream nextEvents = consumer.receive(partitionId, 15, Duration.ofSeconds(40)); ``` -### Consume events using an Event Processor Client +### Consume events using an EventProcessorClient -To consume events for all partitions of an Event Hub, you'll create an [`EventProcessorClient`][source_eventprocessorclient] for a -specific consumer group. When an Event Hub is created, it provides a default consumer group that can be used to get -started. +To consume events for all partitions of an Event Hub, you'll create an [`EventProcessorClient`][EventProcessorClient] +for a specific consumer group. When an Event Hub is created, it provides a default consumer group that can be used to +get started. -The [`EventProcessorClient`][source_eventprocessorclient] will delegate processing of events to a callback function that you -provide, allowing you to focus on the logic needed to provide value while the processor holds responsibility for +The [`EventProcessorClient`][EventProcessorClient] will delegate processing of events to a callback function that you +provide, allowing you to focus on the logic needed to provide value while the processor holds responsibility for managing the underlying consumer operations. -In our example, we will focus on building the [`EventProcessorClient`][source_eventprocessorclient], use the -[`InMemoryCheckpointStore`][source_inmemorycheckpointstore] available in samples, and a callback function that -processes events received from the Event Hub and writes to console. +In our example, we will focus on building the [`EventProcessorClient`][EventProcessorClient], use the +[`InMemoryCheckpointStore`][InMemoryCheckpointStore] available in samples, and a callback function that processes events +received from the Event Hub and writes to console. ```java class Program { @@ -316,9 +324,9 @@ class Program { // This will start the processor. It will start processing events from all partitions. eventProcessorClient.start(); - + // (for demo purposes only - adding sleep to wait for receiving events) - TimeUnit.SECONDS.sleep(2); + TimeUnit.SECONDS.sleep(2); // When the user wishes to stop processing events, they can call `stop()`. eventProcessorClient.stop(); @@ -332,7 +340,7 @@ class Program { You can set the `AZURE_LOG_LEVEL` environment variable to view logging statements made in the client library. For example, setting `AZURE_LOG_LEVEL=2` would show all informational, warning, and error log messages. The log levels can -be found here: [log levels][source_loglevels]. +be found here: [log levels][LogLevels]. ### Enable AMQP transport logging @@ -364,17 +372,17 @@ This is a general exception for AMQP related failures, which includes the AMQP e that caused this exception as ErrorContext. 'isTransient' is a boolean indicating if the exception is a transient error or not. If true, then the request can be retried; otherwise not. -The [ErrorCondition][source_errorcondition] contains error conditions common to the AMQP protocol and used by Azure +[`AmqpErrorCondition`][AmqpErrorCondition] contains error conditions common to the AMQP protocol and used by Azure services. When an AMQP exception is thrown, examining the error condition field can inform developers as to why the AMQP exception occurred and if possible, how to mitigate this exception. A list of all the AMQP exceptions can be found in [OASIS AMQP Version 1.0 Transport Errors][oasis_amqp_v1_error]. -The [ErrorContext][source_errorcontext] in the [AmqpException][source_amqpexception] provides information about the AMQP +The [`AmqpErrorContext`][AmqpErrorContext] in the [`AmqpException`][AmqpException] provides information about the AMQP session, link, or connection that the exception occurred in. This is useful to diagnose which level in the transport this exception occurred at and whether it was an issue in one of the producers or consumers. -The recommended way to solve the specific exception the AMQP exception represents is to follow the [Event Hubs Messaging -Exceptions][event_hubs_messaging_exceptions] guidance. +The recommended way to solve the specific exception the AMQP exception represents is to follow the +[Event Hubs Messaging Exceptions][event_hubs_messaging_exceptions] guidance. #### Operation cancelled exception @@ -400,12 +408,14 @@ advantage of the full feature set of the Azure Event Hubs service. In order to h the following set of sample is available: - [Inspect Event Hub and partition properties][sample_get_event_hubs_metadata] -- [Publish an event batch to an Event Hub][sample_publish_eventdatabatch] +- [Publish events using Microsoft identity platform][sample_publish_identity] - [Publish events to a specific Event Hub partition with partition identifier][sample_publish_partitionId] - [Publish events to a specific Event Hub partition with partition key][sample_publish_partitionKey] +- [Publish events to an Event Hub with a size-limited batch][sample_publish_size_limited] - [Publish events with custom metadata][sample_publish_custom_metadata] - [Consume events from an Event Hub partition][sample_consume_event] -- [Save the last read event and resume from that point][sample_sequence_number] +- [Consume events starting from an event sequence number][sample_consume_sequence_number] +- [Consume events from all partitions using EventProcessorClient][sample_event_processor] ## Contributing @@ -429,26 +439,27 @@ Guidelines](./CONTRIBUTING.md) for more information. [oasis_amqp_v1]: http://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-overview-v1.0-os.html [performance_tuning]: https://github.com/Azure/azure-sdk-for-java/wiki/Performance-Tuning [qpid_proton_j_apache]: http://qpid.apache.org/proton/ -[sample_consume_event]: ./src/samples/java/com/azure/messaging/eventhubs/ConsumeEvent.java -[sample_event_processor]: ./src/samples/java/com/azure/messaging/eventhubs/EventProcessorSample.java [sample_examples]: ./src/samples/java/com/azure/messaging/eventhubs/ +[sample_consume_event]: ./src/samples/java/com/azure/messaging/eventhubs/ConsumeEvents.java +[sample_consume_sequence_number]: ./src/samples/java/com/azure/messaging/eventhubs/ConsumeEventsFromKnownSequenceNumberPosition.java +[sample_event_processor]: ./src/samples/java/com/azure/messaging/eventhubs/EventProcessorSample.java [sample_get_event_hubs_metadata]: ./src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java [sample_publish_custom_metadata]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java -[sample_publish_eventdatabatch]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventDataBatch.java +[sample_publish_identity]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java [sample_publish_partitionId]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java [sample_publish_partitionKey]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java -[sample_sequence_number]: ./src/samples/java/com/azure/messaging/eventhubs/ConsumeEventsFromKnownSequenceNumberPosition.java -[source_amqpexception]: ../../core/azure-core-amqp/src/main/java/com/azure/core/amqp/exception/AmqpException.java +[sample_publish_size_limited]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java [source_code]: ./ -[source_errorcondition]: ../../core/azure-core-amqp/src/main/java/com/azure/core/amqp/exception/ErrorCondition.java -[source_errorcontext]: ../../core/azure-core-amqp/src/main/java/com/azure/core/amqp/exception/ErrorContext.java -[source_eventhubasyncclient]: ./src/main/java/com/azure/messaging/eventhubs/EventHubAsyncClient.java -[source_eventhubasyncproducerclient]: ./src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java -[source_eventhubclient]: ./src/main/java/com/azure/messaging/eventhubs/EventHubClient.java -[source_eventHubProducerClient]: ./src/main/java/com/azure/messaging/eventhubs/EventHubProducerClient.java -[source_eventprocessorclient]: ./src/main/java/com/azure/messaging/eventhubs/EventProcessorClient.java -[source_CreateBatchOptions]: ./src/main/java/com/azure/messaging/eventhubs/models/CreateBatchOptions.java -[source_inmemorycheckpointstore]: ./src/samples/java/com/azure/messaging/eventhubs/InMemoryCheckpointStore.java -[source_loglevels]: ../../core/azure-core/src/main/java/com/azure/core/util/logging/ClientLogger.java +[AmqpException]: ../../core/azure-core-amqp/src/main/java/com/azure/core/amqp/exception/AmqpException.java +[AmqpErrorCondition]: ../../core/azure-core-amqp/src/main/java/com/azure/core/amqp/exception/AmqpErrorCondition.java +[AmqpErrorContext]: ../../core/azure-core-amqp/src/main/java/com/azure/core/amqp/exception/AmqpErrorContext.java +[EventHubConsumerAsyncClient]: ./src/main/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClient.java +[EventHubConsumerClient]: ./src/main/java/com/azure/messaging/eventhubs/EventHubConsumerClient.java +[EventHubProducerAsyncClient]: ./src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java +[EventHubProducerClient]: ./src/main/java/com/azure/messaging/eventhubs/EventHubProducerClient.java +[EventProcessorClient]: ./src/main/java/com/azure/messaging/eventhubs/EventProcessorClient.java +[CreateBatchOptions]: ./src/main/java/com/azure/messaging/eventhubs/models/CreateBatchOptions.java +[InMemoryCheckpointStore]: ./src/samples/java/com/azure/messaging/eventhubs/InMemoryCheckpointStore.java +[LogLevels]: ../../core/azure-core/src/main/java/com/azure/core/util/logging/ClientLogger.java ![Impressions](https://azure-sdk-impressions.azurewebsites.net/api/impressions/azure-sdk-for-java%2Fsdk%2Feventhubs%2Fazure-messaging-eventhubs%2FREADME.png) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClient.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClient.java index 3af7f25ef199..30014b962d0d 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClient.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClient.java @@ -29,8 +29,8 @@ import static com.azure.core.util.FluxUtil.monoError; /** - * A consumer responsible for reading {@link EventData} from a specific Event Hub partition in the context of a specific - * consumer group. + * An asynchronous consumer responsible for reading {@link EventData} from either a specific Event Hub partition + * or all partitions in the context of a specific consumer group. * *

Creating an {@link EventHubConsumerAsyncClient}

*

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

Creating a synchronous consumer

*

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

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

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

* Allowing automatic routing of partitions is recommended when: diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProperties.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProperties.java index 97c64513a32c..15fe0fb2b2e3 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProperties.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProperties.java @@ -41,7 +41,7 @@ public final class EventHubProperties { } /** - * Gets the Event Hub name + * Gets the name of the Event Hub. * * @return Name of the Event Hub. */ @@ -50,7 +50,7 @@ public String getName() { } /** - * Gets the instant, in UTC, at which Event Hub was created at. + * Gets the instant, in UTC, at which Event Hub was created. * * @return The instant, in UTC, at which the Event Hub was created. */ diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEvent.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEvents.java similarity index 99% rename from sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEvent.java rename to sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEvents.java index 1aaea8b63d86..f89dcab19dfb 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEvent.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEvents.java @@ -16,7 +16,7 @@ /** * Sample demonstrates how to receive events from an Azure Event Hub instance. */ -public class ConsumeEvent { +public class ConsumeEvents { private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); private static final int NUMBER_OF_EVENTS = 10; diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEventsFromKnownSequenceNumberPosition.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEventsFromKnownSequenceNumberPosition.java index 733ef6095470..a2559f15e574 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEventsFromKnownSequenceNumberPosition.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEventsFromKnownSequenceNumberPosition.java @@ -4,30 +4,27 @@ import com.azure.messaging.eventhubs.models.EventPosition; import com.azure.messaging.eventhubs.models.SendOptions; -import reactor.core.Disposable; import java.time.Duration; -import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicBoolean; import static java.nio.charset.StandardCharsets.UTF_8; /** - * Sample demonstrates how to receive events starting from the specific sequence number position in an Event Hub instance. + * Sample demonstrates how to receive events starting from the specific sequence number position in an Event Hub + * instance. It also demonstrates how to publish events to a specific partition. */ public class ConsumeEventsFromKnownSequenceNumberPosition { private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); - private static long lastEnqueuedSequenceNumber = -1; - private static String lastEnqueuedSequencePartitionId = null; /** - * Main method to invoke this demo about how to receive event from a known sequence number position in an Azure Event Hub instance. + * Main method to invoke this demo about how to receive event from a known sequence number position in an Azure + * Event Hub instance. * * @param args Unused arguments to the program. - * @throws InterruptedException The countdown latch was interrupted while waiting for this sample to - * complete. */ - public static void main(String[] args) throws InterruptedException { - Semaphore semaphore = new Semaphore(0); + public static void main(String[] args) { + final AtomicBoolean isRunning = new AtomicBoolean(true); // The connection string value can be obtained by: // 1. Going to your Event Hubs namespace in Azure Portal. @@ -36,76 +33,59 @@ public static void main(String[] args) throws InterruptedException { // 4. Copying the connection string from the policy's properties. String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; - EventHubClientBuilder builder = new EventHubClientBuilder() - .connectionString(connectionString) - .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME); - - EventHubConsumerAsyncClient earliestConsumer = builder.buildAsyncConsumerClient(); - - earliestConsumer.getPartitionIds().flatMap(partitionId -> earliestConsumer.getPartitionProperties(partitionId)) - .subscribe( - properties -> { - if (!properties.isEmpty()) { - lastEnqueuedSequenceNumber = properties.getLastEnqueuedSequenceNumber(); - lastEnqueuedSequencePartitionId = properties.getId(); - } - }, - error -> System.err.println("Error occurred while fetching partition properties: " + error.toString()), - () -> { - // Releasing the semaphore now that we've finished querying for partition properties. - semaphore.release(); - }); - - System.out.println("Waiting for partition properties to complete..."); - // Acquiring the semaphore so that this sample does not end before all the partition properties are fetched. - semaphore.acquire(); - System.out.printf("Last enqueued sequence number: %s%n", lastEnqueuedSequenceNumber); + final EventHubClientBuilder builder = new EventHubClientBuilder() + .connectionString(connectionString); + + // The consumer group is required for consuming events. + final EventHubConsumerAsyncClient consumer = builder + .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .buildAsyncConsumerClient(); + + // Find the first non-empty partition we can start consuming from. + // Block on it because we don't know what partition to start reading from, yet. + final PartitionProperties nonEmptyPartition = consumer.getPartitionIds() + .flatMap(partitionId -> consumer.getPartitionProperties(partitionId)) + .filter(properties -> !properties.isEmpty()) + .blockFirst(OPERATION_TIMEOUT); // Make sure to have at least one non-empty event hub in order to continue the sample execution // if you don't have an non-empty event hub, try with another example 'SendEvent' in the same directory. - if (lastEnqueuedSequenceNumber == -1 || lastEnqueuedSequencePartitionId == null) { - System.err.println("All event hubs are empty"); + if (nonEmptyPartition == null) { + System.err.println("All event hub partitions are empty"); System.exit(0); } - // Create a consumer. - // The "$Default" consumer group is created by default. This value can be found by going to the Event Hub - // instance you are connecting to, and selecting the "Consumer groups" page. EventPosition.latest() tells the - // service we only want events that are sent to the partition after we begin listening. - EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() - .connectionString(connectionString) - .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) - .buildAsyncConsumerClient(); + // ex. The last enqueued sequence number is 99. If isInclusive is true, the received event starting from + // the same event with sequence number of '99'. Otherwise, the event with sequence number of '100' will + // be the first event received. + final EventPosition position = EventPosition.fromSequenceNumber( + nonEmptyPartition.getLastEnqueuedSequenceNumber(), true); - // We start receiving any events that come from `firstPartition`, print out the contents, and decrement the - // countDownLatch. - final EventPosition position = EventPosition.fromSequenceNumber(lastEnqueuedSequenceNumber, false); - Disposable subscription = consumer.receiveFromPartition(lastEnqueuedSequencePartitionId, position).subscribe(partitionEvent -> { - EventData event = partitionEvent.getData(); - String contents = new String(event.getBody(), UTF_8); - // ex. The last enqueued sequence number is 99. If isInclusive is true, the received event starting from the same - // event with sequence number of '99'. Otherwise, the event with sequence number of '100' will be the first - // event received. - System.out.println(String.format("Receiving an event starting from the sequence number: %s. Contents: %s", - event.getSequenceNumber(), contents)); + // We start receiving any events that come from that non-empty partition, print out the contents. + // We keep receiving events while `takeWhile` resolves to true, that is, the program is still running. + consumer.receiveFromPartition(nonEmptyPartition.getId(), position) + .takeWhile(ignored -> isRunning.get()) + .subscribe(partitionEvent -> { + EventData event = partitionEvent.getData(); + String contents = new String(event.getBody(), UTF_8); - semaphore.release(); - }); + System.out.println(String.format("Event sequence number number: %s. Contents: %s%n", + event.getSequenceNumber(), contents)); + }); - EventHubProducerAsyncClient producer = builder.buildAsyncProducerClient(); + // Create a producer. + final EventHubProducerClient producer = builder.buildProducerClient(); - // Because the consumer is only listening to new events, we need to send some events to that partition. - // This sends the events to `lastEnqueuedSequencePartitionId`. - SendOptions sendOptions = new SendOptions().setPartitionId(lastEnqueuedSequencePartitionId); + // Because the consumer is only listening to new events after the last enqueued event was received, we need to + // send some events to that partition. + final SendOptions sendOptions = new SendOptions().setPartitionId(nonEmptyPartition.getId()); + producer.send(new EventData("Hello world!" .getBytes(UTF_8)), sendOptions); - producer.send(new EventData("Hello world!".getBytes(UTF_8)), sendOptions).block(OPERATION_TIMEOUT); - // Acquiring the semaphore so that this sample does not end before all events are fetched. - semaphore.acquire(); + // Set isRunning to false so we stop taking events. + isRunning.set(false); // Dispose and close of all the resources we've created. - subscription.dispose(); producer.close(); consumer.close(); - earliestConsumer.close(); } } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClientJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClientJavaDocCodeSamples.java index d6e2ffc1ce59..8485f7088c90 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClientJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClientJavaDocCodeSamples.java @@ -23,7 +23,8 @@ public void initialization() { // BEGIN: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.instantiation // The required parameters are `consumerGroup` and a way to authenticate with Event Hubs using credentials. EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() - .connectionString("event-hub-instance-connection-string") + .connectionString("Endpoint={fully-qualified-namespace};SharedAccessKeyName={policy-name};" + + "SharedAccessKey={key};EntityPath={eh-name}") .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) .buildAsyncConsumerClient(); // END: com.azure.messaging.eventhubs.eventhubconsumerasyncclient.instantiation diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java index ecfb4c6ada5c..109117f647d4 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java @@ -2,21 +2,16 @@ // Licensed under the MIT License. package com.azure.messaging.eventhubs; -import java.util.concurrent.Semaphore; - /** - * Demonstrates how to fetch metadata from an Event Hub's partitions. + * Demonstrates how to fetch metadata from an Event Hub's partitions using synchronous client. */ public class GetEventHubMetadata { /** * Demonstrates how to get metadata from an Event Hub's partitions. * * @param args Unused arguments to the sample. - * @throws InterruptedException if the semaphore could not be acquired. */ - public static void main(String[] args) throws InterruptedException { - Semaphore semaphore = new Semaphore(1); - + public static void main(String[] args) { // The connection string value can be obtained by: // 1. Going to your Event Hubs namespace in Azure Portal. // 2. Creating an Event Hub instance. @@ -24,36 +19,29 @@ public static void main(String[] args) throws InterruptedException { // 4. Copying the connection string from the policy's properties. String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; - // Instantiate a client that will be used to call the service. - EventHubProducerAsyncClient client = new EventHubClientBuilder() + // Instantiate a client that will be used to call the service. Using a try-resource block, so it disposes of + // the client when we are done. + EventHubProducerClient client = new EventHubClientBuilder() .connectionString(connectionString) - .buildAsyncProducerClient(); - - // Acquiring the semaphore so that this sample does not end before all the partition properties are fetched. - semaphore.acquire(); + .buildProducerClient(); // Querying the partition identifiers for the Event Hub. Then calling client.getPartitionProperties with the // identifier to get information about each partition. - client.getPartitionIds().flatMap(partitionId -> client.getPartitionProperties(partitionId)) - .subscribe(properties -> { - System.out.println("The Event Hub has the following properties:"); - System.out.printf( - "Event Hub Name: %s; Partition Id: %s; Is partition empty? %s; First Sequence Number: %s; " - + "Last Enqueued Time: %s; Last Enqueued Sequence Number: %s; Last Enqueued Offset: %s", - properties.getEventHubName(), properties.getId(), properties.isEmpty(), - properties.getBeginningSequenceNumber(), - properties.getLastEnqueuedTime(), - properties.getLastEnqueuedSequenceNumber(), - properties.getLastEnqueuedOffset()); - }, error -> { - System.err.println("Error occurred while fetching partition properties: " + error.toString()); - }, () -> { - // Releasing the semaphore now that we've finished querying for partition properties. - semaphore.release(); - }); + for (String partitionId : client.getPartitionIds()) { + PartitionProperties properties = client.getPartitionProperties(partitionId); + System.out.printf( + "Event Hub Name: %s; Partition Id: %s; Is partition empty? %s; First Sequence Number: %s; " + + "Last Enqueued Time: %s; Last Enqueued Sequence Number: %s; Last Enqueued Offset: %s%n", + properties.getEventHubName(), + properties.getId(), + properties.isEmpty(), + properties.getBeginningSequenceNumber(), + properties.getLastEnqueuedTime(), + properties.getLastEnqueuedSequenceNumber(), + properties.getLastEnqueuedOffset()); + } - System.out.println("Waiting for partition properties to complete..."); - semaphore.acquire(); - System.out.println("Finished."); + // Dispose of the client. + client.close(); } } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventDataBatch.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventDataBatch.java deleted file mode 100644 index 3fb8f293331e..000000000000 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventDataBatch.java +++ /dev/null @@ -1,68 +0,0 @@ -// Copyright (c) Microsoft Corporation. All rights reserved. -// Licensed under the MIT License. -package com.azure.messaging.eventhubs; - -import com.azure.messaging.eventhubs.models.CreateBatchOptions; -import reactor.core.publisher.Flux; - -import java.util.concurrent.atomic.AtomicReference; - -import static java.nio.charset.StandardCharsets.UTF_8; - -/** - * Sample demonstrates how to send an {@link EventDataBatch} to an Azure Event Hub. - */ -public class PublishEventDataBatch { - /** - * Main method to invoke this demo on how to send an {@link EventDataBatch} to an Azure Event Hub. - * - * @param args Unused arguments to the program. - */ - public static void main(String[] args) { - Flux telemetryEvents = Flux.just( - new EventData("Roast beef".getBytes(UTF_8)), - new EventData("Cheese".getBytes(UTF_8)), - new EventData("Tofu".getBytes(UTF_8)), - new EventData("Turkey".getBytes(UTF_8))); - - // The connection string value can be obtained by: - // 1. Going to your Event Hubs namespace in Azure Portal. - // 2. Creating an Event Hub instance. - // 3. Creating a "Shared access policy" for your Event Hub instance. - // 4. Copying the connection string from the policy's properties. - final String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};" - + "SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; - - // Create a producer. - EventHubProducerAsyncClient producer = new EventHubClientBuilder() - .connectionString(connectionString) - .buildAsyncProducerClient(); - - // Creating a batch where we want the events ending up in the same partition by setting the partition key. - final CreateBatchOptions options = new CreateBatchOptions() - .setPartitionKey("sandwiches") - .setMaximumSizeInBytes(256); - final AtomicReference currentBatch = new AtomicReference<>( - producer.createBatch(options).block()); - - // The sample Flux contains three events, but it could be an infinite stream of telemetry events. - telemetryEvents.subscribe(event -> { - final EventDataBatch batch = currentBatch.get(); - if (!batch.tryAdd(event)) { - producer.createBatch(options).map(newBatch -> { - currentBatch.set(newBatch); - return producer.send(batch); - }).block(); - } - }, error -> System.err.println("Error received:" + error), - () -> { - final EventDataBatch batch = currentBatch.getAndSet(null); - if (batch != null) { - producer.send(batch).block(); - } - - // Disposing of our producer. - producer.close(); - }); - } -} diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java index d40335198791..19ef651f105b 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java @@ -3,7 +3,9 @@ package com.azure.messaging.eventhubs; import com.azure.messaging.eventhubs.models.CreateBatchOptions; +import reactor.core.Exceptions; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import java.time.Duration; import java.util.concurrent.atomic.AtomicReference; @@ -11,14 +13,14 @@ import static java.nio.charset.StandardCharsets.UTF_8; /** - * Sample demonstrates how to sent events to specific event hub by define partition ID in producer option only. + * Sample demonstrates how to sent events to specific event hub by defining partition id using + * {@link CreateBatchOptions#setPartitionId(String)}. */ public class PublishEventsToSpecificPartition { private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); /** - * Main method to invoke this demo about how to send a batch of events with partition ID configured in producer - * option to an Azure Event Hub instance. + * Main method to invoke this demo about how to send a batch of events with partition id configured. * * @param args Unused arguments to the program. */ @@ -50,31 +52,51 @@ public static void main(String[] args) { // Create a batch to send the events. final CreateBatchOptions options = new CreateBatchOptions() - .setPartitionId(firstPartition) - .setMaximumSizeInBytes(256); + .setPartitionId(firstPartition); final AtomicReference currentBatch = new AtomicReference<>( producer.createBatch(options).block()); // We try to add as many events as a batch can fit based on the event size and send to Event Hub when // the batch can hold no more events. Create a new batch for next set of events and repeat until all events // are sent. - data.subscribe(event -> { + final Mono sendOperation = data.flatMap(event -> { final EventDataBatch batch = currentBatch.get(); - if (!batch.tryAdd(event)) { + if (batch.tryAdd(event)) { + return Mono.empty(); + } + + // The batch is full, so we create a new batch and send the batch. Mono.when completes when both operations + // have completed. + return Mono.when( + producer.send(batch), producer.createBatch(options).map(newBatch -> { currentBatch.set(newBatch); - return producer.send(batch); - }).block(); - } - }, error -> System.err.println("Error received:" + error), - () -> { + + // Add that event that we couldn't before. + if (!newBatch.tryAdd(event)) { + throw Exceptions.propagate(new IllegalArgumentException(String.format( + "Event is too large for an empty batch. Max size: %s. Event: %s", + newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); + } + + return newBatch; + })); + }).then() + .doFinally(signal -> { final EventDataBatch batch = currentBatch.getAndSet(null); if (batch != null) { - producer.send(batch).block(); + producer.send(batch).block(OPERATION_TIMEOUT); } - - // Disposing of our producer. - producer.close(); }); + + // The sendOperation creation and assignment is not a blocking call. It does not get invoked until there is a + // subscriber to that operation. For the purpose of this example, we block so the program does not end before + // the send operation is complete. Any of the `.subscribe` overloads also work to start the Mono asynchronously. + try { + sendOperation.block(OPERATION_TIMEOUT); + } finally { + // Disposing of our producer. + producer.close(); + } } } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java new file mode 100644 index 000000000000..c89383280c20 --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java @@ -0,0 +1,100 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.azure.messaging.eventhubs; + +import com.azure.core.credential.TokenCredential; +import com.azure.identity.DefaultAzureCredentialBuilder; +import reactor.core.Exceptions; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.time.Duration; +import java.util.concurrent.atomic.AtomicReference; + +import static java.nio.charset.StandardCharsets.UTF_8; + +/** + * Sample demonstrates how to send an {@link EventDataBatch} to an Azure Event Hub using Azure Identity. + */ +public class PublishEventsWithAzureIdentity { + private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); + + /** + * Main method to invoke this demo on how to send an {@link EventDataBatch} to an Azure Event Hub. + * + * @param args Unused arguments to the program. + */ + public static void main(String[] args) { + Flux telemetryEvents = Flux.just( + new EventData("Roast beef".getBytes(UTF_8)), + new EventData("Cheese".getBytes(UTF_8)), + new EventData("Tofu".getBytes(UTF_8)), + new EventData("Turkey".getBytes(UTF_8))); + + // The default azure credential checks multiple locations for credentials and determines the best one to use. + // For the purpose of this sample, create a service principal and set the following environment variables. + // See https://docs.microsoft.com/azure/active-directory/develop/howto-create-service-principal-portal for + // information on how to create a service principal. + System.setProperty("AZURE_CLIENT_ID", "<>"); + System.setProperty("AZURE_CLIENT_ID", "<>"); + System.setProperty("AZURE_TENANT_ID", "<>"); + + // DefaultAzureCredentialBuilder exists inside the azure-identity package. + TokenCredential credential = new DefaultAzureCredentialBuilder() + .build(); + + // Create a producer. + // "<>" will look similar to "{your-namespace}.servicebus.windows.net" + // "<>" will be the name of the Event Hub instance you created inside the Event Hubs namespace. + EventHubProducerAsyncClient producer = new EventHubClientBuilder() + .credential( + "<>", + "<>", + credential) + .buildAsyncProducerClient(); + + final AtomicReference currentBatch = new AtomicReference<>( + producer.createBatch().block()); + + // The sample Flux contains three events, but it could be an infinite stream of telemetry events. + final Mono sendOperation = telemetryEvents.flatMap(event -> { + final EventDataBatch batch = currentBatch.get(); + if (batch.tryAdd(event)) { + return Mono.empty(); + } + + // The batch is full, so we create a new batch and send the batch. Mono.when completes when both operations + // have completed. + return Mono.when( + producer.send(batch), + producer.createBatch().map(newBatch -> { + currentBatch.set(newBatch); + + // Add that event that we couldn't before. + if (!newBatch.tryAdd(event)) { + throw Exceptions.propagate(new IllegalArgumentException(String.format( + "Event is too large for an empty batch. Max size: %s. Event: %s", + newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); + } + + return newBatch; + })); + }).then() + .doFinally(signal -> { + final EventDataBatch batch = currentBatch.getAndSet(null); + if (batch != null) { + producer.send(batch).block(OPERATION_TIMEOUT); + } + }); + + // The sendOperation creation and assignment is not a blocking call. It does not get invoked until there is a + // subscriber to that operation. For the purpose of this example, we block so the program does not end before + // the send operation is complete. Any of the `.subscribe` overloads also work to start the Mono asynchronously. + try { + sendOperation.block(OPERATION_TIMEOUT); + } finally { + // Disposing of our producer. + producer.close(); + } + } +} diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java index fb4e1560b295..3da7d8128637 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java @@ -2,8 +2,9 @@ // Licensed under the MIT License. package com.azure.messaging.eventhubs; -import com.azure.messaging.eventhubs.models.CreateBatchOptions; +import reactor.core.Exceptions; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import java.time.Duration; import java.util.concurrent.atomic.AtomicReference; @@ -11,7 +12,9 @@ import static java.nio.charset.StandardCharsets.UTF_8; /** - * Sample demonstrates how to sent events to a specific event hub by defining partition ID in producer option only. + * Sample demonstrates how to sent events with custom metadata to Event Hubs using {@link EventData#getProperties()}. + * Allows the service to load-balance the events between all partitions by using + * {@link EventHubProducerAsyncClient#createBatch()} which uses the default set of create batch options. */ public class PublishEventsWithCustomMetadata { private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); @@ -37,62 +40,68 @@ public static void main(String[] args) { // Because an event consists mainly of an opaque set of bytes, it may be difficult for consumers of those events // to make informed decisions about how to process them. // - // In order to allow event publishers to offer better context for consumers, event data may also contain custom metadata, - // in the form of a set of key/value pairs. This metadata is not used by, or in any way meaningful to, the Event Hubs - // service; it exists only for coordination between event publishers and consumers. + // In order to allow event publishers to offer better context for consumers, event data may also contain custom + // metadata, in the form of a set of key/value pairs. This metadata is not used by, or in any way meaningful to, + // the Event Hubs service; it exists only for coordination between event publishers and consumers. // - // One common scenario for the inclusion of metadata is to provide a hint about the type of data contained by an event, - // so that consumers understand its format and can deserialize it appropriately. - // - // We will publish two events based on simple sentences, but will attach some custom metadata with - // pretend type names and other hints. Note that the set of metadata is unique to an event; there is no need for every - // event in a batch to have the same metadata properties available nor the same data type for those properties. + // One common scenario for the inclusion of metadata is to provide a hint about the type of data contained by an + // event, so that consumers understand its format and can deserialize it appropriately. EventData firstEvent = new EventData("EventData Sample 1".getBytes(UTF_8)); firstEvent.getProperties().put("EventType", "com.microsoft.samples.hello-event"); firstEvent.getProperties().put("priority", 1); firstEvent.getProperties().put("score", 9.0); - EventData secEvent = new EventData("EventData Sample 2".getBytes(UTF_8)); - secEvent.getProperties().put("EventType", "com.microsoft.samples.goodbye-event"); - secEvent.getProperties().put("priority", "17"); - secEvent.getProperties().put("blob", 10); - - final Flux data = Flux.just(firstEvent, secEvent); + EventData secondEvent = new EventData("EventData Sample 2".getBytes(UTF_8)); + secondEvent.getProperties().put("EventType", "com.microsoft.samples.goodbye-event"); + secondEvent.getProperties().put("priority", "17"); + secondEvent.getProperties().put("blob", 10); - // We want to send events to the a specific partition. For the sake of this sample, we take the first partition - // identifier. - // .blockFirst() here is used to synchronously block until the first partition id is emitted. The maximum wait - // time is set by passing in the OPERATION_TIMEOUT value. If no item is emitted before the timeout elapses, a - // TimeoutException is thrown. - String firstPartition = producer.getPartitionIds().blockFirst(OPERATION_TIMEOUT); + final Flux data = Flux.just(firstEvent, secondEvent); - // Create a batch to send the events. - final CreateBatchOptions options = new CreateBatchOptions() - .setPartitionId(firstPartition) - .setMaximumSizeInBytes(256); final AtomicReference currentBatch = new AtomicReference<>( - producer.createBatch(options).block()); + producer.createBatch().block()); // We try to add as many events as a batch can fit based on the event size and send to Event Hub when // the batch can hold no more events. Create a new batch for next set of events and repeat until all events // are sent. - data.subscribe(event -> { + final Mono sendOperation = data.flatMap(event -> { final EventDataBatch batch = currentBatch.get(); - if (!batch.tryAdd(event)) { - producer.createBatch(options).map(newBatch -> { - currentBatch.set(newBatch); - return producer.send(batch); - }).block(); + if (batch.tryAdd(event)) { + return Mono.empty(); } - }, error -> System.err.println("Error received:" + error), - () -> { + + // The batch is full, so we create a new batch and send the batch. Mono.when completes when both operations + // have completed. + return Mono.when( + producer.send(batch), + producer.createBatch().map(newBatch -> { + currentBatch.set(newBatch); + + // Add that event that we couldn't before. + if (!newBatch.tryAdd(event)) { + throw Exceptions.propagate(new IllegalArgumentException(String.format( + "Event is too large for an empty batch. Max size: %s. Event: %s", + newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); + } + + return newBatch; + })); + }).then() + .doFinally(signal -> { final EventDataBatch batch = currentBatch.getAndSet(null); if (batch != null) { - producer.send(batch).block(); + producer.send(batch).block(OPERATION_TIMEOUT); } - - // Disposing of our producer. - producer.close(); }); + + // The sendOperation creation and assignment is not a blocking call. It does not get invoked until there is a + // subscriber to that operation. For the purpose of this example, we block so the program does not end before + // the send operation is complete. Any of the `.subscribe` overloads also work to start the Mono asynchronously. + try { + sendOperation.block(OPERATION_TIMEOUT); + } finally { + // Disposing of our producer. + producer.close(); + } } } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java index 23918d80a7b7..2496e26e06f3 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java @@ -3,20 +3,24 @@ package com.azure.messaging.eventhubs; import com.azure.messaging.eventhubs.models.CreateBatchOptions; +import reactor.core.Exceptions; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import java.time.Duration; import java.util.concurrent.atomic.AtomicReference; import static java.nio.charset.StandardCharsets.UTF_8; /** - * Send a list of events with send option configured + * Send a Flux of events using a partition key. */ public class PublishEventsWithPartitionKey { + private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); /** - * Main method to invoke this demo about how to send a list of events with partition ID configured in send option - * to an Azure Event Hub instance. + * Main method to invoke this demo about how to send a list of events with partition key configured in + * CreateBatchOptions to an Azure Event Hub instance. * * @param args Unused arguments to the program. */ @@ -43,41 +47,63 @@ public static void main(String[] args) { // the Event Hubs service keep different batches of events together on the same partition. This can be // accomplished by setting a partition key when publishing the events. // - // The partition key is NOT the identifier of a specific partition. Rather, it is an arbitrary piece of string data - // that Event Hubs uses as the basis to compute a hash value. Event Hubs will associate the hash value with a specific - // partition, ensuring that any events published with the same partition key are rerouted to the same partition. + // The partition key is NOT the identifier of a specific partition. Rather, it is an arbitrary piece of string + // data that Event Hubs uses as the basis to compute a hash value. Event Hubs will associate the hash value with + // a specific partition, ensuring that any events published with the same partition key are rerouted to the same + // partition. // - // All of event data send to the same partition of the partition key 'basketball' associate with. + // All the event data with partition key 'basketball' end up in the same partition. // - // Note that there is no means of accurately predicting which partition will be associated with a given partition key; - // we can only be assured that it will be a consistent choice of partition. If you have a need to understand which - // exact partition an event is published to, you will need to use an Event Hub producer associated with that partition. + // Note that there is no means of accurately predicting which partition will be associated with a given + // partition key; we can only be assured that it will be a consistent choice of partition. If you have a need to + // understand which exact partition an event is published to, you will need to use + // CreateBatchOptions.setPartitionId(String) when creating the EventDataBatch. final CreateBatchOptions options = new CreateBatchOptions() - .setPartitionKey("basketball") - .setMaximumSizeInBytes(256); + .setPartitionKey("basketball"); final AtomicReference currentBatch = new AtomicReference<>( producer.createBatch(options).block()); // We try to add as many events as a batch can fit based on the event size and send to Event Hub when // the batch can hold no more events. Create a new batch for next set of events and repeat until all events // are sent. - data.subscribe(event -> { + final Mono sendOperation = data.flatMap(event -> { final EventDataBatch batch = currentBatch.get(); - if (!batch.tryAdd(event)) { - producer.createBatch(options).map(newBatch -> { - currentBatch.set(newBatch); - return producer.send(batch); - }).block(); + if (batch.tryAdd(event)) { + return Mono.empty(); } - }, error -> System.err.println("Error received:" + error), - () -> { + + // The batch is full, so we create a new batch and send the batch. Mono.when completes when both operations + // have completed. + return Mono.when( + producer.send(batch), + producer.createBatch().map(newBatch -> { + currentBatch.set(newBatch); + + // Add that event that we couldn't before. + if (!newBatch.tryAdd(event)) { + throw Exceptions.propagate(new IllegalArgumentException(String.format( + "Event is too large for an empty batch. Max size: %s. Event: %s", + newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); + } + + return newBatch; + })); + }).then() + .doFinally(signal -> { final EventDataBatch batch = currentBatch.getAndSet(null); if (batch != null) { - producer.send(batch).block(); + producer.send(batch).block(OPERATION_TIMEOUT); } - - // Disposing of our producer. - producer.close(); }); + + // The sendOperation creation and assignment is not a blocking call. It does not get invoked until there is a + // subscriber to that operation. For the purpose of this example, we block so the program does not end before + // the send operation is complete. Any of the `.subscribe` overloads also work to start the Mono asynchronously. + try { + sendOperation.block(OPERATION_TIMEOUT); + } finally { + // Disposing of our producer. + producer.close(); + } } } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java new file mode 100644 index 000000000000..21544154c737 --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java @@ -0,0 +1,102 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.messaging.eventhubs; + +import com.azure.messaging.eventhubs.models.CreateBatchOptions; +import reactor.core.Exceptions; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.time.Duration; +import java.util.concurrent.atomic.AtomicReference; + +import static java.nio.charset.StandardCharsets.UTF_8; + +/** + * Demonstrates how to publish events when there is a size constraint on batch size using + * {@link CreateBatchOptions#setMaximumSizeInBytes(int)}. + */ +public class PublishEventsWithSizeLimitedBatches { + private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); + + /** + * Main method to invoke this demo on how to send an {@link EventDataBatch} to an Azure Event Hub. + * + * @param args Unused arguments to the program. + */ + public static void main(String[] args) { + Flux telemetryEvents = Flux.just( + new EventData("Roast beef".getBytes(UTF_8)), + new EventData("Cheese".getBytes(UTF_8)), + new EventData("Tofu".getBytes(UTF_8)), + new EventData("Turkey".getBytes(UTF_8))); + + // The connection string value can be obtained by: + // 1. Going to your Event Hubs namespace in Azure Portal. + // 2. Creating an Event Hub instance. + // 3. Creating a "Shared access policy" for your Event Hub instance. + // 4. Copying the connection string from the policy's properties. + String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; + + // Instantiate a client that will be used to call the service. + EventHubProducerAsyncClient producer = new EventHubClientBuilder() + .connectionString(connectionString) + .buildAsyncProducerClient(); + + // In cases where developers need to size limit their batch size, they can use `setMaximumSizeInBytes` to limit + // the size of their EventDataBatch. By default, it will be the max size allowed by the underlying link. + // Since there is no partition id or partition key set, the Event Hubs service will automatically load balance + // the events between all available partitions. + final CreateBatchOptions options = new CreateBatchOptions() + .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. + // We try to add as many events as a batch can fit based on the event size and send to Event Hub when + // the batch can hold no more events. Create a new batch for next set of events and repeat until all events + // are sent. + final Mono sendOperation = telemetryEvents.flatMap(event -> { + final EventDataBatch batch = currentBatch.get(); + if (batch.tryAdd(event)) { + return Mono.empty(); + } + + // The batch is full, so we create a new batch and send the batch. Mono.when completes when both operations + // have completed. + return Mono.when( + producer.send(batch), + producer.createBatch().map(newBatch -> { + currentBatch.set(newBatch); + + // Add that event that we couldn't before. + if (!newBatch.tryAdd(event)) { + throw Exceptions.propagate(new IllegalArgumentException(String.format( + "Event is too large for an empty batch. Max size: %s. Event: %s", + newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); + } + + return newBatch; + })); + }).then() + .doFinally(signal -> { + final EventDataBatch batch = currentBatch.getAndSet(null); + if (batch != null) { + producer.send(batch).block(OPERATION_TIMEOUT); + } + }); + + // The sendOperation creation and assignment is not a blocking call. It does not get invoked until there is a + // subscriber to that operation. For the purpose of this example, we block so the program does not end before + // the send operation is complete. Any of the `.subscribe` overloads also work to start the Mono asynchronously. + try { + sendOperation.block(OPERATION_TIMEOUT); + } finally { + // Disposing of our producer. + producer.close(); + } + } +} + +