EventHubBufferedProducerAsyncClient Class
- java.
lang. Object - com.
azure. messaging. eventhubs. EventHubBufferedProducerAsyncClient
- com.
Implements
public final class EventHubBufferedProducerAsyncClient
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(Duration 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(Consumer<SendBatchFailedContext> sendFailedContext) and onSendBatchSucceeded(Consumer<SendBatchSucceededContext> sendSucceededContext).
- 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 maxWaitTime(Duration maxWaitTime) has elapsed, the buffer is full, or no more events can fit into a batch.
NOTE that Mono 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();
Method Summary
| Modifier and Type | Method and Description |
|---|---|
| void |
close()
Disposes of the producer and all its resources. |
| Mono<Integer> |
enqueueEvent(EventData eventData)
Enqueues an EventData into the buffer to be published to the Event Hub. |
| Mono<Integer> |
enqueueEvent(EventData eventData, SendOptions options)
Enqueues an EventData into the buffer to be published to the Event Hub. |
| Mono<Integer> |
enqueueEvents(Iterable<EventData> events)
Enqueues a set of EventData into the buffer to be published to the Event Hub. |
| Mono<Integer> |
enqueueEvents(Iterable<EventData> events, SendOptions options)
Enqueues a set of EventData into the buffer to be published to the Event Hub. |
| Mono<Void> |
flush()
Attempts to publish all events in the buffer immediately. |
| int |
getBufferedEventCount()
Gets the total number of events that are currently buffered and waiting to be published, across all partitions. |
| int |
getBufferedEventCount(String partitionId)
Gets the number of events that are buffered and waiting to be published for a given partition. |
| String |
getEventHubName()
Gets the Event Hub name this client interacts with. |
|
Mono<Event |
getEventHubProperties()
Retrieves information about an Event Hub, including the number of partitions present and their identifiers. |
| String |
getFullyQualifiedNamespace()
Gets the fully qualified Event Hubs namespace that the connection is associated with. |
| Flux<String> |
getPartitionIds()
Retrieves the identifiers for the partitions of an Event Hub. |
|
Mono<Partition |
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. |
Methods inherited from java.lang.Object
Method Details
close
public void close()
Disposes of the producer and all its resources.
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:
Returns:
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:
Returns:
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:
Returns:
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:
Returns:
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 onSendBatchFailed(Consumer<SendBatchFailedContext> sendFailedContext) and onSendBatchSucceeded(Consumer<SendBatchSucceededContext> sendSucceededContext) handlers. Upon completion of this method, the buffer will be empty.
Returns:
getBufferedEventCount
public int getBufferedEventCount()
Gets the total number of events that are currently buffered and waiting to be published, across all partitions.
Returns:
getBufferedEventCount
public int getBufferedEventCount(String partitionId)
Gets the number of events that are buffered and waiting to be published for a given partition.
Parameters:
Returns:
getEventHubName
public String getEventHubName()
Gets the Event Hub name this client interacts with.
Returns:
getEventHubProperties
public Mono<EventHubProperties> getEventHubProperties()
Retrieves information about an Event Hub, including the number of partitions present and their identifiers.
Returns:
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:
getPartitionIds
public Flux<String> getPartitionIds()
Retrieves the identifiers for the partitions of an Event Hub.
Returns:
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:
Returns:
