### Batch Consumer Console Output Example

Source: https://github.com/emitdotnet/emit/blob/main/samples/batch-consumer/README.md

Example console logs showing the simulator phases and batch processing details. Observe batch sizes and reroute counts.

```text
[Simulator] Phase: Normal (30s, ~30/sec)
[Simulator] Phase: Burst (8s, ~142/sec)
[Simulator] Phase: Quiet (15s, ~5/sec)
```

```text
Processing batch of 25 scans
Mis-sorted: PKG-04821 on lane 3, should be lane 1 (zip 1847)
Batch complete: 25 scans, 2 reroutes, 14 ms
```

--------------------------------

### Start Infrastructure with Docker Compose

Source: https://github.com/emitdotnet/emit/blob/main/samples/README.md

Use this command to start the shared infrastructure for all Emit samples. Ensure MongoDB has sufficient time to elect its primary before running a sample.

```bash
docker compose up -d
```

--------------------------------

### Install Emit Packages (.NET CLI)

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/getting-started/quickstart.mdx

Install the necessary Emit packages using the .NET CLI. This includes the core Emit package, MongoDB integration, Kafka integration, and the JSON serializer for Kafka.

```bash
dotnet add package Emit
dotnet add package Emit.MongoDB
dotnet add package Emit.Kafka
dotnet add package Emit.Kafka.JsonSerializer
```

--------------------------------

### Add Kafka Package

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/kafka/setup.mdx

Install the Emit.Kafka package using the .NET CLI.

```bash
dotnet add package Emit.Kafka
```

--------------------------------

### Install Emit Packages (PackageReference)

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/getting-started/quickstart.mdx

Install the necessary Emit packages using the PackageReference format in your .csproj file. This includes the core Emit package, MongoDB integration, Kafka integration, and the JSON serializer for Kafka.

```xml
<PackageReference Include="Emit" />
<PackageReference Include="Emit.MongoDB" />
<PackageReference Include="Emit.Kafka" />
<PackageReference Include="Emit.Kafka.JsonSerializer" />
```

--------------------------------

### Local Development Commands

Source: https://github.com/emitdotnet/emit/blob/main/docs/README.md

Run these commands to start a local development server, build the static site, or preview the built site.

```bash
npm run dev    # http://localhost:4321 — live reload
npm run build  # static output → dist/
npm run preview # preview the built site locally
```

--------------------------------

### Run Inventory Snapshot Worker (PostgreSQL)

Source: https://github.com/emitdotnet/emit/blob/main/samples/distributed-locks/README.md

Starts the inventory snapshot worker using PostgreSQL as the persistence backend. Ensure infrastructure is started first.

```bash
dotnet run --project InventorySnapshot.PostgreSQL
```

--------------------------------

### Run Building Sentinel with PostgreSQL

Source: https://github.com/emitdotnet/emit/blob/main/samples/building-sentinel/README.md

Starts the Building Sentinel application using PostgreSQL as the persistence backend. Ensure infrastructure is running first.

```bash
dotnet run --project BuildingSentinel.PostgreSQL
```

--------------------------------

### Configure Dead Letter Queue (DLQ)

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/advanced/configuration.mdx

Configure a dead letter queue for failed messages. This example shows basic DLQ setup and configuration with provisioning options like message retention.

```csharp
kafka.DeadLetter("orders.dlt");
```

```csharp
// With provisioning options:
kafka.DeadLetter("orders.dlt", dlq =>
{
    dlq.Provisioning(options =>
    {
        options.Retention = TimeSpan.FromDays(30);
    });
});
```

--------------------------------

### Run Inventory Snapshot Worker (MongoDB)

Source: https://github.com/emitdotnet/emit/blob/main/samples/distributed-locks/README.md

Starts the inventory snapshot worker using MongoDB as the persistence backend. Ensure infrastructure is started first.

```bash
dotnet run --project InventorySnapshot.MongoDB
```

--------------------------------

### Install Emit Health Check Packages

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/observability/health-checks.mdx

Install the Emit health check packages that correspond to your registered providers. Only install the packages you need.

```bash
dotnet add package Emit.Kafka.HealthChecks
dotnet add package Emit.MongoDB.HealthChecks
dotnet add package Emit.EntityFrameworkCore.HealthChecks
```

