Benzene

Event Hub Stream Processing

Handle high-throughput Event Hub streams with Benzene, and understand exactly where Benzene's responsibility ends and the Azure Functions runtime's begins.

Problem Statement

You're ingesting a high-volume stream through Azure Event Hubs (telemetry, clickstream, change events) and want to process it with Benzene's message-handler pipeline instead of hand-rolling per-event dispatch logic. Doing this well means understanding a few things the Azure Functions getting-started guide doesn't go into:

This cookbook works through a realistic handler for a high-throughput scenario and answers each of those honestly, citing the actual source in src/Benzene.Azure.Function.EventHub/.

Prerequisites

Installation

dotnet add package Benzene.Azure.Function.EventHub --prerelease

This also needs Microsoft.Azure.Functions.Worker.Extensions.EventHubs (version 6.5.0 is what Benzene itself builds against) referenced directly in your function app project, same as the other Microsoft worker packages — see Azure Functions Setup, step 2.

dotnet add package Microsoft.Azure.Functions.Worker.Extensions.EventHubs --version 6.5.0

Step-by-Step Implementation

1. The trigger function and pipeline

This is the same shape as the Event Hubs section of the getting-started guide:

using Benzene.Azure.Function.EventHub;

app.UseEventHub(eventHub => eventHub
    .UseBenzeneMessage(direct => direct
        .UseMessageHandlers()));

using Azure.Messaging.EventHubs;
using Benzene.Azure.Function.Core;
using Benzene.Azure.Function.EventHub.Function;
using Microsoft.Azure.Functions.Worker;

public class TelemetryEventHubFunction
{
    private readonly IAzureFunctionApp _app;

    public TelemetryEventHubFunction(IAzureFunctionApp app)
    {
        _app = app;
    }

    [Function("telemetry-event-hub")]
    public Task Run(
        [EventHubTrigger("telemetry-hub", ConsumerGroup = "%TelemetryConsumerGroup%", Connection = "EventHubConnection")]
        EventData[] events)
    {
        return _app.HandleEventHub(events);
    }
}

ConsumerGroup is read from configuration here (%TelemetryConsumerGroup%) rather than hardcoded, which matters once you have more than one consumer of the same hub — see Troubleshooting below.

2. How Benzene actually processes a batch — read the source, not the assumption

[EventHubTrigger] hands you the whole batch as EventData[]. It's tempting to assume Benzene loops over that array and processes events one at a time, in partition order. It doesn't. Reading src/Benzene.Azure.Function.EventHub/Function/EventHubApplication.cs:

public class EventHubApplication : EntryPointMiddlewareApplication<EventData[]>
{
    public EventHubApplication(IMiddlewarePipeline<EventHubContext> pipelineBuilder, IServiceResolverFactory serviceResolverFactory)
        : base(new MiddlewareMultiApplication<EventData[], EventHubContext>(
                new TransportMiddlewarePipeline<EventHubContext>("event-hub", pipelineBuilder),
        @event => @event.Select(EventHubContext.CreateInstance).ToArray()),
            serviceResolverFactory)
    { }
}

Every EventData in the batch is wrapped in its own EventHubContext, and the fan-out is done by MiddlewareMultiApplication<TEvent, TContext> (src/Benzene.Core.Middleware/MiddlewareMultiApplication.cs):

public Task HandleAsync(TEvent @event, IServiceResolverFactory serviceResolverFactory)
{
    var tasks = mapper(@event).Select(async context =>
        {
            using var scope = serviceResolverFactory.CreateScope();
            await pipeline.HandleAsync(context, scope);
        })
        .ToArray();

    return Task.WhenAll(tasks);
}

Two consequences follow directly from this, and they matter for a high-throughput handler:

3. A realistic handler

using Benzene.Abstractions.MessageHandlers;
using Benzene.Abstractions.Results;
using Benzene.Core.MessageHandlers;
using Benzene.Results;

[Message("telemetry:reading")]
public class TelemetryReadingHandler : IMessageHandler<TelemetryReadingRequest, TelemetryReadingResponse>
{
    private readonly ITelemetryStore _store;

    public TelemetryReadingHandler(ITelemetryStore store)
    {
        _store = store;
    }

