Event-Driven приложения с Apache Kafka и KafkaFlow в .NET

.AddConsumer(consumer => consumer
    .Topic("orders-topic")
    .WithGroupId("order-processing")
    .WithBufferSize(100)
    .WithWorkersCount(10) // Настройка параллельной обработки
    .AddMiddlewares(/* ... */)
)

.AddMiddlewares(middlewares => middlewares
    .AddSerializer<JsonMessageSerializer>() // Или Protobuf, или Avro...
)

.AddMiddlewares(middlewares => middlewares
    .BatchConsume(100, TimeSpan.FromSeconds(5))
    .Add<BatchProcessingMiddleware>()
)

.AddMiddlewares(middlewares => middlewares
    .AddDeserializer<JsonMessageDeserializer>()
    .AddTypedHandlers(handlers => handlers
        .WithHandlerLifetime(InstanceLifetime.Transient)
        .AddHandler<OrderCreatedHandler>()
    )
    .AddCustomErrorHandler((context, exception) => {
        // Логика перенаправления в DLQ
        return Task.CompletedTask;
    })
)

public class IdempotentOrderHandler : IMessageHandler<OrderCreated>
{
    private readonly IOrderRepository _repository;
    
    public async Task Handle(IMessageContext context, OrderCreated message)
    {
        // Проверка, обрабатывали ли мы уже это сообщение
        if (await _repository.HasProcessedMessageAsync(message.EventId))
            return;
            
        // Обработка сообщения
        await _repository.ProcessOrderAsync(message);
        
        // Фиксация обработки
        await _repository.MarkMessageAsProcessedAsync(message.EventId);
    }
}

.AddProducer("compressed-events", producer => producer
    .DefaultTopic("some-high-volume-topic")
    .WithCompression(CompressionType.Gzip)
    .AddMiddlewares(/* ... */)
)

public class AvroSchemaRegistrySerializer : ISerializer
{
    private readonly ISchemaRegistryClient _schemaRegistry;
    
    public async Task SerializeAsync(object message, Stream output, ISerializerContext context)
    {
        // Логика сериализации с использованием Schema Registry
    }
    
    public async Task<object> DeserializeAsync(Stream input, Type type, ISerializerContext context)
    {
        // Логика десериализации с использованием Schema Registry
    }
}

var cluster = provider.GetRequiredService<IClusterManager>();
var consumer = cluster.GetConsumer("consumer-name");
 
// Получение информации о смещениях
var offsets = await consumer.GetOffsetsAsync();
 
// Приостановка/возобновление работы консьюмера
await consumer.PauseAsync();
await consumer.ResumeAsync();
 
// Перемотка к определенной позиции
await consumer.RewindAsync(offsets);

.AddConsumer(consumer => consumer
    .Topic("bulk-operations")
    .WithName("bulkConsumer")
    .AddMiddlewares(middlewares => middlewares
        .ThrottleConsumer(t => t
            .ByOtherConsumersLag("priorityConsumer")
            .WithInterval(TimeSpan.FromSeconds(5))
            .AddAction(a => a.AboveThreshold(100).ApplyDelay(1000))
            .AddAction(a => a.AboveThreshold(1000).ApplyDelay(10000))
        )
    )
)

var config = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "sample-consumer",
    AutoOffsetReset = AutoOffsetReset.Earliest
};
 
using var consumer = new ConsumerBuilder<string, string>(config).Build();
consumer.Subscribe("test-topic");
 
while (true)
{
    var consumeResult = consumer.Consume(CancellationToken.None);
    
    // Здесь нам самим нужно позаботиться о десериализации,
    // обработке ошибок, многопоточности и т.д.
    
    Console.WriteLine($"Received: {consumeResult.Message.Value}");
}

services.AddKafka(kafka => kafka
    .AddCluster(cluster => cluster
        .WithBrokers(new[] { "localhost:9092" })
        .AddConsumer(consumer => consumer
            .Topic("test-topic")
            .WithGroupId("sample-consumer")
            .WithAutoOffsetReset(AutoOffsetReset.Earliest)
            .WithBufferSize(100)
            .WithWorkersCount(10)
            .AddMiddlewares(m => m
                .AddDeserializer<JsonMessageDeserializer>()
                .AddTypedHandlers(h => h
                    .AddHandler<MyMessageHandler>()
                )
            )
        )
    )
);