--------------------------------

### Add Emit Testing Package

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/advanced/testing.mdx

Install the Emit.Testing package using the .NET CLI.

```bash
dotnet add package Emit.Testing
```

--------------------------------

### Install MongoDB Persistence Package

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/persistence/mongodb.mdx

Add the Emit.Persistence.MongoDB package to your project using the .NET CLI.

```bash
dotnet add package Emit.Persistence.MongoDB
```

--------------------------------

### Add EF Core Persistence Package

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/persistence/efcore.mdx

Install the EF Core persistence package for Emit using the .NET CLI.

```bash
dotnet add package Emit.Persistence.EntityFrameworkCore
```

--------------------------------

### Run Building Sentinel with MongoDB

Source: https://github.com/emitdotnet/emit/blob/main/samples/building-sentinel/README.md

Starts the Building Sentinel application using MongoDB as the persistence backend. Ensure infrastructure is running first.

```bash
dotnet run --project BuildingSentinel.MongoDB
```

--------------------------------

### Add Emit OpenTelemetry Package

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/observability/tracing.mdx

Install the Emit OpenTelemetry package using the .NET CLI.

```bash
dotnet add package Emit.OpenTelemetry
```

--------------------------------

### Add Emit Packages

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/getting-started/installation.mdx

Install the core Emit package along with Kafka and MongoDB integrations using the .NET CLI.

```bash
dotnet add package Emit
dotnet add package Emit.Kafka
dotnet add package Emit.MongoDB
```

--------------------------------

### Implement a Kafka Consumer

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/getting-started/quickstart.mdx

Implement the `IConsumer` interface to process incoming messages. This example logs details of a `PizzaOrdered` message.

```csharp
public class PizzaOrderedConsumer(ILogger<PizzaOrderedConsumer> logger)
    : IConsumer<PizzaOrdered>
{
    public Task ConsumeAsync(
        ConsumeContext<PizzaOrdered> context,
        CancellationToken cancellationToken)
    {
        var pizza = context.Message;
        logger.LogInformation(
            "New order {PizzaId} for {CustomerId}: {Toppings}",
            pizza.PizzaId,
            pizza.CustomerId,
            string.Join(", ", pizza.Toppings));

        return Task.CompletedTask;
    }
}

```

--------------------------------

### Minimal Kafka Registration

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/kafka/setup.mdx

Configure Emit to use Kafka with minimal settings, including bootstrap servers and topic definitions with serializers and deserializers. This example shows how to set up a topic for both producers and a consumer group.

```csharp
builder.Services.AddEmit(emit =>
{
    emit.AddKafka(kafka =>
    {
        kafka.ConfigureClient(cfg =>
        {
            cfg.BootstrapServers = "localhost:9092";
        });

        kafka.Topic<string, PizzaOrdered>("pizzas", topic =>
        {
            topic.SetKeySerializer(Confluent.Kafka.Serializers.Utf8);
            topic.SetValueSerializer(/* your serializer */);
            topic.SetKeyDeserializer(Confluent.Kafka.Deserializers.Utf8);
            topic.SetValueDeserializer(/* your deserializer */);

            topic.Producer();

            topic.ConsumerGroup("pizza-kitchen", group =>
            {
                group.AddConsumer<PizzaOrderedConsumer>();
            });
        });
    });
});
```

--------------------------------

### Register Kafka Consumer

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/kafka/consumers.mdx

Configure Kafka topics and consumer groups within the Emit framework. This example shows how to register a consumer for the 'pizzas' topic.

```csharp
kafka.Topic<string, PizzaOrdered>("pizzas", topic =>
{
    topic.ConsumerGroup("pizza-kitchen", group =>
    {
        group.AddConsumer<PizzaOrderedConsumer>();
    });
});
```

--------------------------------

### Run Batch Consumer Application

Source: https://github.com/emitdotnet/emit/blob/main/samples/batch-consumer/README.md

Command to run the batch consumer application using .NET CLI. Ensure infrastructure is started first.

```bash
dotnet run --project BatchConsumer
```

--------------------------------

### Inline Activity Enrichment with Delegate

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/observability/tracing.mdx

