Package com.azure.messaging.eventhubs

Class EventHubBufferedProducerAsyncClient

java.lang.Object
com.azure.messaging.eventhubs.EventHubBufferedProducerAsyncClient
All Implemented Interfaces:
Closeable, AutoCloseable
public final class EventHubBufferedProducerAsyncClient extends Object implements Closeable
A client responsible for publishing instances of EventData to a specific Event Hub. Depending on the options specified when events are enqueued, they may be automatically assigned to a partition, grouped according to the specified partition key, or assigned a specifically requested partition.

The EventHubBufferedProducerAsyncClient does not publish immediately, instead using a deferred model where events are collected into a buffer so that they may be efficiently batched and published when the batch is full or the maxWaitTime has elapsed with no new events enqueued.

This model is intended to shift the burden of batch management from callers, at the cost of non-deterministic timing, for when events will be published. There are additional trade-offs to consider, as well:

  • If the application crashes, events in the buffer will not have been published. To prevent data loss, callers are encouraged to track publishing progress using onSendBatchFailed and onSendBatchSucceeded.
  • Events specifying a partition key may be assigned a different partition than those using the same key with other producers.
  • In the unlikely event that a partition becomes temporarily unavailable, the EventHubBufferedProducerAsyncClient may take longer to recover than other producers.

In scenarios where it is important to have events published immediately with a deterministic outcome, ensure that partition keys are assigned to a partition consistent with other publishers, or where maximizing availability is a requirement, using EventHubProducerAsyncClient or EventHubProducerClient is recommended.

Sample: Creating an EventHubBufferedProducerAsyncClient

The following code sample demonstrates the creation of the asynchronous client EventHubBufferedProducerAsyncClient. The fullyQualifiedNamespace is the Event Hubs Namespace's host name. It is listed under the "Essentials" panel after navigating to the Event Hubs Namespace via Azure Portal. The producer is set to publish events every 60 seconds with a buffer size of 1500 events for each partition. The examples shown in this document use a credential object named DefaultAzureCredential for authentication, which is appropriate for most scenarios, including local development and production environments. Additionally, we recommend using managed identity for authentication in production environments. You can find more information on different ways of authenticating and their corresponding credential types in the Azure Identity documentation.

 TokenCredential credential = new DefaultAzureCredentialBuilder().build();

 // "<<fully-qualified-namespace>>" will look similar to "{your-namespace}.servicebus.windows.net"
 // "<<event-hub-name>>" will be the name of the Event Hub instance you created inside the Event Hubs namespace.
 EventHubBufferedProducerAsyncClient client = new EventHubBufferedProducerClientBuilder()
     .credential("fully-qualified-namespace", "event-hub-name", credential)
     .onSendBatchSucceeded(succeededContext -> {
         System.out.println("Successfully published events to: " + succeededContext.getPartitionId());
     })
     .onSendBatchFailed(failedContext -> {
         System.out.printf("Failed to published events to %s. Error: %s%n",
             failedContext.getPartitionId(), failedContext.getThrowable());
     })
     .maxWaitTime(Duration.ofSeconds(60))
     .maxEventBufferLengthPerPartition(1500)
     .buildAsyncClient();

Sample: Enqueuing and publishing events

The following code sample demonstrates enqueuing a set of events in the buffered producer. The producer stores these events in an internal queue and publishes them when EventHubBufferedProducerClientBuilder.maxWaitTime(Duration) has elapsed, the buffer is full, or no more events can fit into a batch.

NOTE that Mono<Integer> returned must be subscribed to, or eventually subscribed to if chained to reactive operators in order to start the operation.

 TokenCredential credential = new DefaultAzureCredentialBuilder().build();

 // "<<fully-qualified-namespace>>" will look similar to "{your-namespace}.servicebus.windows.net"
 // "<<event-hub-name>>" will be the name of the Event Hub instance you created inside the Event Hubs namespace.
 EventHubBufferedProducerAsyncClient client = new EventHubBufferedProducerClientBuilder()
     .credential("fully-qualified-namespace", "event-hub-name", credential)
     .onSendBatchSucceeded(succeededContext -> {
         System.out.println("Successfully published events to: " + succeededContext.getPartitionId());
     })
     .onSendBatchFailed(failedContext -> {
         System.out.printf("Failed to published events to %s. Error: %s%n",
             failedContext.getPartitionId(), failedContext.getThrowable());
     })
     .buildAsyncClient();

 List<EventData> events = Arrays.asList(new EventData("maple"), new EventData("aspen"),
     new EventData("oak"));

 // Enqueues the events to be published.
 client.enqueueEvents(events).subscribe(numberOfEvents -> {
     System.out.printf("There are currently: %d events in buffer.%n", numberOfEvents);
 }, error -> {
         System.err.println("Error occurred enqueueing events: " + error);
     },
     () -> {
         System.out.println("Events successfully enqueued.");
     });

 // Seconds later, enqueue another event.
 client.enqueueEvent(new EventData("bonsai")).subscribe(numberOfEvents -> {
     System.out.printf("There are %d events in the buffer.%n", numberOfEvents);
 }, error -> {
         System.err.println("Error occurred enqueueing events: " + error);
     },
     () -> {
         System.out.println("Event successfully enqueued.");
     });

 // Causes any buffered events to be flushed before closing underlying connection.
 client.close();