var config = new ConsumerConfig
{
    // ...
    StatisticsIntervalMs = 5000
};
 
consumer.Statistics += (_, stats) => 
{
    // Самостоятельно парсим JSON и отправляем метрики куда-то
    Console.WriteLine(stats);
};

try
{
    var consumeResult = consumer.Consume(cancellationToken);
    ProcessMessage(consumeResult.Message.Value);
}
catch (ConsumeException ex)
{
    // Ошибка на стороне Kafka
    logger.LogError(ex, "Ошибка потребления");
}
catch (Exception ex)
{
    // Ошибка в нашем коде
    logger.LogError(ex, "Ошибка обработки");
    // А что делать дальше? Retry? Пропустить? DLQ?
}

.AddMiddlewares(middlewares => middlewares
    .AddDeserializer<JsonMessageDeserializer>()
    .AddTypedHandlers(h => h.AddHandler<MyHandler>())
    .AddCustomErrorHandler((context, exception) =>
    {
        logger.LogError(exception, "Ошибка обработки {MessageId}", context.Message.Id);
        
        // Отправляем в DLQ
        return _deadLetterProducer.ProduceAsync(
            "dead-letter-queue",
            context.Message.Key,
            new DeadLetterMessage(context.Message, exception)
        );
    })
)

var config = new ProducerConfig
{
    BootstrapServers = "localhost:9092",
    TransactionalId = "unique-transaction-id"
};
 
using var producer = new ProducerBuilder<string, string>(config).Build();
producer.InitTransactions(TimeSpan.FromSeconds(10));
 
try
{
    producer.BeginTransaction();
    // Отправка сообщений
    producer.Produce("topic", new Message<string, string> { ... });
    producer.Produce("another-topic", new Message<string, string> { ... });
    producer.CommitTransaction();
}
catch
{
    producer.AbortTransaction();
    throw;
}

.WithWorkersCount(Environment.ProcessorCount * 2)

.WithBufferSize(1000)

// Тестирование обработчика
[Fact]
public async Task Handler_ShouldProcessMessage_Correctly()
{
    // Arrange
    var handler = new MyMessageHandler(mockRepository.Object);
    var message = new MyMessage { Id = Guid.NewGuid(), Content = "test" };
    var context = new MessageContextMock<MyMessage>(message);
    
    // Act
    await handler.Handle(context, message);
    
    // Assert
    mockRepository.Verify(r => r.SaveAsync(It.Is<MyEntity>(e => e.Id == message.Id)), Times.Once);
}

public class OrderCreatedHandler : IMessageHandler<OrderCreated>
{
    public Task Handle(IMessageContext context, OrderCreated message)
    {
        // Тип сообщения уже известен, можно сразу работать с бизнес-моделью
        return Task.CompletedTask;
    }
}

public class LoggingMiddleware : IMessageMiddleware
{
    private readonly ILogger<LoggingMiddleware> _logger;
 
    public LoggingMiddleware(ILogger<LoggingMiddleware> logger)
    {
        _logger = logger;
    }
 
    public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
    {
        _logger.LogInformation("Обработка сообщения: {MessageId}", context.Message.Key);
        
        var stopwatch = Stopwatch.StartNew();
        await next(context);
        stopwatch.Stop();
        
        _logger.LogInformation("Сообщение {MessageId} обработано за {ElapsedMs}мс", 
            context.Message.Key, stopwatch.ElapsedMilliseconds);
    }
}

.AddConsumer(consumer => consumer
    .Topic("orders-topic")
    .WithGroupId("order-processing")
    .AddMiddlewares(middlewares => middlewares
        .Add<LoggingMiddleware>()
        .AddDeserializer<JsonMessageDeserializer>()
        .AddTypedHandlers(handlers => handlers
            .AddHandler<OrderCreatedHandler>()
        )
    )
)

public class OrderSaga
{
    private readonly IProducerAccessor _producers;
    
    public OrderSaga(IProducerAccessor producers)
    {
        _producers = producers;
    }
    
    // Шаг 1: Создание заказа
    public class CreateOrderHandler : IMessageHandler<CreateOrderCommand>
    {
        private readonly OrderSaga _saga;
        private readonly IOrderRepository _repository;
        