For inline enrichment without a dedicated class, register a delegate directly with ConfigureTracing. This example adds an 'env' tag.

```csharp
emit.ConfigureTracing(tracing =>
{
    tracing.EnrichActivity((activity, context) =>
    {
        activity.SetTag("env", "production");
    });
});
```

--------------------------------

### Console Output: Lock Acquired

Source: https://github.com/emitdotnet/emit/blob/main/samples/distributed-locks/README.md

Example console output when a worker successfully acquires the distributed lock and proceeds with computing the inventory snapshot.

```text
[Simulator] Delivery    — BOLT-M8    +34 units
[Simulator] Shipment    — GLOVE-L    -12 units
[Simulator] Discrepancy — PUMP-2HP   -3 units
[Snapshot]  Lock acquired — computing inventory snapshot...
[Snapshot]  #1 — 10 products, 3,987 units across 4 warehouse(s) in 4.2ms
```

--------------------------------

### Register Emit Services with MongoDB and Kafka

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/getting-started/quickstart.mdx

Configure and register Emit services in your application's dependency injection container. This example sets up MongoDB for the outbox and distributed locking, and Kafka for message production and consumption with JSON schema serialization.

```csharp
using MongoDB.Driver;

builder.Services.AddSingleton<IMongoClient>(
    new MongoClient("mongodb://localhost:27017"));

builder.Services.AddEmit(emit =>
{
    emit.AddMongoDb(mongo =>
    {
        mongo.Configure((sp, ctx) =>
        {
            ctx.Client = sp.GetRequiredService<IMongoClient>();
            ctx.Database = ctx.Client.GetDatabase("pizzeria-db");
        })
        .UseOutbox()
        .UseDistributedLock();
    });

    emit.AddKafka(kafka =>
    {
        kafka.ConfigureClient(cfg =>
        {
            cfg.BootstrapServers = "localhost:9092";
        });

        kafka.ConfigureSchemaRegistry(cfg =>
        {
            cfg.Url = "localhost:8081";
        });

        kafka.AutoProvision();

        kafka.Topic<string, PizzaOrdered>("pizzas", topic =>
        {
            topic.SetKeySerializer(Confluent.Kafka.Serializers.Utf8);
            topic.SetJsonSchemaValueSerializer();
            topic.SetKeyDeserializer(Confluent.Kafka.Deserializers.Utf8);
            topic.SetJsonSchemaValueDeserializer();

            topic.Producer();

            topic.ConsumerGroup("pizza-kitchen", group =>
            {
                group.AddConsumer<PizzaOrderedConsumer>();
            });
        });
    });
});

```

--------------------------------

### Configure Consumer Error Policy

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/error-handling/error-policies.mdx

Define an error policy for a consumer group using `OnError`. This example shows how to handle `KitchenCapacityException` by retrying and then dead-lettering, `InvalidAddressException` by dead-lettering, and all other exceptions with a default dead-letter action.

```csharp
topic.ConsumerGroup("pizza-kitchen", group =>
{
    group.OnError(err =>
    {
        err.When<KitchenCapacityException>(action => action
            .Retry(3, Backoff.Exponential(TimeSpan.FromMilliseconds(200)))
            .DeadLetter());

        err.When<InvalidAddressException>(action => action.DeadLetter());

        err.Default(action => action.DeadLetter());
    });

    group.AddConsumer<PizzaOrderedConsumer>();
});
```

--------------------------------

### Route Key Selector Examples

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/kafka/routing.mdx

Demonstrates different ways to define a route key selector function. The selector can extract a field, use an enum value, or derive a key based on conditional logic.

```csharp
// String discriminator field
selector: ctx => ctx.Message.EventType
```

```csharp
// Enum value
selector: ctx => ctx.Message.OrderKind
```

```csharp
// Derived value
selector: ctx => ctx.Message.CorrelationId.StartsWith("B2B") ? "b2b" : "b2c"
```

--------------------------------

### Simulate Circuit Breaker with Docker

Source: https://github.com/emitdotnet/emit/blob/main/samples/batch-consumer/README.md

Commands to stop and start the MongoDB container to test the circuit breaker functionality. The consumer pauses when MongoDB is stopped and resumes when it's restarted.

