Azure Eventhub Dotnet

1---
2name: azure-eventhub-dotnet
3description: |
4  Azure Event Hubs SDK for .NET. Use for high-throughput event streaming: sending events (EventHubProducerClient, EventHubBufferedProducerClient), receiving events (EventProcessorClient with checkpointing), partition management, and real-time data ingestion. Triggers: "Event Hubs", "event streaming", "EventHubProducerClient", "EventProcessorClient", "send events", "receive events", "checkpointing", "partition".
5package: Azure.Messaging.EventHubs
6---
7 
8# Azure.Messaging.EventHubs (.NET)
9 
10High-throughput event streaming SDK for sending and receiving events via Azure Event Hubs.
11 
12## Installation
13 
14```bash
15# Core package (sending and simple receiving)
16dotnet add package Azure.Messaging.EventHubs
17 
18# Processor package (production receiving with checkpointing)
19dotnet add package Azure.Messaging.EventHubs.Processor
20 
21# Authentication
22dotnet add package Azure.Identity
23 
24# For checkpointing (required by EventProcessorClient)
25dotnet add package Azure.Storage.Blobs
26```
27 
28**Current Versions**: Azure.Messaging.EventHubs v5.12.2, Azure.Messaging.EventHubs.Processor v5.12.2
29 
30## Environment Variables
31 
32```bash
33EVENTHUB_FULLY_QUALIFIED_NAMESPACE=<namespace>.servicebus.windows.net
34EVENTHUB_NAME=<event-hub-name>
35 
36# For checkpointing (EventProcessorClient)
37BLOB_STORAGE_CONNECTION_STRING=<storage-connection-string>
38BLOB_CONTAINER_NAME=<checkpoint-container>
39 
40# Alternative: Connection string auth (not recommended for production)
41EVENTHUB_CONNECTION_STRING=Endpoint=sb://<namespace>.servicebus.windows.net/;SharedAccessKeyName=...
42```
43 
44## Authentication
45 
46```csharp
47using Azure.Identity;
48using Azure.Messaging.EventHubs;
49using Azure.Messaging.EventHubs.Producer;
50 
51// Always use DefaultAzureCredential for production
52var credential = new DefaultAzureCredential();
53 
54var fullyQualifiedNamespace = Environment.GetEnvironmentVariable("EVENTHUB_FULLY_QUALIFIED_NAMESPACE");
55var eventHubName = Environment.GetEnvironmentVariable("EVENTHUB_NAME");
56 
57var producer = new EventHubProducerClient(
58    fullyQualifiedNamespace,
59    eventHubName,
60    credential);
61```
62 
63**Required RBAC Roles**:
64- **Sending**: `Azure Event Hubs Data Sender`
65- **Receiving**: `Azure Event Hubs Data Receiver`
66- **Both**: `Azure Event Hubs Data Owner`
67 
68## Client Types
69 
70| Client | Purpose | When to Use |
71|--------|---------|-------------|
72| `EventHubProducerClient` | Send events immediately in batches | Real-time sending, full control over batching |
73| `EventHubBufferedProducerClient` | Automatic batching with background sending | High-volume, fire-and-forget scenarios |
74| `EventHubConsumerClient` | Simple event reading | Prototyping only, NOT for production |
75| `EventProcessorClient` | Production event processing | **Always use this for receiving in production** |
76 
77## Core Workflow
78 
79### 1. Send Events (Batch)
80 
81```csharp
82using Azure.Identity;
83using Azure.Messaging.EventHubs;
84using Azure.Messaging.EventHubs.Producer;
85 
86await using var producer = new EventHubProducerClient(
87    fullyQualifiedNamespace,
88    eventHubName,
89    new DefaultAzureCredential());
90 
91// Create a batch (respects size limits automatically)
92using EventDataBatch batch = await producer.CreateBatchAsync();
93 
94// Add events to batch
95var events = new[]
96{
97    new EventData(BinaryData.FromString("{\"id\": 1, \"message\": \"Hello\"}")),
98    new EventData(BinaryData.FromString("{\"id\": 2, \"message\": \"World\"}"))
99};
100 
101foreach (var eventData in events)
102{
103    if (!batch.TryAdd(eventData))
104    {
105        // Batch is full - send it and create a new one
106        await producer.SendAsync(batch);
107        batch = await producer.CreateBatchAsync();
108        
109        if (!batch.TryAdd(eventData))
110        {
111            throw new Exception("Event too large for empty batch");
112        }
113    }
114}
115 
116// Send remaining events
117if (batch.Count > 0)
118{
119    await producer.SendAsync(batch);
120}
121```
122 
123### 2. Send Events (Buffered - High Volume)
124 
125```csharp
126using Azure.Messaging.EventHubs.Producer;
127 
128var options = new EventHubBufferedProducerClientOptions
129{
130    MaximumWaitTime = TimeSpan.FromSeconds(1)
131};
132 
133await using var producer = new EventHubBufferedProducerClient(
134    fullyQualifiedNamespace,
135    eventHubName,
136    new DefaultAzureCredential(),
137    options);
138 
139// Handle send success/failure
140producer.SendEventBatchSucceededAsync += args =>
141{
142    Console.WriteLine($"Batch sent: {args.EventBatch.Count} events");
143    return Task.CompletedTask;
144};
145 
146producer.SendEventBatchFailedAsync += args =>
147{
148    Console.WriteLine($"Batch failed: {args.Exception.Message}");
149    return Task.CompletedTask;
150};
151 
152// Enqueue events (sent automatically in background)
153for (int i = 0; i < 1000; i++)
154{
155    await producer.EnqueueEventAsync(new EventData($"Event {i}"));
156}
157 
158// Flush remaining events before disposing
159await producer.FlushAsync();
160```
161 
162### 3. Receive Events (Production - EventProcessorClient)
163 
164```csharp
165using Azure.Identity;
166using Azure.Messaging.EventHubs;
167using Azure.Messaging.EventHubs.Consumer;
168using Azure.Messaging.EventHubs.Processor;
169using Azure.Storage.Blobs;
170 
171// Blob container for checkpointing
172var blobClient = new BlobContainerClient(
173    Environment.GetEnvironmentVariable("BLOB_STORAGE_CONNECTION_STRING"),
174    Environment.GetEnvironmentVariable("BLOB_CONTAINER_NAME"));
175 
176await blobClient.CreateIfNotExistsAsync();
177 
178// Create processor
179var processor = new EventProcessorClient(
180    blobClient,
181    EventHubConsumerClient.DefaultConsumerGroup,
182    fullyQualifiedNamespace,
183    eventHubName,
184    new DefaultAzureCredential());
185 
186// Handle events
187processor.ProcessEventAsync += async args =>
188{
189    Console.WriteLine($"Partition: {args.Partition.PartitionId}");
190    Console.WriteLine($"Data: {args.Data.EventBody}");
191    
192    // Checkpoint after processing (or batch checkpoints)
193    await args.UpdateCheckpointAsync();
194};
195 
196// Handle errors
197processor.ProcessErrorAsync += args =>
198{
199    Console.WriteLine($"Error: {args.Exception.Message}");
200    Console.WriteLine($"Partition: {args.PartitionId}");
201    return Task.CompletedTask;
202};
203 
204// Start processing
205await processor.StartProcessingAsync();
206 
207// Run until cancelled
208await Task.Delay(Timeout.Infinite, cancellationToken);
209 
210// Stop gracefully
211await processor.StopProcessingAsync();
212```
213 
214### 4. Partition Operations
215 
216```csharp
217// Get partition IDs
218string[] partitionIds = await producer.GetPartitionIdsAsync();
219 
220// Send to specific partition (use sparingly)
221var options = new SendEventOptions
222{
223    PartitionId = "0"
224};
225await producer.SendAsync(events, options);
226 
227// Use partition key (recommended for ordering)
228var batchOptions = new CreateBatchOptions
229{
230    PartitionKey = "customer-123"  // Events with same key go to same partition
231};
232using var batch = await producer.CreateBatchAsync(batchOptions);
233```
234 
235## EventPosition Options
236 
237Control where to start reading:
238 
239```csharp
240// Start from beginning
241EventPosition.Earliest
242 
243// Start from end (new events only)
244EventPosition.Latest
245 
246// Start from specific offset
247EventPosition.FromOffset(12345)
248 
249// Start from specific sequence number
250EventPosition.FromSequenceNumber(100)
251 
252// Start from specific time
253EventPosition.FromEnqueuedTime(DateTimeOffset.UtcNow.AddHours(-1))
254```
255 
256## ASP.NET Core Integration
257 
258```csharp
259// Program.cs
260using Azure.Identity;
261using Azure.Messaging.EventHubs.Producer;
262using Microsoft.Extensions.Azure;
263 
264builder.Services.AddAzureClients(clientBuilder =>
265{
266    clientBuilder.AddEventHubProducerClient(
267        builder.Configuration["EventHub:FullyQualifiedNamespace"],
268        builder.Configuration["EventHub:Name"]);
269    
270    clientBuilder.UseCredential(new DefaultAzureCredential());
271});
272 
273// Inject in controller/service
274public class EventService
275{
276    private readonly EventHubProducerClient _producer;
277    
278    public EventService(EventHubProducerClient producer)
279    {
280        _producer = producer;
281    }
282    
283    public async Task SendAsync(string message)
284    {
285        using var batch = await _producer.CreateBatchAsync();
286        batch.TryAdd(new EventData(message));
287        await _producer.SendAsync(batch);
288    }
289}
290```
291 
292## Best Practices
293 
2941. **Use `EventProcessorClient` for receiving** — Never use `EventHubConsumerClient` in production
2952. **Checkpoint strategically** — After N events or time interval, not every event
2963. **Use partition keys** — For ordering guarantees within a partition
2974. **Reuse clients** — Create once, use as singleton (thread-safe)
2985. **Use `await using`** — Ensures proper disposal
2996. **Handle `ProcessErrorAsync`** — Always register error handler
3007. **Batch events** — Use `CreateBatchAsync()` to respect size limits
3018. **Use buffered producer** — For high-volume scenarios with automatic batching
302 
303## Error Handling
304 
305```csharp
306using Azure.Messaging.EventHubs;
307 
308try
309{
310    await producer.SendAsync(batch);
311}
312catch (EventHubsException ex) when (ex.Reason == EventHubsException.FailureReason.ServiceBusy)
313{
314    // Retry with backoff
315    await Task.Delay(TimeSpan.FromSeconds(5));
316}
317catch (EventHubsException ex) when (ex.IsTransient)
318{
319    // Transient error - safe to retry
320    Console.WriteLine($"Transient error: {ex.Message}");
321}
322catch (EventHubsException ex)
323{
324    // Non-transient error
325    Console.WriteLine($"Error: {ex.Reason} - {ex.Message}");
326}
327```
328 
329## Checkpointing Strategies
330 
331| Strategy | When to Use |
332|----------|-------------|
333| Every event | Low volume, critical data |
334| Every N events | Balanced throughput/reliability |
335| Time-based | Consistent checkpoint intervals |
336| Batch completion | After processing a logical batch |
337 
338```csharp
339// Checkpoint every 100 events
340private int _eventCount = 0;
341 
342processor.ProcessEventAsync += async args =>
343{
344    // Process event...
345    
346    _eventCount++;
347    if (_eventCount >= 100)
348    {
349        await args.UpdateCheckpointAsync();
350        _eventCount = 0;
351    }
352};
353```
354 
355## Related SDKs
356 
357| SDK | Purpose | Install |
358|-----|---------|---------|
359| `Azure.Messaging.EventHubs` | Core sending/receiving | `dotnet add package Azure.Messaging.EventHubs` |
360| `Azure.Messaging.EventHubs.Processor` | Production processing | `dotnet add package Azure.Messaging.EventHubs.Processor` |
361| `Azure.ResourceManager.EventHubs` | Management plane (create hubs) | `dotnet add package Azure.ResourceManager.EventHubs` |
362| `Microsoft.Azure.WebJobs.Extensions.EventHubs` | Azure Functions binding | `dotnet add package Microsoft.Azure.WebJobs.Extensions.EventHubs` |
363