        public async Task Handle(IMessageContext context, CreateOrderCommand command)
        {
            try {
                var order = await _repository.CreateOrderAsync(command);
                
                // Переход к следующему шагу
                await _saga._producers["order-saga"]
                    .ProduceAsync("reserveInventory", order.Id.ToString(), 
                        new ReserveInventoryCommand { OrderId = order.Id, Items = order.Items });
            }
            catch (Exception) {
                // Публикация события об ошибке
                await _saga._producers["order-saga"]
                    .ProduceAsync("orderFailed", command.OrderId.ToString(), 
                        new OrderFailedEvent { OrderId = command.OrderId, Reason = "Failed to create order" });
            }
        }
    }
    
    // Другие обработчики для шагов саги...
}

public class AccountAggregate
{
    public Guid Id { get; private set; }
    public decimal Balance { get; private set; }
    
    // Применение события
    public void Apply(AccountCredited event)
    {
        Balance += event.Amount;
    }
    
    public void Apply(AccountDebited event)
    {
        Balance -= event.Amount;
    }
}
 
public class AccountEventHandler : 
    IMessageHandler<AccountCredited>,
    IMessageHandler<AccountDebited>
{
    private readonly IEventStore _eventStore;
    
    public async Task Handle(IMessageContext context, AccountCredited event)
    {
        // Сохранение события
        await _eventStore.SaveEventAsync(event);
        
        // Обновление проекции (read model)
        var account = await _eventStore.GetAggregateAsync<AccountAggregate>(event.AccountId);
        account.Apply(event);
        await _eventStore.UpdateProjectionAsync(account);
    }
    
    public async Task Handle(IMessageContext context, AccountDebited event)
    {
        // Аналогичная логика для дебета
    }
}

public class RxMiddleware<T> : IMessageMiddleware
{
    private readonly Subject<T> _subject = new Subject<T>();
    
    public IObservable<T> MessageStream => _subject.AsObservable();
    
    public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
    {
        if (context.Message.Value is T value)
        {
            _subject.OnNext(value);
        }
        
        await next(context);
    }
}

var rxMiddleware = new RxMiddleware<OrderCreated>();
 
// Регистрация middleware
.AddMiddlewares(middlewares => middlewares
    .Add(rxMiddleware)
    .AddDeserializer<JsonMessageDeserializer>()
    .AddTypedHandlers(/* ... */)
)
 
// Использование Rx-потока
rxMiddleware.MessageStream
    .Buffer(TimeSpan.FromSeconds(5), 100) // Группировка сообщений по времени или количеству
    .Where(batch => batch.Count > 0)
    .Subscribe(async batch => {
        // Пакетная обработка заказов
        await ProcessOrderBatchAsync(batch);
    });

rxMiddleware.MessageStream
    .GroupBy(metric => metric.ServerId)
    .SelectMany(group => group
        .Window(TimeSpan.FromMinutes(5))
        .SelectMany(window => window
            .Aggregate(
                new ServerStatistics(),
                (stats, metric) => stats.AddMetric(metric)
            )
        )
    )
    .Where(stats => stats.HasAnomaly())
    .Subscribe(stats => AlertOperations(stats));

public class CreateProductCommandHandler : IMessageHandler<CreateProductCommand>
{
    private readonly IRepository<Product> _repository;
    private readonly IProducerAccessor _producers;
    
    public async Task Handle(IMessageContext context, CreateProductCommand command)
    {
        // Обработка команды
        var product = new Product 
        { 
            Id = Guid.NewGuid(), 
            Name = command.Name,
            Price = command.Price 
        };
        
        await _repository.SaveAsync(product);
        
        // Публикация события
        await _producers["product-events"]
            .ProduceAsync("products", product.Id.ToString(), 
                new ProductCreatedEvent 
                { 
                    Id = product.Id,
                    Name = product.Name,
                    Price = product.Price,
                    CreatedAt = DateTime.UtcNow 
                });
    }
}

public class CircuitBreakerMiddleware : IMessageMiddleware
{
    private readonly CircuitBreakerPolicy _circuitBreaker;
    