    public async Task<IBenzeneResult<TelemetryReadingResponse>> HandleAsync(TelemetryReadingRequest request)
    {
        await _store.RecordAsync(request.DeviceId, request.Value, request.Timestamp);
        return BenzeneResult.Ok(new TelemetryReadingResponse { Accepted = true });
    }
}

public class TelemetryReadingRequest
{
    public string DeviceId { get; set; }
    public double Value { get; set; }
    public DateTimeOffset Timestamp { get; set; }
}

public class TelemetryReadingResponse
{
    public bool Accepted { get; set; }
}

This handler is invoked because BenzeneMessageEventHubHandler (src/Benzene.Azure.Function.EventHub/Function/BenzeneMessageEventHubHandler.cs) deserializes each EventData.EventBody into a BenzeneMessageRequest { Topic, Headers, Body } envelope and only handles it if Topic is non-null:

protected override bool CanHandle(BenzeneMessageRequest request)
{
    return request?.Topic != null;
}

protected override BenzeneMessageRequest TryExtractRequest(EventHubContext context)
{
    try
    {
        return _serializer.Deserialize<BenzeneMessageRequest>(context.EventData.EventBody.ToString());
    }
    catch
    {
        return default;
    }
}

That means your Event Hub producer needs to publish that envelope shape — {"topic": "telemetry:reading", "body": "{...serialized TelemetryReadingRequest...}"} — not a bare JSON payload. If you're publishing from a Benzene client (another Benzene service, or a test), the MessageBuilder/AsEventHubBenzeneMessage() helpers produce this shape for you (see Testing below). If your producer is a non-Benzene device or service emitting raw JSON telemetry with no topic wrapper, CanHandle returns false for every event, and — because TryExtractRequest swallows deserialization exceptions and MiddlewareRouter.HandleAsync just calls next() when a router can't handle a request — the event is silently dropped with no error unless you've registered something else after UseBenzeneMessage to catch it. For non- enveloped producers, write your own IMiddleware<EventHubContext> (see the next section) rather than trying to force raw payloads through UseBenzeneMessage.

4. Reaching data that never makes it into the envelope

EventHubContext only exposes the raw EventData:

public class EventHubContext
{
    public static EventHubContext CreateInstance(EventData eventData) => new(eventData);
    public EventData EventData { get; }
}

Partition key, sequence number, enqueued time, and any custom application properties set by the producer (EventData.Properties) are all on that EventData object, but none of them flow into BenzeneMessageRequest/your handler's request type automatically. To use them — for metrics, filtering, or routing decisions — add your own middleware to the Event Hub pipeline, before UseBenzeneMessage:

using Azure.Messaging.EventHubs;
using Benzene.Abstractions.Middleware;
using Benzene.Azure.Function.EventHub;
using Microsoft.Extensions.Logging;

public class TelemetryEventMetadataMiddleware : IMiddleware<EventHubContext>
{
    private readonly ILogger<TelemetryEventMetadataMiddleware> _logger;

    public TelemetryEventMetadataMiddleware(ILogger<TelemetryEventMetadataMiddleware> logger)
    {
        _logger = logger;
    }

    public string Name => "TelemetryEventMetadata";

    public Task HandleAsync(EventHubContext context, Func<Task> next)
    {
        var eventData = context.EventData;

        // Custom application properties set by the producer (EventData.Properties), not part of
        // the Benzene envelope.
        if (eventData.Properties.TryGetValue("schema-version", out var schemaVersion)
            && !"2".Equals(schemaVersion?.ToString()))
        {
            _logger.LogWarning(
                "Skipping event with unsupported schema-version {SchemaVersion} on partition {PartitionKey}, sequence {SequenceNumber}",
                schemaVersion, eventData.PartitionKey, eventData.SequenceNumber);
            return Task.CompletedTask; // short-circuits: does not call next()
        }

        _logger.LogDebug(
            "Processing event enqueued at {EnqueuedTime} on partition {PartitionKey}, offset {Offset}",
            eventData.EnqueuedTime, eventData.PartitionKey, eventData.Offset);

        return next();
    }
}

Register it ahead of UseBenzeneMessage:

