Producing Messages to Azure Event Hubs

This guide will walk you through the process of sending messages (events) to an Azure Event Hub using various programming languages and SDKs.

Using the Azure SDKs

Azure provides robust SDKs for various languages that simplify the process of interacting with Event Hubs. We'll cover examples in C#, Python, and Node.js.

C# Example

First, ensure you have the Azure.Messaging.EventHubs NuGet package installed.

using Azure.Messaging.EventHubs;
using System;
using System.Text;
using System.Threading.Tasks;

public class EventHubProducer
{
    private const string connectionString = "";
    private const string eventHubName = "";

    public static async Task Main(string[] args)
    {
        await using var producerClient = new EventHubProducerClient(connectionString, eventHubName);

        try
        {
            using EventDataBatch eventBatch = await producerClient.CreateBatchAsync();

            for (int i = 1; i <= 5; i++)
            {
                var eventBody = $"Message {i}";
                if (!eventBatch.TryAddMessage(new EventData(Encoding.UTF8.GetBytes(eventBody))))
                {
                    throw new Exception($"The event {i} is too large for the batch and cannot be sent.");
                }
            }

            await producerClient.SendAsync(eventBatch);
            Console.WriteLine("A batch of events has been sent.");
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Error sending events: {ex.Message}");
        }
    }
}

Python Example

Install the SDK using pip:

pip install azure-eventhub

Then, use the following code:

import os
from azure.eventhub import EventHubProducerClient, EventData

CONNECTION_STR = os.environ["EVENTHUB_CONNECTION_STR"]
EVENTHUB_NAME = os.environ["EVENTHUB_NAME"]

async def run():
    # Create a producer client to send events to the Event Hub
    producer = EventHubProducerClient.from_connection_string(
        CONNECTION_STR, event_hub_name=EVENTHUB_NAME
    )

    async with producer:
        # Prepare a batch of events
        batch = await producer.create_batch()
        for i in range(5):
            event_body = f"Message {i+1}"
            if not batch.add(EventData(event_body)):
                print(f"Batch is full. Skipping message {i+1}")
                break
        
        # Send the batch of events
        await producer.send_batch(batch)
        print("Batch of events sent.")

if __name__ == "__main__":
    import asyncio
    asyncio.run(run())

Node.js Example

Install the necessary packages:

npm install @azure/event-hubs

Use the following JavaScript code:

const { EventHubProducerClient } = require("@azure/event-hubs");

const connectionString = process.env.EVENTHUB_CONNECTION_STRING;
const eventHubName = process.env.EVENTHUB_NAME;

async function main() {
    const producerClient = new EventHubProducerClient(connectionString, eventHubName);

    try {
        const batch = await producerClient.createBatch();

        for (let i = 1; i <= 5; i++) {
            const eventBody = `Message ${i}`;
            if (!batch.tryAdd({ body: eventBody })) {
                console.log(`Message ${i} is too large for the batch and cannot be sent.`);
                break;
            }
        }

        await producerClient.sendBatch(batch);
        console.log("A batch of events has been sent.");

    } catch (err) {
        console.error("Error sending events:", err);
    } finally {
        await producerClient.close();
    }
}

main().catch(error => {
    console.error("An error occurred:", error);
});

Key Concepts for Producing Messages

Event Data Structure

Each message sent to Event Hubs is represented as an EventData object (or equivalent in different SDKs). This object typically contains:

• Body: The actual payload of the message, usually a byte array.
• Properties: Custom metadata associated with the event.
• PartitionKey: Used for session affinity, ensuring messages with the same PartitionKey are sent to the same partition.
• Sequence Number: Assigned by Event Hubs upon ingestion.
• Offset: The position of the event within a partition.

Batching

For efficiency, it's recommended to send messages in batches. The SDKs provide methods to create and manage these batches. There are limits to the size of a single batch (typically 1MB). The TryAddMessage or add methods will return false if a message cannot be added to the current batch.

Partitioning

Event Hubs partitions messages across multiple partitions for scalability and parallel processing. When sending messages:

• If you specify a PartitionKey, Event Hubs deterministically assigns the message to a partition based on the hash of the key. This is useful for maintaining order within a specific logical stream of events.
• If you do not specify a PartitionKey, Event Hubs will distribute messages across partitions in a round-robin fashion (or based on other internal load-balancing strategies).

Error Handling

Always implement robust error handling. Common issues include:

• Network connectivity problems.
• Incorrect connection strings or credentials.
• Messages exceeding the maximum allowed size.
• Throttling by Azure services.

The SDKs throw exceptions for errors, which should be caught and handled appropriately. Consider implementing retry mechanisms for transient errors.

Next Steps

Once you've mastered producing messages, you'll want to learn how to consume them. Refer to our guide on Consuming Messages.