    public CircuitBreakerMiddleware()
    {
        _circuitBreaker = Policy
            .Handle<HttpRequestException>()
            .CircuitBreakerAsync(
                exceptionsAllowedBeforeBreaking: 5,
                durationOfBreak: TimeSpan.FromMinutes(1)
            );
    }
    
    public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
    {
        try
        {
            await _circuitBreaker.ExecuteAsync(async () => await next(context));
        }
        catch (BrokenCircuitException)
        {
            // Логика при открытом предохранителе
            // Например, отправка в DLQ или альтернативная обработка
        }
    }
}

public class OrderService
{
    private readonly IRepository<Order> _orderRepository;
    private readonly IOutboxRepository _outboxRepository;
    
    public async Task CreateOrderAsync(CreateOrderRequest request)
    {
        using var transaction = await _orderRepository.BeginTransactionAsync();
        
        // Основная логика
        var order = new Order { /* ... */ };
        await _orderRepository.SaveAsync(order);
        
        // Сохранение события в outbox
        var @event = new OrderCreatedEvent { /* ... */ };
        await _outboxRepository.SaveEventAsync(@event, transaction);
        
        await transaction.CommitAsync();
    }
}
 
// Отдельный процесс для публикации
public class OutboxProcessor
{
    private readonly IOutboxRepository _outboxRepository;
    private readonly IProducerAccessor _producers;
    
    public async Task ProcessAsync(CancellationToken cancellationToken)
    {
        while (!cancellationToken.IsCancellationRequested)
        {
            var pendingEvents = await _outboxRepository.GetPendingEventsAsync(100);
            
            foreach (var eventRecord in pendingEvents)
            {
                await _producers["outbox"]
                    .ProduceAsync(
                        eventRecord.Topic,
                        eventRecord.Key,
                        eventRecord.Event
                    );
                
                await _outboxRepository.MarkAsProcessedAsync(eventRecord.Id);
            }
            
            await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken);
        }
    }
}

public class ServiceRoutingMiddleware : IMessageMiddleware
{
    private readonly IServiceRegistry _serviceRegistry;
    private readonly IProducerAccessor _producers;
    
    public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
    {
        if (context.Message.Value is IRoutableMessage routable)
        {
            var destinationService = await _serviceRegistry.ResolveServiceAsync(routable.Destination);
            
            if (destinationService != null)
            {
                // Динамическое определение топика по назначению
                await _producers["service-router"]
                    .ProduceAsync(
                        destinationService.TopicName,
                        context.Message.Key,
                        context.Message.Value
                    );
                
                // Прерываем цепочку middleware, так как сообщение перенаправлено
                return;
            }
        }
        
        // Если это не маршрутизируемое сообщение или служба не найдена,
        // продолжаем стандартную обработку
        await next(context);
    }
}

public class Startup
{
    public void ConfigureServices(IServiceCollection services)
    {
        services.AddControllers();
        
        services.AddKafka(kafka => kafka
            .AddCluster(cluster => cluster
                .WithBrokers(new[] { "localhost:9092" })
                .AddConsumer(/* ... */)
                .AddProducer(/* ... */)
            )
        );
        
        // Интеграция с HealthChecks
        services.AddHealthChecks()
            .AddKafkaFlowHealthCheck();
    }
    
    public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
    {
        app.UseRouting();
        
        app.UseEndpoints(endpoints =>
        {
            endpoints.MapControllers();
            endpoints.MapHealthChecks("/health");
            
            // Добавление Kafka Flow Dashboard
            endpoints.MapKafkaFlowDashboard("/kafka-dashboard");
        });
        
        // Запуск Kafka Flow
        var bus = app.ApplicationServices.CreateKafkaBus();
        bus.StartAsync().GetAwaiter().GetResult();
    }
}

public class OrderCreated
{
    public Guid OrderId { get; set; }
    public string CustomerEmail { get; set; }
    public List<OrderItem> Items { get; set; }
    public decimal TotalAmount { get; set; }
    public DateTime CreatedAt { get; set; }
}
 
public class OrderItem
{
    public Guid ProductId { get; set; }
    public string ProductName { get; set; }
    public int Quantity { get; set; }
    public decimal UnitPrice { get; set; }
}

public class OrderProducer
{
    private readonly IProducerAccessor _producers;
 
