EventHubProducerAsyncClient Class

An asynchronous producer responsible for transmitting EventData to a specific Event Hub, grouped together in batches. Depending on the CreateBatchOptions specified when creating an EventDataBatch, the events may be automatically routed to an available partition or specific to a partition. More information and specific recommendations for strategies to use when publishing events is in: Distribute events to partitions

Allowing automatic routing of partitions is recommended when:

• The sending of events needs to be highly available.
• The event data should be evenly distributed among all available partitions.

If no partition id is specified, the following rules are used for automatically selecting one:

1. Distribute the events equally amongst all available partitions using a round-robin approach.
2. If a partition becomes unavailable, the Event Hubs service will automatically detect it and forward the message to another available 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".

Sample: Construct a EventHubProducerAsyncClient

The following code sample demonstrates the creation of the asynchronous client EventHubProducerAsyncClient. 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.

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.
 EventHubProducerAsyncClient producer = new EventHubClientBuilder()
     .credential("<<fully-qualified-namespace>>", "<<event-hub-name>>",
         credential)
     .buildAsyncProducerClient();

Sample: Create a producer and publish events to any partition

The following code sample demonstrates how to send a set of events to Event Hub. Events are distributed via automatic routing because no options were set when creating the EventDataBatch via createBatch(). Using EventDataBatch is recommended because multiple events can be sent across the underlying connection with a single message.

createBatch() and send(EventDataBatch batch) are non-blocking calls. After setting up the operation, its async representation is returned. The Mono must be subscribed to, like the sample below, to start publishing the event batch.

// Creating a batch without options set, will allow for automatic routing of events to any partition.
 producer.createBatch().flatMap(batch -> {
     batch.tryAdd(new EventData("test-event-1"));
     batch.tryAdd(new EventData("test-event-2"));
     return producer.send(batch);
 }).subscribe(unused -> {
 },
     error -> System.err.println("Error occurred while sending batch:" + error),
     () -> System.out.println("Send complete."));

Sample: Publish events to partition "1"

The following code sample demonstrates how to send a set of events to Event Hub to partition "1". EventDataBatch via createBatch(CreateBatchOptions options). Partition identifiers can be obtained using getPartitionIds(). Using EventDataBatch is recommended because multiple events can be sent across the underlying connection with a single message.

createBatch(CreateBatchOptions options) and send(EventDataBatch batch) are non-blocking calls. After setting up the operation, its async representation is returned. The Mono must be subscribed to, like the sample below, to start publishing the event batch.

CreateBatchOptions options = new CreateBatchOptions().setPartitionId("1");
 producer.createBatch(options).flatMap(batch -> {
     batch.tryAdd(new EventData("test-event-1"));
     batch.tryAdd(new EventData("test-event-2"));
     return producer.send(batch);
 }).subscribe(unused -> {
 },
     error -> System.err.println("Error occurred while sending batch to partition 1:" + error),
     () -> System.out.println("Send to partition 1 complete."));

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

In the code sample below, all events with the same partition key, "bread" are sent to the same partition. When setPartitionId(String partitionId) is specified, it tells the Event Hubs service that these events belong to the same group and should belong to the same partition. Useful in the case where developers want events to end up in the same partition, but do not care which partition it ends up in.

createBatch(CreateBatchOptions options) and send(EventDataBatch batch) are non-blocking calls. After setting up the operation, its async representation is returned. The Mono must be subscribed to, like the sample below, to start publishing the event batch.

CreateBatchOptions options = new CreateBatchOptions().setPartitionKey("bread");

 producer.createBatch(options).flatMap(batch -> {
     batch.tryAdd(new EventData("sourdough"));
     batch.tryAdd(new EventData("rye"));
     return producer.send(batch);
 }).subscribe(unused -> {
 },
     error -> System.err.println("Error occurred while sending batch:" + error),
     () -> System.out.println("Send complete."));

Sample: Publish events using a size-limited EventDataBatch

In the code sample below, all batches are created with a max size of 256 bytes using setMaximumSizeInBytes(int maximumSizeInBytes) is specified. Events inside the batch are automatically routed because no partition id or partition key are specified.

createBatch(CreateBatchOptions options) and send(EventDataBatch batch) are non-blocking calls. After setting up the operation, its async representation is returned. The Mono must be subscribed to, like the sample below, to start publishing the event batch.

Flux<EventData> telemetryEvents = Flux.just(firstEvent, secondEvent);

 // Setting `setMaximumSizeInBytes` when creating a batch, limits the size of that batch.
 // In this case, all the batches created with these options are limited to 256 bytes.
 CreateBatchOptions options = new CreateBatchOptions().setMaximumSizeInBytes(256);
 AtomicReference<EventDataBatch> currentBatch = new AtomicReference<>(
     producer.createBatch(options).block());

 // The sample Flux contains two events, but it could be an infinite stream of telemetry events.
 Disposable publishingOperation = telemetryEvents.flatMap(event -> {
     EventDataBatch batch = currentBatch.get();

     if (batch.tryAdd(event)) {
         return Mono.empty();
     }

     // Send the current batch then create another size-limited EventDataBatch and try to fit the event into
     // this new batch.
     return producer.send(batch).then(
         producer.createBatch(options).map(newBatch -> {
             currentBatch.set(newBatch);

             // Add the event that did not fit in the previous batch.
             if (!newBatch.tryAdd(event)) {
                 return Mono.error(new IllegalArgumentException(
                     "Event was too large to fit in an empty batch. Max size: "
                         + newBatch.getMaxSizeInBytes()));
             }

             return Mono.empty();
         }));
 }).subscribe(unused -> {
 }, error -> {
     System.out.println("Error occurred publishing events: " + error);
 }, () -> {
     System.out.println("Completed publishing operation.");
 });