app.UseEventHub(eventHub => eventHub
    .Use(resolver => new TelemetryEventMetadataMiddleware(resolver.GetService<ILogger<TelemetryEventMetadataMiddleware>>()))
    .UseBenzeneMessage(direct => direct
        .UseMessageHandlers()));

EventData.Properties, PartitionKey, SequenceNumber, Offset, and EnqueuedTime are all standard Azure.Messaging.EventHubs.EventData members — nothing Benzene-specific — Benzene simply hands you the object untouched.

5. Batching and checkpointing are entirely the runtime's job

This is worth stating plainly: Benzene has no API for batch size, prefetch, or checkpointing. EventHubContext exposes exactly one property (EventData); there is no Checkpoint() method, no batch-size configuration, and nothing in Benzene.Azure.Function.EventHub's source references checkpointing at all. All of that is configured through host.json, entirely by the Microsoft.Azure.Functions.Worker.Extensions.EventHubs extension, independent of Benzene:

{
  "version": "2.0",
  "extensions": {
    "eventHubs": {
      "maxEventBatchSize": 100,
      "minEventBatchSize": 25,
      "maxWaitTime": "00:00:05",
      "batchCheckpointFrequency": 5,
      "prefetchCount": 300
    }
  }
}

None of the [EventHubTrigger] attribute properties on your trigger function change this either — ConsumerGroup and Connection are the only two that matter for routing/auth; batch tuning is host.json-only.

6. Checkpointing on failure, and why Benzene doesn't help with poison events

This is the part that's easy to get wrong, so it's worth being precise:

If you want poison events to interact with the runtime's retry policy, you have to bridge that gap yourself — Benzene does not do it for you. One way: inspect the result and rethrow inside your own trigger function or an Event Hub-level middleware, so the exception actually reaches the runtime:

using Benzene.Abstractions.Middleware;
using Benzene.Azure.Function.EventHub;
using Benzene.Core.Messages.BenzeneMessage;
using Benzene.Results;

public class RethrowOnServiceUnavailableMiddleware : IMiddleware<BenzeneMessageContext>
{
    public string Name => "RethrowOnServiceUnavailable";

    public async Task HandleAsync(BenzeneMessageContext context, Func<Task> next)
    {
        await next();

        if (context.BenzeneMessageResponse.StatusCode == BenzeneResultStatus.ServiceUnavailable)
        {
            // Escalate to an actual exception so host.json's retry policy (and, once retries are
            // exhausted, the runtime's failure logging) can act on it.
            throw new InvalidOperationException(
                $"Handler for topic '{context.BenzeneMessageRequest.Topic}' returned ServiceUnavailable");
        }
    }
}

Add it inside the direct-message pipeline, before UseMessageHandlers:

app.UseEventHub(eventHub => eventHub
    .UseBenzeneMessage(direct => direct
        .Use(resolver => new RethrowOnServiceUnavailableMiddleware())
        .UseMessageHandlers()));

Even with this in place, remember there's still no DLQ underneath — once retries (if any) are exhausted, the event is checkpointed past regardless. For real poison-message handling, log the raw EventData (body, partition, offset, sequence number) somewhere durable — a storage table, a separate "dead" Event Hub, a queue — from within your handler or a middleware, before deciding whether to let the failure surface, so you have something to replay or inspect later. Benzene gives you the hooks (middleware, EventHubContext.EventData) to build that; it doesn't ship it.

Testing

Use Benzene.Testing and the Event Hub test helpers, exactly as the message-handling tests in test/Benzene.Core.Test/Azure/EventHubPipelineTest.cs do:

using Benzene.Azure.Function.Core;
using Benzene.Azure.Function.EventHub.TestHelpers;
using Benzene.Testing;
using Microsoft.Extensions.DependencyInjection;
using Moq;

var mockTelemetryStore = new Mock<ITelemetryStore>();

var app = BenzeneTestHost.Create<StartUp>()
    .WithServices(services => services.AddScoped(_ => mockTelemetryStore.Object))
    .BuildAzureFunctionApp();

var request = MessageBuilder
    .Create("telemetry:reading", new TelemetryReadingRequest { DeviceId = "sensor-1", Value = 21.5, Timestamp = DateTimeOffset.UtcNow })
    .AsEventHubBenzeneMessage();