    public OrderProducer(IProducerAccessor producers)
    {
        _producers = producers;
    }
 
    public async Task PublishOrderCreatedAsync(OrderCreated order)
    {
        await _producers["order-events"]
            .ProduceAsync(
                "orders", 
                order.OrderId.ToString(), 
                order);
    }
}

public class OrderCreatedHandler : IMessageHandler<OrderCreated>
{
    private readonly IEmailService _emailService;
    private readonly IOrderRepository _orderRepository;
    private readonly ILogger<OrderCreatedHandler> _logger;
 
    public OrderCreatedHandler(
        IEmailService emailService,
        IOrderRepository orderRepository,
        ILogger<OrderCreatedHandler> logger)
    {
        _emailService = emailService;
        _orderRepository = orderRepository;
        _logger = logger;
    }
 
    public async Task Handle(IMessageContext context, OrderCreated order)
    {
        try
        {
            _logger.LogInformation("Обработка заказа {OrderId}", order.OrderId);
            
            // Сохранение заказа в базу данных
            await _orderRepository.SaveOrderAsync(order);
            
            // Отправка подтверждения по электронной почте
            await _emailService.SendOrderConfirmationAsync(
                order.CustomerEmail,
                order.OrderId,
                order.Items,
                order.TotalAmount);
                
            _logger.LogInformation("Заказ {OrderId} успешно обработан", order.OrderId);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Ошибка при обработке заказа {OrderId}", order.OrderId);
            throw; // Пробрасываем исключение для обработки middleware
        }
    }
}

services.AddKafka(kafka => kafka
    .AddCluster(cluster => cluster
        .WithBrokers(new[] { "kafka-broker:9092" })
        .AddProducer("order-events", producer => producer
            .DefaultTopic("orders")
            .AddMiddlewares(middlewares => middlewares
                .AddSerializer<JsonMessageSerializer>()
            )
        )
        .AddConsumer(consumer => consumer
            .Topic("orders")
            .WithGroupId("order-processing")
            .WithBufferSize(100)
            .WithWorkersCount(5)
            .WithAutoOffsetReset(AutoOffsetReset.Earliest)
            .AddMiddlewares(middlewares => middlewares
                .AddDeserializer<JsonMessageDeserializer>()
                .AddTypedHandlers(handlers => handlers
                    .WithHandlerLifetime(InstanceLifetime.Transient)
                    .AddHandler<OrderCreatedHandler>()
                )
                .AddCustomErrorHandler((context, exception) => {
                    // Логика обработки ошибок
                    var order = context.Message.Value as OrderCreated;
                    if (order != null)
                    {
                        var logger = context.ServiceProvider.GetRequiredService<ILogger<Program>>();
                        logger.LogError(exception, "Критическая ошибка при обработке заказа {OrderId}", order.OrderId);
                        
                        // Отправка в Dead Letter Queue
                        var producers = context.ServiceProvider.GetRequiredService<IProducerAccessor>();
                        return producers["error-handler"]
                            .ProduceAsync(
                                "orders-dlq", 
                                order.OrderId.ToString(), 
                                new DeadLetterMessage(order, exception));
                    }
                    return Task.CompletedTask;
                })
            )
        )
    )
);

[Fact]
public async Task OrderCreatedHandler_ShouldSendConfirmationEmail()
{
    // Arrange
    var emailService = Substitute.For<IEmailService>();
    var orderRepository = Substitute.For<IOrderRepository>();
    var logger = Substitute.For<ILogger<OrderCreatedHandler>>();
    
    var handler = new OrderCreatedHandler(emailService, orderRepository, logger);
    
    var order = new OrderCreated
    {
        OrderId = Guid.NewGuid(),
        CustomerEmail = "customer@example.com",
        Items = new List<OrderItem> { /* ... */ },
        TotalAmount = 99.99m,
        CreatedAt = DateTime.UtcNow
    };
    
    var context = Substitute.For<IMessageContext>();
    
    // Act
    await handler.Handle(context, order);
    
    // Assert
    await orderRepository.Received(1).SaveOrderAsync(order);
    await emailService.Received(1).SendOrderConfirmationAsync(
        order.CustomerEmail,
        order.OrderId,
        order.Items,
        order.TotalAmount);
}