See Also:

  • Method Details

    • getFullyQualifiedNamespace

      public String getFullyQualifiedNamespace()
      Gets the fully qualified Event Hubs namespace that the connection is associated with. This is likely similar to {yournamespace}.servicebus.windows.net.
      Returns:
      The fully qualified Event Hubs namespace that the connection is associated with

    • getEventHubName

      public String getEventHubName()
      Gets the Event Hub name this client interacts with.
      Returns:
      The Event Hub name this client interacts with.

    • getEventHubProperties

      public Mono<EventHubProperties> getEventHubProperties()
      Retrieves information about an Event Hub, including the number of partitions present and their identifiers.
      Returns:
      The set of information for the Event Hub that this client is associated with.

    • getPartitionIds

      public Flux<String> getPartitionIds()
      Retrieves the identifiers for the partitions of an Event Hub.
      Returns:
      A Flux of identifiers for the partitions of an Event Hub.

    • getPartitionProperties

      public Mono<PartitionProperties> getPartitionProperties(String partitionId)
      Retrieves information about a specific partition for an Event Hub, including elements that describe the available events in the partition event stream.
      Parameters:
      partitionId - The unique identifier of a partition associated with the Event Hub.
      Returns:
      The set of information for the requested partition under the Event Hub this client is associated with.
      Throws:
      NullPointerException - if partitionId is null.
      IllegalArgumentException - if partitionId is empty.

    • getBufferedEventCount

      public int getBufferedEventCount()
      Gets the total number of events that are currently buffered and waiting to be published, across all partitions.
      Returns:
      The total number of events that are currently buffered and waiting to be published, across all partitions.

    • getBufferedEventCount

      public int getBufferedEventCount(String partitionId)
      Gets the number of events that are buffered and waiting to be published for a given partition.
      Parameters:
      partitionId - The partition identifier.
      Returns:
      The number of events that are buffered and waiting to be published for a given partition.
      Throws:
      NullPointerException - if partitionId is null.
      IllegalArgumentException - if partitionId is empty.

    • enqueueEvent

      public Mono<Integer> enqueueEvent(EventData eventData)
      Enqueues an EventData into the buffer to be published to the Event Hub. If there is no capacity in the buffer when this method is invoked, it will wait for space to become available and ensure that the eventData has been enqueued. When this call returns, the eventData has been accepted into the buffer, but it may not have been published yet. Publishing will take place at a nondeterministic point in the future as the buffer is processed.
      Parameters:
      eventData - The event to be enqueued into the buffer and, later, published.
      Returns:
      The total number of events that are currently buffered and waiting to be published, across all partitions.
      Throws:
      NullPointerException - if eventData is null.
      IllegalStateException - if the producer was closed while queueing an event.

    • enqueueEvent

      public Mono<Integer> enqueueEvent(EventData eventData, SendOptions options)
      Enqueues an EventData into the buffer to be published to the Event Hub. If there is no capacity in the buffer when this method is invoked, it will wait for space to become available and ensure that the eventData has been enqueued. When this call returns, the eventData has been accepted into the buffer, but it may not have been published yet. Publishing will take place at a nondeterministic point in the future as the buffer is processed.
      Parameters:
      eventData - The event to be enqueued into the buffer and, later, published.
      options - The set of options to apply when publishing this event. If partitionKey and partitionId are not set, then the event is distributed round-robin amongst all the partitions.
      Returns:
      The total number of events that are currently buffered and waiting to be published, across all partitions.
      Throws:
      NullPointerException - if eventData or options is null.
      IllegalArgumentException - if getPartitionId is set and is not valid.
      IllegalStateException - if the producer was closed while queueing an event.

    • enqueueEvents

      public Mono<Integer> enqueueEvents(Iterable<EventData> events)
      Enqueues a set of EventData into the buffer to be published to the Event Hub. If there is insufficient capacity in the buffer when this method is invoked, it will wait for space to become available and ensure that all EventData in the events set have been enqueued. When this call returns, the events have been accepted into the buffer, but it may not have been published yet. Publishing will take place at a nondeterministic point in the future as the buffer is processed.
      Parameters:
      events - The set of events to be enqueued into the buffer and, later, published.
      Returns:
      The total number of events that are currently buffered and waiting to be published, across all partitions.
      Throws:
      NullPointerException - if events is null.
      IllegalStateException - if the producer was closed while queueing an event.

    • enqueueEvents

      public Mono<Integer> enqueueEvents(Iterable<EventData> events, SendOptions options)
      Enqueues a set of EventData into the buffer to be published to the Event Hub. If there is insufficient capacity in the buffer when this method is invoked, it will wait for space to become available and ensure that all EventData in the events set have been enqueued. When this call returns, the events have been accepted into the buffer, but it may not have been published yet. Publishing will take place at a nondeterministic point in the future as the buffer is processed.
      Parameters:
      events - The set of events to be enqueued into the buffer and, later, published.
      options - The set of options to apply when publishing this event.
      Returns:
      The total number of events that are currently buffered and waiting to be published, across all partitions.
      Throws:
      NullPointerException - if eventData or options is null.
      IllegalArgumentException - if getPartitionId is set and is not valid.
      IllegalStateException - if the producer was closed while queueing an event.

    • flush

      public Mono<Void> flush()
      Attempts to publish all events in the buffer immediately. This may result in multiple batches being published, the outcome of each of which will be individually reported by the EventHubBufferedProducerClientBuilder.onSendBatchFailed(Consumer) and EventHubBufferedProducerClientBuilder.onSendBatchSucceeded(Consumer) handlers. Upon completion of this method, the buffer will be empty.
      Returns:
      A mono that completes when the buffers are empty.

    • close

      public void close()
      Disposes of the producer and all its resources.
      Specified by:
      close in interface AutoCloseable
      Specified by:
      close in interface Closeable