microsoft / microsoft/aspire

Add builder pattern API for multiple Event Hubs in Aspire.Azure.Messaging.EventHubs

Open
#12,181 0 comments 0 reactions 0 assignees View on GitHub
area-integrations
Dominant language
C#
Stars
6.3k
Forks
991
Avg merge
2d 15h
Merged PRs (30d)
196

Description

## Is there an existing issue for this?

- [x] I have searched the existing issues

## Is your feature request related to a problem? Please describe the problem.

Currently, if you want to work with multiple Event Hubs from the same namespace, you must register each client separately, potentially creating duplicate connections:

```csharp
// Current approach - separate registrations
builder.AddAzureEventHubProducerClient("telemetry");
builder.AddAzureEventHubProducerClient("logs");
builder.AddAzureEventHubProducerClient("metrics");

builder.AddAzureEventHubConsumerClient("orders");
builder.AddAzureEventProcessorClient("payments");

// Each registration may create separate connections
// No clear way to group clients by namespace
// Can't easily inject specific producers/consumers by Event Hub name
```

The Cosmos DB integration already solves this with `AddAzureCosmosDatabase()` and `CosmosDatabaseBuilder`, and similar proposals exist for Blob Storage (#12179) and Service Bus (#12180), but Event Hubs lacks an equivalent pattern.

## Describe the solution you'd like

Add a **builder pattern API** that allows registering multiple producers/consumers for different Event Hubs within the same namespace:

```csharp
// Proposed API
builder.AddAzureEventHubNamespace("events")
.AddKeyedProducer("telemetry")
.AddKeyedProducer("logs")
.AddKeyedProducer("metrics")
.AddKeyedConsumer("orders", consumerGroup: "$Default")
.AddKeyedProcessor("payments", consumerGroup: "$Default", checkpointStore: "checkpoints")
.AddKeyedBufferedProducer("analytics", options =>
{
options.MaximumWaitTime = TimeSpan.FromSeconds(5);
});

// Usage in services
public class TelemetryService(
[FromKeyedServices("telemetry")] EventHubProducerClient telemetryProducer,
[FromKeyedServices("logs")] EventHubProducerClient logsProducer,
[FromKeyedServices("metrics")] EventHubProducerClient metricsProducer)
{
// All producers share connection to the same Event Hubs namespace
public async Task LogEvent(string eventData)
{
await telemetryProducer.SendAsync([new EventData(eventData)]);
}
}

public class OrderProcessor(
[FromKeyedServices("orders")] EventHubConsumerClient ordersConsumer) : BackgroundService
{
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await foreach (var partition in ordersConsumer.ReadEventsAsync(stoppingToken))
{
// Process order events
}
}
}
```

### Benefits

1. **Resource Efficiency**: Share connection/namespace configuration across multiple Event Hubs
2. **Better DI Integration**: Producers/consumers registered as keyed services by Event Hub name
3. **Cleaner Code**: Fluent builder API for declaring multiple Event Hubs in same namespace
4. **Logical Grouping**: Clear namespace-level organization of Event Hubs
5. **Consistency**: Matches patterns in Cosmos DB, Blob Storage (#12179), and Service Bus (#12180)
6. **Type Safety**: Keyed services allow injecting specific clients by Event Hub name

### Implementation Requirements

1. **New `EventHubNamespaceBuilder` class**:
- Constructor accepting `IHostApplicationBuilder`, connection name, settings
- Store namespace-level configuration (fully qualified namespace, credentials)
- `AddKeyedProducer(string eventHubName, Action?)` - Register producer for specific Event Hub
- `AddKeyedBufferedProducer(string eventHubName, Action?)` - Register buffered producer
- `AddKeyedConsumer(string eventHubName, string consumerGroup, Action?)` - Register consumer
- `AddKeyedProcessor(string eventHubName, string consumerGroup, string checkpointStoreConnectionName, Action?)` - Register processor with checkpoint store
- `AddKeyedPartitionReceiver(string eventHubName, string consumerGroup, string partitionId, Action?)` - Register partition receiver

2. **New extension methods** in `AspireEventHubsExtensions`:
- `AddAzureEventHubNamespace()` - Returns `EventHubNamespaceBuilder`
- `AddKeyedAzureEventHubNamespace()` - Returns `EventHubNamespaceBuilder` with keyed namespace

3. **Playground sample enhancement**:
- Update `AspireEventHub` playground to use new builder pattern
- Demonstrate multiple Event Hubs in same namespace
- Show producer, consumer, and processor registration
- Example with checkpoint store integration

### API Consistency

**Cosmos DB (existing):**
```csharp
builder.AddAzureCosmosDatabase("db")
.AddKeyedContainer("entries")
.AddKeyedContainer("users");
```

**Blob Storage (proposed - #12179):**
```csharp
builder.AddAzureBlobService("storage")
.AddKeyedContainer("images")
.AddKeyedContainer("documents");
```

**Service Bus (proposed - #12180):**
```csharp
builder.AddAzureServiceBus("messaging")
.AddKeyedSender("orders")
.AddKeyedProcessor("commands");
```

**Event Hubs (this proposal):**
```csharp
builder.AddAzureEventHubNamespace("events")
.AddKeyedProducer("telemetry")
.AddKeyedConsumer("orders", consumerGroup: "$Default")
.AddKeyedProcessor("payments", "$Default", "checkpoints");
```

### Example Usage Scenarios

#### Scenario 1: Multiple Producers in Same Namespace
```csharp
builder.AddAzureEventHubNamespace("monitoring")
.AddKeyedProducer("telemetry")
.AddKeyedProducer("logs")
.AddKeyedProducer("metrics")
.AddKeyedProducer("traces");
```

#### Scenario 2: Producer + Consumer for Same Event Hub
```csharp
builder.AddAzureEventHubNamespace("orders")
.AddKeyedProducer("order-events")
.AddKeyedConsumer("order-events", "$Default");
```

#### Scenario 3: Event Processor with Checkpoint Store
```csharp
builder.AddAzureBlobServiceClient("checkpoints");

builder.AddAzureEventHubNamespace("transactions")
.AddKeyedProcessor("payments", "$Default", "checkpoints", options =>
{
options.LoadBalancingUpdateInterval = TimeSpan.FromSeconds(10);
})
.AddKeyedProcessor("refunds", "$Default", "checkpoints");
```

#### Scenario 4: Multiple Consumer Groups
```csharp
builder.AddAzureEventHubNamespace("analytics")
.AddKeyedProducer("raw-events")
.AddKeyedConsumer("raw-events", "aggregator")
.AddKeyedConsumer("raw-events", "archiver")
.AddKeyedConsumer("raw-events", "real-time");
```

#### Scenario 5: Mixed Producers, Consumers, and Processors
```csharp
builder.AddAzureEventHubNamespace("platform")
.AddKeyedProducer("commands")
.AddKeyedConsumer("events", "$Default")
.AddKeyedBufferedProducer("metrics", options =>
{
options.MaximumWaitTime = TimeSpan.FromSeconds(1);
})
.AddKeyedProcessor("notifications", "email-group", "checkpoints");
```

## Additional context

This feature would bring the Event Hubs integration to feature parity with Cosmos DB and aligns with proposed enhancements for Blob Storage (#12179) and Service Bus (#12180). It provides a consistent API pattern across all Azure integrations in Aspire.

**Additional Considerations:**
- **Checkpoint Store Requirement**: `EventProcessorClient` requires a checkpoint store (typically Blob Storage). The builder should validate that a checkpoint store connection is provided
- **Consumer Groups**: Each consumer/processor requires a consumer group. Should default to `$Default` but allow override
- **Connection String Parsing**: Event Hub names are typically in the connection string (`EntityPath=myhub`), but the builder should allow explicit specification for clarity
- **Buffered Producer Lifecycle**: `EventHubBufferedProducerClient` has automatic flushing - users should be aware of disposal/lifetime management

**Migration Path:**
Existing code continues to work unchanged. The builder pattern is purely additive:
```csharp
// Still supported
builder.AddAzureEventHubProducerClient("telemetry");

// New pattern (optional)
builder.AddAzureEventHubNamespace("events")
.AddKeyedProducer("telemetry");
```

Reference implementation:
- `src/Components/Aspire.Microsoft.Azure.Cosmos/CosmosDatabaseBuilder.cs`
- `src/Components/Aspire.Microsoft.Azure.Cosmos/AspireMicrosoftAzureCosmosExtensions.cs`
- `playground/AspireEventHub/EventHubsConsumer/Program.cs`
- `playground/AspireEventHub/EventHubsApi/Program.cs`

Related issues: #12179 (Blob Storage), #12180 (Service Bus)

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.