await app.HandleEventHub(request);

mockTelemetryStore.Verify(x => x.RecordAsync("sensor-1", 21.5, It.IsAny<DateTimeOffset>()));

HandleEventHub takes params EventData[], so you can pass a whole batch to exercise the parallel-fan-out behavior directly:

var batch = Enumerable.Range(0, 50)
    .Select(i => MessageBuilder.Create("telemetry:reading", new TelemetryReadingRequest { DeviceId = $"sensor-{i}", Value = i }).AsEventHubBenzeneMessage())
    .ToArray();

await app.HandleEventHub(batch);

mockTelemetryStore.Verify(x => x.RecordAsync(It.IsAny<string>(), It.IsAny<double>(), It.IsAny<DateTimeOffset>()), Times.Exactly(50));

For pipeline-only tests without a full StartUp, InlineAzureFunctionStartUp works the same way as it does in azure-functions.md's testing section:

var app = new InlineAzureFunctionStartUp()
    .ConfigureServices(services => services
        .UsingBenzene(x => x.AddMessageHandlers(typeof(TelemetryReadingHandler).Assembly))
        .AddSingleton(mockTelemetryStore.Object))
    .Configure(app => app
        .UseEventHub(eventHub => eventHub
            .UseBenzeneMessage(direct => direct
                .UseMessageHandlers())))
    .Build();

Troubleshooting

Partition and consumer-group misconfiguration

Handler never gets invoked, but no error appears

Per step 3: confirm the producer is actually publishing the {topic, body} envelope BenzeneMessageEventHubHandler expects. If Topic deserializes to null (or the body isn't valid JSON at all), the event is routed to next() in the pipeline and, if nothing else is registered to handle it, disappears with no exception, no log, no result — because CanHandle returning false is a normal "not for me" outcome, not a failure.

A single bad event seems to have no effect, but data loss still happened

Remember MessageHandler.HandleAsync converts your handler's exceptions into a ServiceUnavailable result rather than throwing (see step 6). If you're not logging inside the handler or checking the returned status somewhere, a systematically failing event type can churn through your pipeline indefinitely, checkpointing past every time, with nothing visible unless you've wired up logging/diagnostics (AddDiagnostics() — see Monitoring & Diagnostics) or the rethrow pattern above.

Ordering bugs that only show up under load

If your handler assumes events for the same key arrive and complete in order, but you're seeing interleaved writes, revisit step 2 — MiddlewareMultiApplication processes every event in a batch concurrently via Task.WhenAll. Partition-level ordering from Event Hubs does not survive into per-event concurrent handler execution.

Variations

Kafka-compatible endpoint

Event Hubs' Kafka-compatible endpoint is handled by a separate package (Benzene.Azure.Function.Kafka, app.UseKafka(...)), with its own trigger shape ([KafkaTrigger]/KafkaRecord[]). There is no UseBenzeneMessage envelope bridge for Kafka today — dispatch there is by [Message]/topic directly on the decoded UTF-8 JSON body. See the Kafka section of the getting-started guide.

Correlation across a batch

Each event in a batch gets its own DI scope, so IBenzeneInvocation (populated via app.UseBenzeneInvocation() — see Correlation and tracing) resolves independently per event, not once per Function invocation. If you need a single correlation identifier that spans the whole triggered batch (rather than per-event), you'll need to generate and thread it through yourself, e.g. via a middleware ahead of the per-event fan-out that stashes a batch ID somewhere your per-event middleware can pick up.

Consuming without Azure Functions (self-hosted worker)

This whole cookbook assumes an Azure Functions Event Hub trigger, where — as step 5 stresses — batching and checkpointing are entirely the runtime's job. If you want to consume an event hub from a long-running process you own instead, use Benzene.Azure.EventHub (not Benzene.Azure.Function.EventHub). It's a self-hosted worker: worker.UseEventHub(config, clientFactory, eh => eh.UseMessageHandlers()) runs the SDK's EventProcessorClient and dispatches each event through the same [Message]/topic model (from the event's "topic" property).

The key inversion from the trigger: here Benzene owns what the runtime owned above.

See Worker Service Setup, Part B for the full host wiring.

Further Reading