public class KafkaIntegrationTests : IClassFixture<KafkaFixture>
{
    private readonly KafkaFixture _fixture;
    
    public KafkaIntegrationTests(KafkaFixture fixture)
    {
        _fixture = fixture;
    }
    
    [Fact]
    public async Task ShouldProcessOrderCorrectly()
    {
        // Arrange
        var orderCreated = new OrderCreated { /* ... */ };
        
        // Act
        await _fixture.OrderProducer.PublishOrderCreatedAsync(orderCreated);
        
        // Ждем обработки сообщения
        await Task.Delay(TimeSpan.FromSeconds(2));
        
        // Assert
        var savedOrder = await _fixture.OrderRepository.GetOrderAsync(orderCreated.OrderId);
        Assert.NotNull(savedOrder);
        Assert.Equal(orderCreated.TotalAmount, savedOrder.TotalAmount);
    }
}
 
public class KafkaFixture : IDisposable
{
    private readonly IHost _host;
    
    public OrderProducer OrderProducer { get; }
    public IOrderRepository OrderRepository { get; }
    
    public KafkaFixture()
    {
        _host = new HostBuilder()
            .ConfigureServices(services => {
                // Настройка тестового контейнера
                services.AddSingleton<IOrderRepository, InMemoryOrderRepository>();
                services.AddSingleton<IEmailService, NoOpEmailService>();
                
                // Настройка KafkaFlow
                services.AddKafka(/* ... */);
            })
            .Build();
            
        _host.Start();
        
        OrderProducer = _host.Services.GetRequiredService<OrderProducer>();
        OrderRepository = _host.Services.GetRequiredService<IOrderRepository>();
    }
    
    public void Dispose()
    {
        _host.Dispose();
    }
}

public class CorrelationMiddleware : IMessageMiddleware
{
    private readonly ILogger<CorrelationMiddleware> _logger;
 
    public CorrelationMiddleware(ILogger<CorrelationMiddleware> logger)
    {
        _logger = logger;
    }
 
    public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
    {
        string correlationId = null;
 
        // Пытаемся извлечь существующий correlationId из заголовков
        var headers = context.Headers;
        if (headers.TryGetValue("X-Correlation-ID", out var existingId))
        {
            correlationId = Encoding.UTF8.GetString(existingId);
        }
        else
        {
            // Если отсутствует, создаем новый
            correlationId = Guid.NewGuid().ToString();
            
            // И добавляем в заголовки для последующих сообщений
            context.Headers.Add("X-Correlation-ID", Encoding.UTF8.GetBytes(correlationId));
        }
 
        // Добавляем в логи
        using (LogContext.PushProperty("CorrelationId", correlationId))
        {
            _logger.LogInformation("Обработка сообщения с CorrelationId {CorrelationId}", correlationId);
            await next(context);
        }
    }
}

public class OpenTelemetryMiddleware : IMessageMiddleware
{
    private readonly Tracer _tracer;
 
    public OpenTelemetryMiddleware(TracerProvider tracerProvider)
    {
        _tracer = tracerProvider.GetTracer("KafkaFlow");
    }
 
    public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
    {
        // Извлекаем контекст трассировки из заголовков, если есть
        var propagationContext = Propagators.DefaultTextMapPropagator.Extract(
            default, context.Headers, ExtractTraceContextFromHeaders);
 
        using var span = _tracer.StartSpan(
            $"Process {context.Topic}",
            SpanKind.Consumer,
            new SpanContext(propagationContext.ActivityContext));
            
        span.SetAttribute("messaging.system", "kafka");
        span.SetAttribute("messaging.destination", context.Topic);
        span.SetAttribute("messaging.kafka.key", context.Message.Key);
 
        try
        {
            await next(context);
            span.SetStatus(Status.Ok);
        }
        catch (Exception ex)
        {
            span.SetStatus(Status.Error.WithDescription(ex.Message));
            span.RecordException(ex);
            throw;
        }
    }
 
    private IEnumerable<string> ExtractTraceContextFromHeaders(Headers headers, string key)
    {
        if (headers.TryGetValue(key, out var value))
        {
            return new[] { Encoding.UTF8.GetString(value) };
        }
        return Enumerable.Empty<string>();
    }
}

public class EncryptionMiddleware : IMessageMiddleware
{
    private readonly IEncryptionService _encryptionService;
 