```bash
docker stop samples-mongo-1     # circuit breaker opens, consumer pauses 15 s
```

```bash
docker start samples-mongo-1    # circuit breaker closes, processing resumes
```

--------------------------------

### Transactional Batch Consumer Example

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/kafka/batch-consuming.mdx

A batch consumer decorated with [Transactional] to ensure all outbox messages produced within the handler are committed atomically. Each retry attempt starts a new transaction.

```csharp
[Transactional]
public sealed class ScanBatchConsumer(
    IEventProducer<string, RerouteCommand> producer)
    : IBatchConsumer<PackageScan>
{
    public async Task ConsumeAsync(
        ConsumeContext<MessageBatch<PackageScan>> context,
        CancellationToken cancellationToken)
    {
        foreach (var item in context.Message.Items)
        {
            if (NeedsReroute(item.Message))
            {
                await producer.ProduceAsync(item.Message.Barcode, /* ... */, cancellationToken);
            }
        }
        // All produced messages commit atomically on success.
    }
}
```

--------------------------------

### Configure Kafka Producers with Outbox

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/outbox/index.mdx

Sets up Kafka producers, with one routing through the outbox by default and another explicitly bypassing it using `UseDirect()`.

```csharp
emit.AddKafka(kafka =>
{
    kafka.ConfigureClient(cfg => cfg.BootstrapServers = "localhost:9092");

    kafka.Topic<string, PizzaOrdered>("pizzas", topic =>
    {
        topic.Producer();  // routes through outbox by default
    });

    kafka.Topic<string, AnalyticsEvent>("analytics", topic =>
    {
        topic.Producer(p => p.UseDirect());  // bypasses the outbox
    });
});

```

--------------------------------

### Build and Test Commands for Emit .NET

Source: https://github.com/emitdotnet/emit/blob/main/CLAUDE.md

Standard .NET CLI commands for building, formatting, running unit tests, and integration tests for the Emit project. Integration tests require Docker.

```bash
dotnet build
```

```bash
dotnet format
```

```bash
dotnet test tests/Emit.UnitTests/Emit.UnitTests.csproj
```

```bash
dotnet test tests/Emit.IntegrationTests/Emit.IntegrationTests.csproj
```

--------------------------------

### Configure DLQ Provisioning Options

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/error-handling/dead-letter-queue.mdx

Configure provisioning options for the dead-letter topic, such as the number of partitions and retention period. This is done similarly to configuring any other topic.

```csharp
kafka.DeadLetter("orders.dlt", dlq =>
{
    dlq.Provisioning(options =>
    {
        options.NumPartitions = 3;
        options.Retention = TimeSpan.FromDays(30);
    });
});
```

--------------------------------

### Add FluentValidation Integration

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/error-handling/validation.mdx

Install the `Emit.FluentValidation` package to integrate FluentValidation with Emit. This allows you to use your existing `AbstractValidator<T>` classes.

```bash
dotnet add package Emit.FluentValidation
```

--------------------------------

### Register Mediator and Handlers

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/mediator/index.mdx

Register the mediator and its handlers using `AddEmit` and `AddMediator`. This setup is independent of other Emit configurations like Kafka.

```csharp
builder.Services.AddEmit(emit =>
{
    emit.AddMediator(mediator =>
    {
        mediator.AddHandler<PlacePizzaOrderHandler>();
        mediator.AddHandler<GetPizzaStatusHandler>();
    });
});
```

--------------------------------

### Auto-provisioning Topics with Custom Settings

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/kafka/setup.mdx

Enable auto-provisioning of Kafka topics and configure specific settings like the number of partitions, replication factor, retention period, and compression type for newly created topics.

```csharp
emit.AddKafka(kafka =>
{
    kafka.AutoProvision();

    kafka.Topic<string, PizzaOrdered>("pizzas", topic =>
    {
        topic.Provisioning(options =>
        {
            options.NumPartitions = 6;
            options.ReplicationFactor = 3;
            options.Retention = TimeSpan.FromDays(14);
            options.CompressionType = TopicCompressionType.Zstd;
        });

        // ... serializers, producer, consumer groups ...
    });
});
```

--------------------------------

