feat(eventhubs): add single-event send to EventHubProducerClient and EventHubProducerAsyncClient

eng/lintingconfigs/revapi/track2/revapi.json

@@ -328,6 +328,14 @@
           "annotationType": "com\\.azure\\.cosmos\\.util\\.Beta",
           "justification": "Cosmos SDK removes the Beta annotation to GA its APIs. Only Beta annotation removals are permitted; removal of any other annotation (Deprecated, Jackson, etc.) will still fail the build."
         },
+        {
+          "code": "java.method.visibilityIncreased",
+          "old": {
+            "matcher": "regex",
+            "match": "method (void|reactor\\.core\\.publisher\\.Mono<java\\.lang\\.Void>) com\\.azure\\.messaging\\.eventhubs\\.EventHubProducer(Async)?Client::send\\(com\\.azure\\.messaging\\.eventhubs\\.EventData(, com\\.azure\\.messaging\\.eventhubs\\.models\\.SendOptions)?\\)"
+          },
+          "justification": "The single-event send methods on the Event Hubs producer clients change from package-private to public. This adds API surface and removes none, so callers stay source and binary compatible. See https://github.com/Azure/azure-sdk-for-java/issues/41395"
+        },
         {
           "code": "java.method.removed",
           "old": {

sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md

@@ -4,6 +4,12 @@
 
 ### Features Added
 
+- Added public single-event send methods to the producer clients: `EventHubProducerClient.send(EventData)`,
+  `EventHubProducerClient.send(EventData, SendOptions)`, `EventHubProducerAsyncClient.send(EventData)`, and
+  `EventHubProducerAsyncClient.send(EventData, SendOptions)`. These methods were present but package-private. You no
+  longer need to wrap a single event in an `EventDataBatch` or a list. For high throughput, continue to use
+  `EventDataBatch`. ([#41395](https://github.com/Azure/azure-sdk-for-java/issues/41395))
+
 ### Breaking Changes
 
 ### Bugs Fixed

sdk/eventhubs/azure-messaging-eventhubs/README.md

@@ -34,6 +34,7 @@ documentation][event_hubs_product_docs] | [Samples][sample_examples] | [Troubles
 - [Examples](#examples)
   - [Publish events to an Event Hub](#publish-events-to-an-event-hub)
     - [Create an Event Hub producer and publish events](#create-an-event-hub-producer-and-publish-events)
+    - [Publish a single event](#publish-a-single-event)
     - [Publish events using partition identifier](#publish-events-using-partition-identifier)
     - [Publish events using partition key](#publish-events-using-partition-key)
   - [Consume events from an Event Hub partition](#consume-events-from-an-event-hub-partition)
@@ -244,6 +245,21 @@ producer.close();
 Note that `EventDataBatch.tryAdd(EventData)` is not thread-safe. Please make sure to synchronize the method access
 when using multiple threads to add events.
 
+#### Publish a single event
+
+A batch is not necessary to publish one event. The producer clients have a `send(EventData)` overload that publishes a
+single event. There is also a `send(EventData, SendOptions)` overload that accepts a partition id or a partition key.
+The synchronous overload is shown below. `EventHubProducerAsyncClient` has the same overloads and returns a
+`Mono<Void>`.
+
+```java com.azure.messaging.eventhubs.eventhubproducerclient.send#EventData
+EventData event = new EventData("maple");
+producer.send(event);
+```
+
+Use these overloads for low volume publishing. For high throughput, use `EventDataBatch` or `send(Iterable)`, because
+each `send(EventData)` call makes its own network round trip.
+
 #### Publish events using partition identifier
 
 Many Event Hub operations take place within the scope of a specific partition. Any client can call

sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClient.java

@@ -398,10 +398,25 @@ public Mono<EventDataBatch> createBatch(CreateBatchOptions options) {
      * <a href="https://docs.microsoft.com/azure/event-hubs/event-hubs-quotas">Azure Event Hubs Quotas and Limits</a>.
      * </p>
      *
+     * <!-- src_embed com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#EventData -->
+     * <pre>
+     * EventData event = new EventData&#40;&quot;maple&quot;&#41;;
+     *
+     * producer.send&#40;event&#41;
+     *     .subscribe&#40;unused -&gt; &#123;
+     *     &#125;,
+     *         error -&gt; System.err.println&#40;&quot;Error occurred while sending event:&quot; + error&#41;,
+     *         &#40;&#41; -&gt; System.out.println&#40;&quot;Send complete.&quot;&#41;&#41;;
+     * </pre>
+     * <!-- end com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#EventData -->
+     *
      * @param event Event to send to the service.
      * @return A {@link Mono} that completes when the event is pushed to the service.
+     * @throws NullPointerException if {@code event} is {@code null}.
+     * @throws AmqpException if the size of {@code event} exceeds the maximum size of a single batch.
      */
-    Mono<Void> send(EventData event) {
+    @ServiceMethod(returns = ReturnType.SINGLE)
+    public Mono<Void> send(EventData event) {
         if (event == null) {
             return monoError(LOGGER, new NullPointerException("'event' cannot be null."));
         }
@@ -410,8 +425,23 @@ Mono<Void> send(EventData event) {
     }
 
     /**
-     * Sends a single event to the associated Event Hub with the send options. If the size of the single event exceeds
-     * the maximum size allowed, an exception will be triggered and the send will fail.
+     * <p>Sends a single event to the associated Event Hub with the send options. If the size of the single event
+     * exceeds the maximum size allowed, an exception will be triggered and the send will fail. For high throughput
+     * publishing scenarios, using {@link EventDataBatch} to publish events is recommended. Batches are created using
+     * {@link #createBatch()} and {@link #createBatch(CreateBatchOptions)}.</p>
+     *
+     * <!-- src_embed com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#EventData-SendOptions -->
+     * <pre>
+     * EventData event = new EventData&#40;&quot;Melbourne&quot;&#41;;
+     *
+     * SendOptions sendOptions = new SendOptions&#40;&#41;.setPartitionKey&#40;&quot;cities&quot;&#41;;
+     * producer.send&#40;event, sendOptions&#41;
+     *     .subscribe&#40;unused -&gt; &#123;
+     *     &#125;,
+     *         error -&gt; System.err.println&#40;&quot;Error occurred while sending event:&quot; + error&#41;,
+     *         &#40;&#41; -&gt; System.out.println&#40;&quot;Send complete.&quot;&#41;&#41;;
+     * </pre>
+     * <!-- end com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#EventData-SendOptions -->
      *
      * <p>
      * For more information regarding the maximum event size allowed, see
@@ -422,8 +452,11 @@ Mono<Void> send(EventData event) {
      * @param event Event to send to the service.
      * @param options The set of options to consider when sending this event.
      * @return A {@link Mono} that completes when the event is pushed to the service.
+     * @throws NullPointerException if {@code event} or {@code options} is {@code null}.
+     * @throws AmqpException if the size of {@code event} exceeds the maximum size of a single batch.
      */
-    Mono<Void> send(EventData event, SendOptions options) {
+    @ServiceMethod(returns = ReturnType.SINGLE)
+    public Mono<Void> send(EventData event, SendOptions options) {
         if (event == null) {
             return monoError(LOGGER, new NullPointerException("'event' cannot be null."));
         } else if (options == null) {

sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubProducerClient.java

@@ -274,8 +274,16 @@ public EventDataBatch createBatch(CreateBatchOptions options) {
     }
 
     /**
-     * Sends a single event to the associated Event Hub. If the size of the single event exceeds the maximum size
-     * allowed, an exception will be triggered and the send will fail.
+     * <p>Sends a single event to the associated Event Hub. If the size of the single event exceeds the maximum size
+     * allowed, an exception will be triggered and the send will fail. For high throughput publishing, use
+     * {@link EventDataBatch} or the {@link #send(Iterable)} overload to publish more than one event in one call.</p>
+     *
+     * <!-- src_embed com.azure.messaging.eventhubs.eventhubproducerclient.send#EventData -->
+     * <pre>
+     * EventData event = new EventData&#40;&quot;maple&quot;&#41;;
+     * producer.send&#40;event&#41;;
+     * </pre>
+     * <!-- end com.azure.messaging.eventhubs.eventhubproducerclient.send#EventData -->
      *
      * <p>
      * For more information regarding the maximum event size allowed, see
@@ -284,15 +292,28 @@ public EventDataBatch createBatch(CreateBatchOptions options) {
      * </p>
      *
      * @param event Event to send to the service.
+     * @throws NullPointerException if {@code event} is {@code null}.
+     * @throws AmqpException if the size of {@code event} exceeds the maximum size of a single batch.
      */
     @ServiceMethod(returns = ReturnType.SINGLE)
-    void send(EventData event) {
+    public void send(EventData event) {
         producer.send(event).block();
     }
 
     /**
-     * Sends a single event to the associated Event Hub with the send options. If the size of the single event exceeds
-     * the maximum size allowed, an exception will be triggered and the send will fail.
+     * <p>Sends a single event to the associated Event Hub with the send options. If the size of the single event
+     * exceeds the maximum size allowed, an exception will be triggered and the send will fail. For high throughput
+     * publishing, use {@link EventDataBatch} or the {@link #send(Iterable, SendOptions)} overload to publish more
+     * than one event in one call.</p>
+     *
+     * <!-- src_embed com.azure.messaging.eventhubs.eventhubproducerclient.send#EventData-SendOptions -->
+     * <pre>
+     * EventData event = new EventData&#40;&quot;Melbourne&quot;&#41;;
+     *
+     * SendOptions sendOptions = new SendOptions&#40;&#41;.setPartitionKey&#40;&quot;cities&quot;&#41;;
+     * producer.send&#40;event, sendOptions&#41;;
+     * </pre>
+     * <!-- end com.azure.messaging.eventhubs.eventhubproducerclient.send#EventData-SendOptions -->
      *
      * <p>
      * For more information regarding the maximum event size allowed, see
@@ -302,9 +323,11 @@ void send(EventData event) {
      *
      * @param event Event to send to the service.
      * @param options The set of options to consider when sending this event.
+     * @throws NullPointerException if {@code event} or {@code options} is {@code null}.
+     * @throws AmqpException if the size of {@code event} exceeds the maximum size of a single batch.
      */
     @ServiceMethod(returns = ReturnType.SINGLE)
-    void send(EventData event, SendOptions options) {
+    public void send(EventData event, SendOptions options) {
         producer.send(event, options).block();
     }
 

sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventHubsJavaDocCodeSamples.java

@@ -487,6 +487,42 @@ public void batchSizeLimitedAsync() {
         // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.createBatch#CreateBatchOptions-int
     }
 
+    /**
+     * Code snippet to demonstrate how to send a single event using
+     * {@link EventHubProducerAsyncClient#send(EventData)}.
+     */
+    public void sendSingleEventSampleAsync() {
+        final EventHubProducerAsyncClient producer = builder.buildAsyncProducerClient();
+        // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#EventData
+        EventData event = new EventData("maple");
+
+        producer.send(event)
+            .subscribe(unused -> {
+            },
+                error -> System.err.println("Error occurred while sending event:" + error),
+                () -> System.out.println("Send complete."));
+        // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#EventData
+    }
+
+    /**
+     * Code snippet to demonstrate how to send a single event using
+     * {@link EventHubProducerAsyncClient#send(EventData, SendOptions)}.
+     */
+    public void sendSingleEventWithPartitionKeySampleAsync() {
+        final EventHubProducerAsyncClient producer = builder.buildAsyncProducerClient();
+
+        // BEGIN: com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#EventData-SendOptions
+        EventData event = new EventData("Melbourne");
+
+        SendOptions sendOptions = new SendOptions().setPartitionKey("cities");
+        producer.send(event, sendOptions)
+            .subscribe(unused -> {
+            },
+                error -> System.err.println("Error occurred while sending event:" + error),
+                () -> System.out.println("Send complete."));
+        // END: com.azure.messaging.eventhubs.eventhubasyncproducerclient.send#EventData-SendOptions
+    }
+
     /**
      * Code snippet to demonstrate how to send a list of events using
      * {@link EventHubProducerAsyncClient#send(Iterable)}.
@@ -650,6 +686,31 @@ public void batchSizeLimited() {
         // END: com.azure.messaging.eventhubs.eventhubproducerclient.createBatch#CreateBatchOptions-int
     }
 
+    /**
+     * Code snippet to demonstrate how to send a single event using {@link EventHubProducerClient#send(EventData)}.
+     */
+    public void sendSingleEventSample() {
+        final EventHubProducerClient producer = builder.buildProducerClient();
+        // BEGIN: com.azure.messaging.eventhubs.eventhubproducerclient.send#EventData
+        EventData event = new EventData("maple");
+        producer.send(event);
+        // END: com.azure.messaging.eventhubs.eventhubproducerclient.send#EventData
+    }
+
+    /**
+     * Code snippet to demonstrate how to send a single event using
+     * {@link EventHubProducerClient#send(EventData, SendOptions)}.
+     */
+    public void sendSingleEventWithPartitionKeySample() {
+        final EventHubProducerClient producer = builder.buildProducerClient();
+        // BEGIN: com.azure.messaging.eventhubs.eventhubproducerclient.send#EventData-SendOptions
+        EventData event = new EventData("Melbourne");
+
+        SendOptions sendOptions = new SendOptions().setPartitionKey("cities");
+        producer.send(event, sendOptions);
+        // END: com.azure.messaging.eventhubs.eventhubproducerclient.send#EventData-SendOptions
+    }
+
     /**
      * Code snippet to demonstrate how to send a list of events using {@link EventHubProducerClient#send(Iterable)}.
      */

sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerAsyncClientTest.java

@@ -203,6 +203,28 @@ void teardown(TestInfo testInfo) {
         messagesCaptor = null;
     }
 
+    /**
+     * Verifies that a null event or null options results in a NullPointerException.
+     */
+    @Test
+    void sendSingleMessageNullArguments() {
+        // Arrange
+        final EventData testData = new EventData(TEST_CONTENTS.getBytes(UTF_8));
+
+        // Act & Assert
+        StepVerifier.create(producer.send((EventData) null))
+            .expectError(NullPointerException.class)
+            .verify(DEFAULT_TIMEOUT);
+
+        StepVerifier.create(producer.send((EventData) null, new SendOptions()))
+            .expectError(NullPointerException.class)
+            .verify(DEFAULT_TIMEOUT);
+
+        StepVerifier.create(producer.send(testData, null))
+            .expectError(NullPointerException.class)
+            .verify(DEFAULT_TIMEOUT);
+    }
+
     /**
      * Verifies that sending multiple events will result in calling producer.send(List<Message>).
      */

sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubProducerClientTest.java

@@ -172,6 +172,26 @@ public void sendSingleMessage() {
         Assertions.assertEquals(Section.SectionType.Data, message.getBody().getType());
     }
 
+    /**
+     * Verifies that a null event or null options throws a NullPointerException.
+     */
+    @Test
+    public void sendSingleMessageNullArguments() {
+        // Arrange
+        final EventHubProducerClient producer = new EventHubProducerClient(asyncProducer);
+        final EventData eventData = new EventData("hello-world".getBytes(UTF_8));
+
+        // Act & Assert
+        try {
+            Assertions.assertThrows(NullPointerException.class, () -> producer.send((EventData) null));
+            Assertions.assertThrows(NullPointerException.class,
+                () -> producer.send((EventData) null, new SendOptions()));
+            Assertions.assertThrows(NullPointerException.class, () -> producer.send(eventData, null));
+        } finally {
+            producer.close();
+        }
+    }
+
     /**
      *Verifies start and end span invoked when sending a single message.
      */