    public EncryptionMiddleware(IEncryptionService encryptionService)
    {
        _encryptionService = encryptionService;
    }
 
    public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
    {
        if (context.Message.Value is ISensitiveData)
        {
            // Шифруем перед отправкой
            var encryptedData = await _encryptionService.EncryptAsync(
                context.Message.Value as ISensitiveData);
                
            // Заменяем данные в контексте сообщения
            context.TransformMessage(message => new Message(
                message.Key,
                encryptedData,
                message.Headers));
        }
 
        await next(context);
    }
}

public class Program
{
    public static async Task Main(string[] args)
    {
        await CreateHostBuilder(args).Build().RunAsync();
    }
 
    public static IHostBuilder CreateHostBuilder(string[] args) =>
        Host.CreateDefaultBuilder(args)
            .ConfigureServices((hostContext, services) =>
            {
                // Регистрация бизнес-сервисов
                services.AddScoped<IOrderRepository, OrderRepository>();
                services.AddScoped<IEmailService, EmailService>();
                
                // Регистрация KafkaFlow
                services.AddKafka(kafka => kafka
                    .AddCluster(cluster => cluster
                        .WithBrokers(hostContext.Configuration.GetSection("Kafka:Brokers").Get<string[]>())
                        
                        // Продюсер для заказов
                        .AddProducer("order-events", producer => producer
                            .DefaultTopic("orders")
                            .AddMiddlewares(middlewares => middlewares
                                .Add<CorrelationMiddleware>()
                                .Add<OpenTelemetryMiddleware>()
                                .AddSerializer<JsonMessageSerializer>()
                            )
                        )
                        
                        // Консьюмер для заказов
                        .AddConsumer(consumer => consumer
                            .Topic("orders")
                            .WithGroupId("order-processing")
                            .WithBufferSize(100)
                            .WithWorkersCount(5)
                            .WithAutoOffsetReset(AutoOffsetReset.Earliest)
                            .AddMiddlewares(middlewares => middlewares
                                .Add<CorrelationMiddleware>()
                                .Add<OpenTelemetryMiddleware>()
                                .AddDeserializer<JsonMessageDeserializer>()
                                .AddTypedHandlers(handlers => handlers
                                    .WithHandlerLifetime(InstanceLifetime.Transient)
                                    .AddHandler<OrderCreatedHandler>()
                                )
                                .AddCustomErrorHandler((context, exception) => {
                                    // Логика обработки ошибок
                                    return Task.CompletedTask;
                                })
                            )
                        )
                    )
                );
                
                // Инициализация OpenTelemetry
                services.AddOpenTelemetry()
                    .WithTracing(builder => builder
                        .AddSource("KafkaFlow")
                        .AddJaegerExporter()
                    );
            });
}

services.AddOpenTelemetry()
.WithMetrics(builder => builder
    .AddMeter("KafkaFlow.Metrics")
    .AddPrometheusExporter());
 
// Добавление своих метрик
var meter = new Meter("KafkaFlow.Metrics");
var messagesProcessedCounter = meter.CreateCounter<int>("messages_processed");
 
// В обработчике сообщений
public async Task Handle(IMessageContext context, SomeMessage message)
{
    // Обработка сообщения
    messagesProcessedCounter.Add(1);
}

.AddCluster(cluster => cluster
    .WithBrokers(new[] { "kafka-broker:9092" })
    .WithSecurityInformation(security => security
        .WithSessionTimeoutMs(30000) // Дайте больше времени для ребалансировки
    )
)

.AddCustomErrorHandler((context, exception) => {
    var logger = context.ServiceProvider.GetRequiredService<ILogger<Program>>();
    logger.LogError(
        exception,
        "Ошибка обработки сообщения: Топик={Topic}, Партиция={Partition}, Смещение={Offset}, Ключ={Key}",
        context.Topic,
        context.Partition,
        context.Offset,
        context.Message.Key);
    return Task.CompletedTask;
})

var consumerLags = await consumer.GetLagAsync();
foreach (var lag in consumerLags)
{
    logger.LogInformation(
        "Lag для топика {Topic}, партиции {Partition}: {Lag}",
        lag.Topic,
        lag.Partition,
        lag.Lag);
}