### Configuring Kafka Serialization

Source: https://github.com/emitdotnet/emit/blob/main/tests/Emit.IntegrationTests/INSTRUCT.md

Illustrates how to configure UTF-8 serialization for Kafka string-to-string topics using a concise method.

```csharp
kafka.Topic<string, string>(topic, t =>
{
    t.UseUtf8Serialization();
    t.Producer();
    t.ConsumerGroup(groupId, group => { ... });
});
```

--------------------------------

### Configure Protobuf Serializer

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/kafka/serialization.mdx

Implement Protobuf serialization for Kafka topics using Emit. Requires the Emit.Kafka.ProtobufSerializer package. TValue must implement IMessage<TValue>. RuleRegistry can be passed for rule evaluation.

```bash
dotnet add package Emit.Kafka.ProtobufSerializer
```

```csharp
kafka.Topic<string, MyProtoMessage>("events", topic =>
{
    topic.SetProtobufValueSerializer();
    topic.SetProtobufValueDeserializer();
});
```

--------------------------------

### Implement IActivityEnricher for Custom Tags

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/observability/tracing.mdx

Implement IActivityEnricher to add application-specific tags to every Emit span. This example adds a 'tenant.id' tag.

```csharp
public class TenantActivityEnricher : IActivityEnricher
{
    private readonly ITenantContext _tenant;

    public TenantActivityEnricher(ITenantContext tenant) => _tenant = tenant;

    public void Enrich(Activity activity, EnrichmentContext context)
    {
        activity.SetTag("tenant.id", _tenant.TenantId);
    }
}
```

--------------------------------

### Exception Matching with Inheritance

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/error-handling/error-policies.mdx

Demonstrates how to match specific exception types and their subclasses using `When<TException>`. More specific exception types should be placed before broader ones to ensure correct evaluation order.

```csharp
group.OnError(err =>
{
    // Matches HttpRequestException and any derived types
    err.When<HttpRequestException>(action => action
        .Retry(3, Backoff.Exponential(TimeSpan.FromSeconds(1)))
        .DeadLetter());

    // Matches InvalidOperationException and subclasses
    err.When<InvalidOperationException>(action => action.Discard());

    err.Default(action => action.DeadLetter());
});
```

--------------------------------

### Add Emit Metrics Instrumentation

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/observability/metrics.mdx

Configure OpenTelemetry metrics to include Emit instrumentation and an OTLP exporter. This is the basic setup for enabling Emit's metrics.

```csharp
builder.Services.AddOpenTelemetry()
    .WithMetrics(metrics =>
    {
        metrics.AddEmitInstrumentation();
        metrics.AddOtlpExporter();
    });
```

--------------------------------

### Using Shared Test Utilities

Source: https://github.com/emitdotnet/emit/blob/main/tests/Emit.IntegrationTests/INSTRUCT.md

Shows how to import and use shared test utilities for capturing messages and polling for conditions.

```csharp
using Emit.IntegrationTests.Integration;  // for DlqCaptureConsumer, InvocationCounter, etc.
```

```csharp
using static Emit.IntegrationTests.Integration.TestHelpers;

await WaitUntilAsync(() => sink.ReceivedMessages.Count >= expected, "Messages not received");
```

--------------------------------

### Accessing Batch Items and Count

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/kafka/batch-consuming.mdx

Iterate through the batch items using `batch.Items`, access individual items by index, or get the total count of messages in the batch.

```csharp
foreach (var item in batch.Items) { /* ... */ }
var first = batch[0];
var count = batch.Count;
```

--------------------------------

### Console Output: Lock Contention

Source: https://github.com/emitdotnet/emit/blob/main/samples/distributed-locks/README.md

Example console output when a worker detects lock contention and skips the inventory snapshot cycle because another instance is already running it.

```text
[Snapshot] Lock contention — another instance is already running the snapshot. Skipping this cycle.
```

--------------------------------

### Register a Batch Consumer for a Kafka Topic

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/kafka/batch-consuming.mdx

Register your custom batch consumer, ScanBatchConsumer, for a specific Kafka topic and consumer group. This setup is done during Kafka configuration.

```csharp
kafka.Topic<string, PackageScan>("sorting.scans", topic =>
{
    topic.ConsumerGroup("sorting.sorter", group =>
    {
        group.AddBatchConsumer<ScanBatchConsumer>();
    });
});
```

--------------------------------

### Configure Tracer Provider with Emit

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/observability/tracing.mdx

Wire up the tracer provider in your application's service configuration to include Emit instrumentation and an OTLP exporter.

```csharp
builder.Services.AddOpenTelemetry()
    .WithTracing(tracing =>
    {
        tracing.AddEmitInstrumentation();
        tracing.AddOtlpExporter();  // or your preferred exporter
    });
```

--------------------------------

### Transactional Outbox with [Transactional] Attribute

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/outbox/index.mdx

Use the `[Transactional]` attribute for consumers and handlers. The pipeline automatically manages the transaction, committing or rolling back. Each retry gets a new transaction.

```csharp
using Emit.Transactional;

[Transactional]
public sealed class MyHandler : IRequestHandler<MyCommand>
{
    public async Task HandleAsync(MyCommand request, CancellationToken ct)
    {
        await eventRepository.InsertAsync(request.Event, ct);
        await producer.ProduceAsync(..., ct);
        // Transaction committed automatically by middleware.
    }
}
```

--------------------------------

### Access Retry Attempt Number

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/error-handling/error-policies.mdx

Check the current retry attempt number within your consumer to implement logic that behaves differently on retries. The attempt number starts at 0 for the initial attempt.

```csharp
public async Task ConsumeAsync(ConsumeContext<OrderPlaced> context)
{
    if (context.RetryAttempt > 0)
        logger.LogInformation("Retry attempt {Attempt}", context.RetryAttempt);

    // 0 = initial attempt, 1 = first retry, 2 = second retry, ...
}
```

--------------------------------

### Default Batch Consumer Configuration

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/kafka/batch-consuming.mdx

Use default settings for MaxSize (100 messages) and Timeout (5 seconds) when no configuration delegate is provided.

```csharp
group.AddBatchConsumer<ScanBatchConsumer>(); // MaxSize = 100, Timeout = 5s
```

--------------------------------

### Implement IBatchConsumer for PackageScans

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/kafka/batch-consuming.mdx

Implement the IBatchConsumer interface to process a batch of PackageScan messages. Access deserialized payloads via `item.Message` and raw transport details via `item.TransportContext`.

```csharp
public class ScanBatchConsumer(ILogger<ScanBatchConsumer> logger)
    : IBatchConsumer<PackageScan>
{
    public async Task ConsumeAsync(
        ConsumeContext<MessageBatch<PackageScan>> context,
        CancellationToken cancellationToken)
    {
        var batch = context.Message;
        logger.LogInformation("Processing {Count} scans", batch.Count);

        foreach (var item in batch.Items)
        {
            var scan = item.Message;       // deserialized payload
            var tc = item.TransportContext; // raw bytes, headers, partition, offset
            // process...
        }
    }
}
```

--------------------------------

### Implement Custom Middleware

Source: https://github.com/emitdotnet/emit/blob/main/docs/src/content/docs/advanced/custom-middleware.mdx

Implement the IMiddleware interface for inbound messages. For outbound middleware, use SendContext<T> instead of ConsumeContext<T>.

```csharp
public class TimingMiddleware<T> : IMiddleware<ConsumeContext<T>>
{
    private readonly ILogger<TimingMiddleware<T>> _logger;

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

    public async Task InvokeAsync(ConsumeContext<T> context, IMiddlewarePipeline<ConsumeContext<T>> next)
    {
        var sw = Stopwatch.StartNew();
        try
        {
            await next.InvokeAsync(context);
        }
        finally
        {
            _logger.LogDebug("Message handled in {ElapsedMs}ms", sw.ElapsedMilliseconds);
        }
    }
}
```

--------------------------------

### Simulate Lock Contention

Source: https://github.com/emitdotnet/emit/blob/main/samples/distributed-locks/README.md

Starts a second instance of the inventory snapshot worker on a different port to simulate lock contention. This instance will detect that another instance already holds the lock and skip its snapshot cycle.

```bash
Urls=http://localhost:5002 dotnet run --project InventorySnapshot.MongoDB
```