Погружение в Kafka: Концепции и примеры на C# с ASP.NET Core

var config = new ProducerConfig
{
    BootstrapServers = "localhost:9092",
    Debug = "broker,topic,msg"
};

   Партиции = (Пропускная способность потребителя / Пропускная способность одного потребителя) × Фактор безопасности

public interface IKafkaProducerService
{
    Task<DeliveryResult<string, string>> ProduceAsync(string topic, string key, string value);
    Task<DeliveryResult<string, TValue>> ProduceAsync<TValue>(string topic, string key, TValue value);
    void Dispose();
}

public class KafkaProducerService : IKafkaProducerService, IDisposable
{
    private readonly IProducer<string, string> _stringProducer;
    private readonly ILogger<KafkaProducerService> _logger;
    private readonly JsonSerializerOptions _jsonOptions;
    private bool _disposed;
 
    public KafkaProducerService(IConfiguration config, ILogger<KafkaProducerService> logger)
    {
        _logger = logger;
        
        var producerConfig = new ProducerConfig
        {
            BootstrapServers = config["Kafka:BootstrapServers"],
            ClientId = config["Kafka:ClientId"] ?? "default-producer",
            Acks = Acks.All,  // Ждём подтверждения от всех реплик
            EnableIdempotence = true,  // Предотвращение дублирования сообщений
            MessageSendMaxRetries = 3,  // Число повторных попыток
            RetryBackoffMs = 1000  // Интервал между попытками
        };
        
        _stringProducer = new ProducerBuilder<string, string>(producerConfig).Build();
        _jsonOptions = new JsonSerializerOptions { PropertyNamingPolicy = JsonNamingPolicy.CamelCase };
    }
 
    public async Task<DeliveryResult<string, string>> ProduceAsync(string topic, string key, string value)
    {
        try
        {
            return await _stringProducer.ProduceAsync(topic, new Message<string, string> 
            { 
                Key = key, 
                Value = value,
                Timestamp = new Timestamp(DateTime.UtcNow)
            });
        }
        catch (ProduceException<string, string> ex)
        {
            _logger.LogError(ex, "Ошибка при отправке сообщения в топик {Topic}. Ключ: {Key}", topic, key);
            throw;
        }
    }
 
    public async Task<DeliveryResult<string, TValue>> ProduceAsync<TValue>(string topic, string key, TValue value)
    {
        string serializedValue = JsonSerializer.Serialize(value, _jsonOptions);
        var result = await ProduceAsync(topic, key, serializedValue);
        
        // Адаптация типа результата для сохранения интерфейса
        return new DeliveryResult<string, TValue>
        {
            Topic = result.Topic,
            Partition = result.Partition,
            Offset = result.Offset,
            Timestamp = result.Timestamp,
            Message = new Message<string, TValue>
            {
                Key = result.Message.Key,
                // Value имитируется, реальный десериализованный объект тут не будет
                Headers = result.Message.Headers
            }
        };
    }
 
    public void Dispose()
    {
        if (_disposed) return;
        
        _stringProducer.Flush(TimeSpan.FromSeconds(10));
        _stringProducer.Dispose();
        _disposed = true;
        GC.SuppressFinalize(this);
    }
}

builder.Services.AddSingleton<IKafkaProducerService, KafkaProducerService>();

[ApiController]
[Route("api/[controller]")]
public class MessagesController : ControllerBase
{
    private readonly IKafkaProducerService _producer;
    private readonly ILogger<MessagesController> _logger;
 
    public MessagesController(IKafkaProducerService producer, ILogger<MessagesController> logger)
    {
        _producer = producer;
        _logger = logger;
    }
 
    [HttpPost("send")]
    public async Task<IActionResult> SendMessage([FromQuery] string topic, [FromBody] MessageDto message)
    {
        if (string.IsNullOrEmpty(topic))
            return BadRequest("Необходимо указать топик");
 
        try
        {
            var result = await _producer.ProduceAsync(topic, message.Key, message.Value);
            _logger.LogInformation(
                "Сообщение отправлено в топик {Topic}, раздел {Partition}, смещение {Offset}",
                result.Topic, result.Partition, result.Offset);
            
            return Ok(new
            {
                topic = result.Topic,
                partition = result.Partition.Value,
                offset = result.Offset.Value,
                timestamp = result.Timestamp.UtcDateTime
            });
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Ошибка при отправке сообщения");
            return StatusCode(500, "Произошла ошибка при отправке сообщения");
        }
    }
}
 
public class MessageDto
{
    public string Key { get; set; }
    public string Value { get; set; }
}

public class CustomPartitioner : IPartitioner
{
    public int Partition(string topic, int partitionCount, ReadOnlySpan<byte> keyData, 
                         ReadOnlySpan<byte> valueData, IHeaders headers)
    {
        if (keyData.Length == 0)
        {
            // Для сообщений без ключа используем случайную партицию
            return new Random().Next(0, partitionCount);
        }
        
        // Простой алгоритм: хэш ключа по модулю числа партиций
        var key = Encoding.UTF8.GetString(keyData);
        return Math.Abs(key.GetHashCode()) % partitionCount;
        
        // Альтернативные стратегии могут включать:
        // - Распределение по географическому признаку
        // - Распределение по времени (например, час дня)
        // - Распределение по типу данных или приоритету
    }
 
    public void Dispose() { }
}

var producer = new ProducerBuilder<string, string>(producerConfig)
    .SetPartitioner(new CustomPartitioner())
    .Build();

public void ProduceWithCallback(string topic, string key, string value, Action<DeliveryReport<string, string>> deliveryHandler)
{
    try
    {
        _stringProducer.Produce(topic, new Message<string, string> { Key = key, Value = value }, deliveryHandler);
    }
    catch (ProduceException<string, string> ex)
    {
        _logger.LogError(ex, "Ошибка при отправке сообщения в топик {Topic}", topic);
        throw;
    }
}

_producer.ProduceWithCallback("orders", "12345", orderJson, report =>
{
    if (report.Error.IsError)
    {
        _logger.LogError("Ошибка доставки сообщения: {Error}", report.Error.Reason);
    }
    else
    {
        _logger.LogInformation(
            "Сообщение успешно доставлено: топик={Topic}, раздел={Partition}, смещение={Offset}",
            report.Topic, report.Partition, report.Offset);
    }
});

try
{
    var result = await _producer.ProduceAsync(topic, key, message);
    // Обработка успешного результата
}
catch (ProduceException<string, string> ex) when (ex.Error.Code == ErrorCode.NetworkException)
{
    _logger.LogWarning("Временная недоступность сети, сообщение будет переотправлено");
    // Логика повторной отправки или постановки в локальную очередь
}
catch (ProduceException<string, string> ex) when (ex.Error.Code == ErrorCode.BrokerNotAvailable)
{
    _logger.LogWarning("Брокер недоступен, проверьте кластер Kafka");
    // Отправка уведомления администратору или запись в очередь повторных попыток
}
catch (ProduceException<string, string> ex)
{
    _logger.LogError(ex, "Другая ошибка отправки: {Reason}", ex.Error.Reason);
    throw;
}

public class KafkaJsonSerializer<T> : ISerializer<T>
{
    private readonly JsonSerializerOptions _options;
    
    public KafkaJsonSerializer(JsonSerializerOptions options = null)
    {
        _options = options ?? new JsonSerializerOptions 
        { 
            PropertyNamingPolicy = JsonNamingPolicy.CamelCase
        };
    }
    
    public byte[] Serialize(T data, SerializationContext context)
    {
        if (data == null)
            return null;
            
        return JsonSerializer.SerializeToUtf8Bytes(data, _options);
    }
}
 
// Использование в продюсере
var producerBuilder = new ProducerBuilder<string, OrderEvent>(producerConfig)
    .SetValueSerializer(new KafkaJsonSerializer<OrderEvent>());

var producerConfig = new ProducerConfig
{
    BootstrapServers = "localhost:9092",
    // Уникальный ID для транзакционного продюсера
    TransactionalId = "transaction-producer-1",
    EnableIdempotence = true,
    // Остальные настройки...
};
 
using var producer = new ProducerBuilder<string, string>(producerConfig).Build();
// Инициализация транзакций (выполняется один раз при запуске)
producer.InitTransactions(TimeSpan.FromSeconds(10));
 
try
{
    // Начало транзакции
    producer.BeginTransaction();
    
    // Отправка нескольких связанных сообщений
    await producer.ProduceAsync("orders", new Message<string, string> { Key = "123", Value = "Order created" });
    await producer.ProduceAsync("inventory", new Message<string, string> { Key = "123", Value = "Items reserved" });
    
    // Фиксация транзакции
    producer.CommitTransaction();
}
catch (Exception ex)
{
    // Откат транзакции при ошибке
    producer.AbortTransaction();
    _logger.LogError(ex, "Ошибка в транзакции Kafka");
    throw;
}

public async Task<DeliveryResult<string, string>> ProduceWithHeadersAsync(
    string topic, string key, string value, Dictionary<string, string> headers)
{
    var message = new Message<string, string>
    {
        Key = key,
        Value = value,
        Headers = new Headers()
    };
    
    // Добавление заголовков
    foreach (var header in headers)
    {
        message.Headers.Add(header.Key, Encoding.UTF8.GetBytes(header.Value));
    }
    
    // Трассировка запросов через заголовки
    message.Headers.Add("correlation-id", Encoding.UTF8.GetBytes(Guid.NewGuid().ToString()));
    message.Headers.Add("timestamp", Encoding.UTF8.GetBytes(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds().ToString()));
    
    return await _producer.ProduceAsync(topic, message);
}

public class KafkaProducerMetrics
{
    private readonly Counter _messageCounter;
    private readonly Counter _errorCounter;
    private readonly Histogram _latencyHistogram;
    
    // Инициализация метрик (например, с Prometheus.NET)
    public KafkaProducerMetrics(IMetricsFactory metricsFactory)
    {
        _messageCounter = metricsFactory.CreateCounter("kafka_producer_messages_total", 
            "Общее количество отправленных сообщений", "topic");
        
        _errorCounter = metricsFactory.CreateCounter("kafka_producer_errors_total", 
            "Количество ошибок при отправке", "topic", "error_type");
        
        _latencyHistogram = metricsFactory.CreateHistogram("kafka_producer_latency_ms", 
            "Время отправки сообщения в миллисекундах", "topic");
    }
    
    public void MessageSent(string topic, TimeSpan latency)
    {
        _messageCounter.Inc(new[] { topic });
        _latencyHistogram.Observe(latency.TotalMilliseconds, new[] { topic });
    }
    
    public void ErrorOccurred(string topic, string errorType)
    {
        _errorCounter.Inc(new[] { topic, errorType });
    }
}

public async Task<DeliveryResult<string, string>> ProduceAsync(string topic, string key, string value)
{
    var startTime = Stopwatch.StartNew();
    try
    {
        var result = await _stringProducer.ProduceAsync(topic, new Message<string, string> { Key = key, Value = value });
        
        // Записываем метрику успешной отправки
        _metrics.MessageSent(topic, startTime.Elapsed);
        
        return result;
    }
    catch (ProduceException<string, string> ex)
    {
        // Записываем метрику ошибки
        _metrics.ErrorOccurred(topic, ex.Error.Code.ToString());
        _logger.LogError(ex, "Ошибка при отправке сообщения в топик {Topic}", topic);
        throw;
    }
}

public interface IKafkaConsumerService
{
    Task StartConsumingAsync(string topic, CancellationToken cancellationToken);
    Task StopConsumingAsync();
    event EventHandler<KafkaMessageReceivedEventArgs> MessageReceived;
}
 
public class KafkaMessageReceivedEventArgs : EventArgs
{
    public string Topic { get; set; }
    public string Key { get; set; }
    public string Value { get; set; }
    public int Partition { get; set; }
    public long Offset { get; set; }
    public DateTime Timestamp { get; set; }
}

public class KafkaConsumerService : IKafkaConsumerService, IDisposable
{
    private readonly IConsumer<string, string> _consumer;
    private readonly ILogger<KafkaConsumerService> _logger;
    private Task _consumeTask;
    private CancellationTokenSource _cancellationTokenSource;
    private bool _disposed;
 
    public event EventHandler<KafkaMessageReceivedEventArgs> MessageReceived;
 
    public KafkaConsumerService(IConfiguration config, ILogger<KafkaConsumerService> logger)
    {
        _logger = logger;
        
        var consumerConfig = new ConsumerConfig
        {
            BootstrapServers = config["Kafka:BootstrapServers"],
            GroupId = config["Kafka:GroupId"] ?? "default-consumer-group",
            AutoOffsetReset = AutoOffsetReset.Earliest,
            EnableAutoCommit = true,
            AutoCommitIntervalMs = 5000,
            FetchWaitMaxMs = 5,
            FetchMaxBytes = 52428800 // 50 МБ
        };
        
        _consumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
    }
 
    public Task StartConsumingAsync(string topic, CancellationToken cancellationToken)
    {
        _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        
        _consumer.Subscribe(topic);
        _logger.LogInformation("Подписка на топик {Topic} выполнена", topic);
        
        _consumeTask = Task.Run(() => ConsumeMessages(_cancellationTokenSource.Token), cancellationToken);
        return Task.CompletedTask;
    }
 
    private async Task ConsumeMessages(CancellationToken cancellationToken)
    {
        try
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                try
                {
                    var consumeResult = _consumer.Consume(cancellationToken);
                    
                    if (consumeResult?.Message == null) continue;
                    
                    _logger.LogDebug(
                        "Сообщение получено: {Topic}-{Partition}@{Offset}: {Key} = {Value}",
                        consumeResult.Topic, consumeResult.Partition, consumeResult.Offset,
                        consumeResult.Message.Key, consumeResult.Message.Value);
                    
                    OnMessageReceived(consumeResult);
                    
                    // Опционально: ручная фиксация смещения
                    // _consumer.Commit(consumeResult);
                }
                catch (ConsumeException ex)
                {
                    _logger.LogError(ex, "Ошибка при потреблении сообщений");
                }
            }
        }
        catch (OperationCanceledException)
        {
            // Нормальное завершение при отмене токена
            _logger.LogInformation("Потребление сообщений остановлено");
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Непредвиденная ошибка при потреблении сообщений");
        }
        finally
        {
            _consumer.Close();
        }
    }
 
    protected virtual void OnMessageReceived(ConsumeResult<string, string> consumeResult)
    {
        MessageReceived?.Invoke(this, new KafkaMessageReceivedEventArgs
        {
            Topic = consumeResult.Topic,
            Key = consumeResult.Message.Key,
            Value = consumeResult.Message.Value,
            Partition = consumeResult.Partition.Value,
            Offset = consumeResult.Offset.Value,
            Timestamp = consumeResult.Message.Timestamp.UtcDateTime
        });
    }
 
    public Task StopConsumingAsync()
    {
        _cancellationTokenSource?.Cancel();
        return _consumeTask ?? Task.CompletedTask;
    }
 
    public void Dispose()
    {
        if (_disposed) return;
        
        _cancellationTokenSource?.Cancel();
        _cancellationTokenSource?.Dispose();
        _consumer?.Dispose();
        
        _disposed = true;
        GC.SuppressFinalize(this);
    }
}

public class KafkaConsumerHostedService : BackgroundService
{
    private readonly IKafkaConsumerService _consumerService;
    private readonly IServiceProvider _serviceProvider;
    private readonly ILogger<KafkaConsumerHostedService> _logger;
    private readonly string _topic;
 
    public KafkaConsumerHostedService(
        IKafkaConsumerService consumerService,
        IServiceProvider serviceProvider,
        IConfiguration config,
        ILogger<KafkaConsumerHostedService> logger)
    {
        _consumerService = consumerService;
        _serviceProvider = serviceProvider;
        _logger = logger;
        _topic = config["Kafka:ConsumerTopic"] ?? "default-topic";
        
        _consumerService.MessageReceived += OnMessageReceived;
    }
 
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("Kafka Consumer фоновая служба запущена");
        
        await _consumerService.StartConsumingAsync(_topic, stoppingToken);
        
        _logger.LogInformation("Kafka Consumer фоновая служба завершила работу");
    }
 
    private async void OnMessageReceived(object sender, KafkaMessageReceivedEventArgs e)
    {
        try
        {
            // Создаём область видимости для обработчика сообщений,
            // чтобы правильно управлять жизненным циклом служб
            using var scope = _serviceProvider.CreateScope();
            var messageHandler = scope.ServiceProvider.GetRequiredService<IKafkaMessageHandler>();
            
            await messageHandler.HandleMessageAsync(e.Topic, e.Key, e.Value);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Ошибка при обработке сообщения");
        }
    }
 
    public override async Task StopAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("Остановка Kafka Consumer фоновой службы");
        
        _consumerService.MessageReceived -= OnMessageReceived;
        await _consumerService.StopConsumingAsync();
        
        await base.StopAsync(cancellationToken);
    }
}

public interface IKafkaMessageHandler
{
    Task HandleMessageAsync(string topic, string key, string value);
}
 
public class OrderMessageHandler : IKafkaMessageHandler
{
    private readonly ILogger<OrderMessageHandler> _logger;
    private readonly IOrderService _orderService;
 
    public OrderMessageHandler(ILogger<OrderMessageHandler> logger, IOrderService orderService)
    {
        _logger = logger;
        _orderService = orderService;
    }
 
    public async Task HandleMessageAsync(string topic, string key, string value)
    {
        _logger.LogInformation("Обработка заказа из Kafka: {Key}", key);
        
        try
        {
            // Десериализация JSON сообщения в объект заказа
            var order = JsonSerializer.Deserialize<Order>(value);
            
            // Выполнение бизнес-логики
            await _orderService.ProcessOrderAsync(order);
            
            _logger.LogInformation("Заказ {OrderId} успешно обработан", order.Id);
        }
        catch (JsonException ex)
        {
            _logger.LogError(ex, "Ошибка десериализации сообщения: {Value}", value);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Ошибка при обработке заказа");
        }
    }
}

// В Program.cs
// Регистрация потребителя
builder.Services.AddSingleton<IKafkaConsumerService, KafkaConsumerService>();
 
// Регистрация обработчика сообщений
builder.Services.AddScoped<IKafkaMessageHandler, OrderMessageHandler>();
 
// Регистрация фоновой службы
builder.Services.AddHostedService<KafkaConsumerHostedService>();

public class ParallelKafkaConsumerService : IKafkaConsumerService, IDisposable
{
    private readonly ILogger<ParallelKafkaConsumerService> _logger;
    private readonly ConsumerConfig _consumerConfig;
    private readonly int _concurrencyLevel;
    private readonly List<IConsumer<string, string>> _consumers = new();
    private readonly List<Task> _consumeTasks = new();
    private CancellationTokenSource _cancellationTokenSource;
    private bool _disposed;
 
    public event EventHandler<KafkaMessageReceivedEventArgs> MessageReceived;
 
    public ParallelKafkaConsumerService(IConfiguration config, ILogger<ParallelKafkaConsumerService> logger)
    {
        _logger = logger;
        
        _consumerConfig = new ConsumerConfig
        {
            BootstrapServers = config["Kafka:BootstrapServers"],
            GroupId = config["Kafka:GroupId"] ?? "parallel-consumer-group",
            AutoOffsetReset = AutoOffsetReset.Earliest,
            EnableAutoCommit = true
        };
        
        // Количество параллельных потребителей - оптимально по количеству партиций
        _concurrencyLevel = int.Parse(config["Kafka:ConcurrencyLevel"] ?? "3");
    }
 
    public Task StartConsumingAsync(string topic, CancellationToken cancellationToken)
    {
        _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        
        for (int i = 0; i < _concurrencyLevel; i++)
        {
            var consumer = new ConsumerBuilder<string, string>(_consumerConfig)
                .SetErrorHandler((_, e) => _logger.LogError("Ошибка потребителя: {Error}", e.Reason))
                .Build();
            
            consumer.Subscribe(topic);
            _consumers.Add(consumer);
            
            var consumerTask = Task.Run(() => ConsumeMessages(consumer, _cancellationTokenSource.Token), 
                                     _cancellationTokenSource.Token);
            _consumeTasks.Add(consumerTask);
            
            _logger.LogInformation("Запущен потребитель #{ConsumerId} для топика {Topic}", i, topic);
        }
        
        return Task.CompletedTask;
    }
 
    private async Task ConsumeMessages(IConsumer<string, string> consumer, CancellationToken cancellationToken)
    {
        try
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                try
                {
                    var consumeResult = consumer.Consume(cancellationToken);
                    
                    if (consumeResult?.Message == null) continue;
                    
                    _logger.LogDebug(
                        "Сообщение получено: {Topic}-{Partition}@{Offset}",
                        consumeResult.Topic, consumeResult.Partition, consumeResult.Offset);
                    
                    OnMessageReceived(consumeResult);
                }
                catch (ConsumeException ex)
                {
                    _logger.LogError(ex, "Ошибка при потреблении сообщений");
                }
            }
        }
        catch (OperationCanceledException)
        {
            _logger.LogInformation("Потребление сообщений остановлено");
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Непредвиденная ошибка при потреблении сообщений");
        }
        finally
        {
            consumer.Close();
        }
    }
 
    protected virtual void OnMessageReceived(ConsumeResult<string, string> consumeResult)
    {
        MessageReceived?.Invoke(this, new KafkaMessageReceivedEventArgs
        {
            Topic = consumeResult.Topic,
            Key = consumeResult.Message.Key,
            Value = consumeResult.Message.Value,
            Partition = consumeResult.Partition.Value,
            Offset = consumeResult.Offset.Value,
            Timestamp = consumeResult.Message.Timestamp.UtcDateTime
        });
    }
 
    public async Task StopConsumingAsync()
    {
        _cancellationTokenSource?.Cancel();
        
        if (_consumeTasks.Any())
        {
            await Task.WhenAll(_consumeTasks);
            _consumeTasks.Clear();
        }
        
        foreach (var consumer in _consumers)
        {
            consumer.Close();
            consumer.Dispose();
        }
        
        _consumers.Clear();
    }
 
    public void Dispose()
    {
        if (_disposed) return;
        
        _cancellationTokenSource?.Cancel();
        _cancellationTokenSource?.Dispose();
        
        foreach (var consumer in _consumers)
        {
            consumer?.Dispose();
        }
        
        _disposed = true;
        GC.SuppressFinalize(this);
    }
}

public class ManualOffsetConsumerService : IKafkaConsumerService
{
    // ... основные поля и конструктор как у обычного сервиса ...
 
    public ManualOffsetConsumerService(IConfiguration config, ILogger<ManualOffsetConsumerService> logger)
    {
        _logger = logger;
        
        var consumerConfig = new ConsumerConfig
        {
            BootstrapServers = config["Kafka:BootstrapServers"],
            GroupId = config["Kafka:GroupId"] ?? "manual-offset-group",
            AutoOffsetReset = AutoOffsetReset.Earliest,
            // Отключаем автоматический коммит оффсетов
            EnableAutoCommit = false
        };
        
        _consumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
    }
 
    private async Task ConsumeMessages(CancellationToken cancellationToken)
    {
        try
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                try
                {
                    var consumeResult = _consumer.Consume(cancellationToken);
                    
                    if (consumeResult?.Message == null) continue;
                    
                    try
                    {
                        // Вызываем обработчик сообщения
                        OnMessageReceived(consumeResult);
                        
                        // После успешной обработки коммитим оффсет
                        _consumer.Commit(consumeResult);
                        
                        _logger.LogDebug(
                            "Оффсет зафиксирован: {Topic}-{Partition}@{Offset}",
                            consumeResult.Topic, consumeResult.Partition, consumeResult.Offset);
                    }
                    catch (Exception ex)
                    {
                        _logger.LogError(ex, "Ошибка при обработке сообщения. Оффсет не зафиксирован.");
                    }
                }
                catch (ConsumeException ex)
                {
                    _logger.LogError(ex, "Ошибка при потреблении сообщений");
                }
            }
        }
        // ... дальнейший код как в обычном сервисе ...
    }
}

public class RetryingConsumerService : IKafkaConsumerService
{
  // ... основные поля и конструктор ...
  private readonly IMessageProcessor _messageProcessor;
  private readonly RetryPolicy _retryPolicy;
 
  public RetryingConsumerService(
      IConfiguration config, 
      ILogger<RetryingConsumerService> logger,
      IMessageProcessor messageProcessor)
  {
      _logger = logger;
      _messageProcessor = messageProcessor;
      
      // Создаём политику повторных попыток с экспоненциальной задержкой
      _retryPolicy = Policy
          .Handle<Exception>()
          .WaitAndRetryAsync(
              retryCount: 5,
              sleepDurationProvider: attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)),
              onRetry: (ex, timeSpan, retryCount, context) =>
              {
                  _logger.LogWarning(
                      "Попытка {RetryCount} обработки сообщения не удалась. Следующая попытка через {RetryDelay}",
                      retryCount, timeSpan);
              });
      
      // ... остальная инициализация ...
  }
 
  private async Task ConsumeMessages(CancellationToken cancellationToken)
  {
      try
      {
          while (!cancellationToken.IsCancellationRequested)
          {
              ConsumeResult<string, string> consumeResult = null;
              
              try
              {
                  consumeResult = _consumer.Consume(cancellationToken);
                  if (consumeResult?.Message == null) continue;
                  
                  // Применяем политику повторных попыток к обработке сообщения
                  await _retryPolicy.ExecuteAsync(async () =>
                  {
                      await _messageProcessor.ProcessAsync(
                          consumeResult.Topic, 
                          consumeResult.Message.Key, 
                          consumeResult.Message.Value);
                  });
                  
                  // После успешной обработки фиксируем оффсет
                  _consumer.Commit(consumeResult);
              }
              catch (Exception ex)
              {
                  _logger.LogError(ex, "Не удалось обработать сообщение после всех повторных попыток");
                  
                  // Опционально: отправка в DLQ (Dead Letter Queue)
                  if (consumeResult != null)
                  {
                      await SendToDlqAsync(consumeResult);
                      // Фиксируем оффсет, чтобы не зацикливаться на проблемном сообщении
                      _consumer.Commit(consumeResult);
                  }
              }
          }
      }
      // ... обработка остальных исключений ...
  }
 
  private async Task SendToDlqAsync(ConsumeResult<string, string> failedMessage)
  {
      try
      {
          // Получаем инстанс продюсера для DLQ
          using var dlqProducer = new ProducerBuilder<string, string>(new ProducerConfig
          {
              BootstrapServers = _consumerConfig.BootstrapServers
          }).Build();
          
          // Создаём имя DLQ-топика
          string dlqTopic = $"{failedMessage.Topic}.dlq";
          
          // Добавляем информацию об ошибке в заголовки
          var headers = new Headers();
          headers.Add("original_topic", Encoding.UTF8.GetBytes(failedMessage.Topic));
          headers.Add("original_partition", BitConverter.GetBytes(failedMessage.Partition.Value));
          headers.Add("original_offset", BitConverter.GetBytes(failedMessage.Offset.Value));
          headers.Add("error_timestamp", Encoding.UTF8.GetBytes(DateTime.UtcNow.ToString("O")));
          
          // Отправляем в DLQ
          await dlqProducer.ProduceAsync(dlqTopic, new Message<string, string>
          {
              Key = failedMessage.Message.Key,
              Value = failedMessage.Message.Value,
              Headers = headers
          });
          
          _logger.LogInformation(
              "Сообщение перемещено в DLQ: {DlqTopic}, ключ: {Key}",
              dlqTopic, failedMessage.Message.Key);
      }
      catch (Exception ex)
      {
          _logger.LogError(ex, "Не удалось отправить сообщение в DLQ");
      }
  }
}

public class OrderedConsumerService : IKafkaConsumerService
{
  // ... основные поля ...
  private readonly ConcurrentDictionary<string, SemaphoreSlim> _keySemaphores = new();
  
  // ... конструктор и прочие методы ...
  
  private async Task ProcessMessageInOrder(ConsumeResult<string, string> consumeResult)
  {
      if (string.IsNullOrEmpty(consumeResult.Message.Key))
      {
          // Для сообщений без ключа порядок не гарантируется
          await ProcessMessageAsync(consumeResult);
          return;
      }
      
      // Получаем или создаём семафор для этого ключа
      var semaphore = _keySemaphores.GetOrAdd(
          consumeResult.Message.Key, 
          _ => new SemaphoreSlim(1, 1));
      
      try
      {
          // Ждём доступа к семафору
          await semaphore.WaitAsync();
          
          // Обрабатываем сообщение
          await ProcessMessageAsync(consumeResult);
      }
      finally
      {
          // Освобождаем семафор
          semaphore.Release();
          
          // Очистка неиспользуемых семафоров
          CleanupSemaphores();
      }
  }
  
  private async Task ProcessMessageAsync(ConsumeResult<string, string> consumeResult)
  {
      try
      {
          // Обработка сообщения
          // ...
          
          // Фиксация оффсета
          _consumer.Commit(consumeResult);
      }
      catch (Exception ex)
      {
          _logger.LogError(ex, "Ошибка при обработке сообщения");
      }
  }
  
  private void CleanupSemaphores()
  {
      // Периодически удаляем семафоры для ключей, которые не используются
      if (_keySemaphores.Count > 1000)  // Порог для очистки
      {
          foreach (var key in _keySemaphores.Keys)
          {
              if (_keySemaphores.TryGetValue(key, out var semaphore) && 
                  semaphore.CurrentCount == 1)  // Семафор не используется
              {
                  if (_keySemaphores.TryRemove(key, out var removedSemaphore))
                  {
                      removedSemaphore.Dispose();
                  }
              }
          }
      }
  }
}

public class KafkaDeserializer<T>
{
  private readonly JsonSerializerOptions _jsonOptions;
  
  public KafkaDeserializer(JsonSerializerOptions jsonOptions = null)
  {
      _jsonOptions = jsonOptions ?? new JsonSerializerOptions
      {
          PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
          PropertyNameCaseInsensitive = true
      };
  }
  
  public T Deserialize(string json)
  {
      if (string.IsNullOrEmpty(json))
          return default;
          
      return JsonSerializer.Deserialize<T>(json, _jsonOptions);
  }
}
 
// Использование в обработчике сообщений
public class GenericMessageHandler<T> : IMessageHandler
{
  private readonly KafkaDeserializer<T> _deserializer;
  private readonly ILogger _logger;
  
  public GenericMessageHandler(ILogger<GenericMessageHandler<T>> logger)
  {
      _deserializer = new KafkaDeserializer<T>();
      _logger = logger;
  }
  
  public Task HandleAsync(ConsumeResult<string, string> consumeResult)
  {
      try
      {
          T message = _deserializer.Deserialize(consumeResult.Message.Value);
          
          if (message == null)
          {
              _logger.LogWarning("Не удалось десериализовать сообщение");
              return Task.CompletedTask;
          }
          
          // Обработка десериализованного объекта
          _logger.LogInformation("Получено сообщение типа {MessageType}", typeof(T).Name);
          
          return ProcessMessageAsync(message);
      }
      catch (JsonException ex)
      {
          _logger.LogError(ex, "Ошибка десериализации сообщения из JSON");
          return Task.CompletedTask;
      }
  }
  
  protected virtual Task ProcessMessageAsync(T message)
  {
      // Переопределяется в конкретных обработчиках
      return Task.CompletedTask;
  }
}

public class ResourceEfficientConsumerService : IKafkaConsumerService, IDisposable, IAsyncDisposable
{
  // ... основные поля ...
  private bool _disposed;
  
  // Диспозер объекта
  public void Dispose()
  {
      Dispose(true);
      GC.SuppressFinalize(this);
  }
  
  // Асинхронная версия диспозера
  public async ValueTask DisposeAsync()
  {
      await DisposeAsyncCore();
      Dispose(false);
      GC.SuppressFinalize(this);
  }
  
  protected virtual void Dispose(bool disposing)
  {
      if (_disposed) return;
      
      if (disposing)
      {
          // Останавливаем потребление сообщений
          _cancellationTokenSource?.Cancel();
          
          // Для потокобезопасного завершения Kafka требует вызов Close()
          foreach (var consumer in _consumers)
          {
              try { consumer.Close(); } catch { /* игнорируем ошибки при закрытии */ }
              try { consumer.Dispose(); } catch { /* игнорируем ошибки при уничтожении */ }
          }
          
          _cancellationTokenSource?.Dispose();
      }
      
      _disposed = true;
  }
  
  protected virtual async ValueTask DisposeAsyncCore()
  {
      if (_consumeTasks.Any())
      {
          try
          {
              // Ждём корректного завершения всех потребителей с таймаутом
              await Task.WhenAny(
                  Task.WhenAll(_consumeTasks),
                  Task.Delay(TimeSpan.FromSeconds(10))
              );
          }
          catch (Exception ex)
          {
              // Логируем, но не выбрасываем исключение при уничтожении
              _logger.LogWarning(ex, "Ошибка при остановке потребителей Kafka");
          }
      }
  }
}

// Пример клиентского кода для работы с расширенным кластером
var config = new ProducerConfig
{
    BootstrapServers = "broker1:9092,broker2:9092,broker3:9092,newbroker:9092",
    // Остальные настройки...
};

// Запуск нескольких экземпляров потребителей с одинаковым GroupId
public static void Main(string[] args)
{
    // Создаём несколько экземпляров приложения-потребителя
    // с общим GroupId, но разными ClientId
    var consumerConfig = new ConsumerConfig
    {
        BootstrapServers = "localhost:9092",
        GroupId = "scaling-consumer-group",
        ClientId = $"consumer-{Environment.MachineName}-{Guid.NewGuid()}",
        AutoOffsetReset = AutoOffsetReset.Earliest
    };
}

public class KafkaMetricsReporter
{
    private Timer _metricsTimer;
    private readonly IProducer<string, string> _producer;
    private readonly IMetricsRegistry _metrics;
    
    public void StartReporting()
    {
        _metricsTimer = new Timer(_ =>
        {
            var stats = _producer.GetMetadata(TimeSpan.FromSeconds(5));
            foreach (var topic in stats.Topics)
            {
                _metrics.RecordGauge("kafka.topic.partition.count", topic.Partitions.Count, 
                    new Dictionary<string, string> { { "topic", topic.Topic } });
            }
        }, null, TimeSpan.Zero, TimeSpan.FromMinutes(1));
    }
}

public class KafkaEtlProcessor : BackgroundService
{
    private readonly IConsumer<string, string> _consumer;
    private readonly IDatabaseService _dbService;
    
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _consumer.Subscribe("raw-data-topic");
        
        while (!stoppingToken.IsCancellationRequested)
        {
            var consumeResult = _consumer.Consume(stoppingToken);
            
            // Этап Transform: преобразование данных
            var transformedData = TransformData(consumeResult.Message.Value);
            
            // Этап Load: загрузка в целевую систему
            await _dbService.SaveDataAsync(transformedData);
            
            // Фиксация обработанного смещения
            _consumer.Commit(consumeResult);
        }
    }
    
    private TransformedData TransformData(string rawData)
    {
        // Логика трансформации сырых данных
        // ...
        return new TransformedData();
    }
}

public class KafkaStreamProcessor
{
    private readonly IConsumer<string, string> _consumer;
    private readonly IProducer<string, string> _producer;
    
    public async Task ProcessStreamAsync(CancellationToken cancellationToken)
    {
        _consumer.Subscribe("input-topic");
        
        // Создаём наблюдаемую последовательность сообщений
        var messageStream = Observable.Create<ConsumeResult<string, string>>(observer =>
        {
            var cancellation = new CancellationTokenSource();
            
            Task.Run(() =>
            {
                try
                {
                    while (!cancellation.Token.IsCancellationRequested)
                    {
                        var message = _consumer.Consume(TimeSpan.FromMilliseconds(100));
                        if (message != null)
                            observer.OnNext(message);
                    }
                    
                    observer.OnCompleted();
                }
                catch (Exception ex)
                {
                    observer.OnError(ex);
                }
            }, cancellation.Token);
            
            return Disposable.Create(() => cancellation.Cancel());
        });
        
        // Применяем LINQ-операции для трансформации потока
        messageStream
            .Where(msg => !string.IsNullOrEmpty(msg.Message.Value))
            .Select(msg => new
            {
                OriginalMessage = msg,
                TransformedValue = TransformValue(msg.Message.Value)
            })
            .Buffer(TimeSpan.FromSeconds(1), 100) // Группируем по времени или количеству
            .Subscribe(async batch =>
            {
                foreach (var item in batch)
                {
                    await _producer.ProduceAsync("output-topic", 
                        new Message<string, string>
                        {
                            Key = item.OriginalMessage.Message.Key,
                            Value = item.TransformedValue
                        });
                    
                    // Коммитим оффсеты после успешной обработки
                    _consumer.Commit(item.OriginalMessage);
                }
            });
        
        await Task.Delay(Timeout.Infinite, cancellationToken);
    }
    
    private string TransformValue(string input)
    {
        // Бизнес-логика трансформации
        return input.ToUpperInvariant(); // Пример простого преобразования
    }
}

public class SchemaRegistryService
{
  private readonly ISchemaRegistryClient _schemaRegistry;
  private readonly ILogger<SchemaRegistryService> _logger;
  private readonly ConcurrentDictionary<string, int> _schemaIdCache = new();
 
  public SchemaRegistryService(IConfiguration config, ILogger<SchemaRegistryService> logger)
  {
      _logger = logger;
      
      var schemaRegistryConfig = new SchemaRegistryConfig
      {
          Url = config["Kafka:SchemaRegistry:Url"] ?? "http://localhost:8081"
      };
      
      _schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig);
  }
 
  public async Task<int> RegisterSchemaAsync<T>(string subject)
  {
      try
      {
          if (_schemaIdCache.TryGetValue(subject, out var cachedId))
              return cachedId;
          
          // Получаем схему из атрибутов класса T
          var schema = AvroSchema.CreateSchema(typeof(T));
          
          // Регистрируем схему или получаем существующий ID
          var schemaId = await _schemaRegistry.RegisterSchemaAsync(subject, schema);
          
          _schemaIdCache[subject] = schemaId;
          _logger.LogInformation("Схема для типа {Type} зарегистрирована с ID {SchemaId}", typeof(T).Name, schemaId);
          
          return schemaId;
      }
      catch (Exception ex)
      {
          _logger.LogError(ex, "Ошибка при регистрации схемы");
          throw;
      }
  }
 
  public async Task<bool> IsCompatibleAsync<T>(string subject)
  {
      try
      {
          var schema = AvroSchema.CreateSchema(typeof(T));
          var isCompatible = await _schemaRegistry.IsCompatibleAsync(subject, schema);
          
          if (!isCompatible)
              _logger.LogWarning("Схема для типа {Type} несовместима с существующей", typeof(T).Name);
              
          return isCompatible;
      }
      catch (Exception ex)
      {
          _logger.LogError(ex, "Ошибка при проверке совместимости схемы");
          return false;
      }
  }
}

public class AvroProducer<TKey, TValue> : IDisposable
{
  private readonly IProducer<TKey, TValue> _producer;
  private readonly ISchemaRegistryClient _schemaRegistry;
  private readonly string _topic;
  
  public AvroProducer(IConfiguration config, ISchemaRegistryClient schemaRegistry)
  {
      _schemaRegistry = schemaRegistry;
      _topic = config["Kafka:Topic"];
      
      var producerConfig = new ProducerConfig
      {
          BootstrapServers = config["Kafka:BootstrapServers"]
      };
      
      // Создаём продюсер с Avro-сериализаторами
      _producer = new ProducerBuilder<TKey, TValue>(producerConfig)
          .SetKeySerializer(new AvroSerializer<TKey>(_schemaRegistry))
          .SetValueSerializer(new AvroSerializer<TValue>(_schemaRegistry))
          .Build();
  }
  
  public async Task<DeliveryResult<TKey, TValue>> ProduceAsync(TKey key, TValue value)
  {
      return await _producer.ProduceAsync(_topic, new Message<TKey, TValue> { Key = key, Value = value });
  }
  
  public void Dispose()
  {
      _producer?.Dispose();
  }
}

// Исходная версия модели
[AvroSchema(@"{
  ""type"": ""record"",
  ""name"": ""OrderCreated"",
  ""fields"": [
    { ""name"": ""OrderId"", ""type"": ""string"" },
    { ""name"": ""CustomerId"", ""type"": ""string"" },
    { ""name"": ""TotalAmount"", ""type"": ""double"" }
  ]
}")]
public class OrderCreated
{
  public string OrderId { get; set; }
  public string CustomerId { get; set; }
  public double TotalAmount { get; set; }
}
 
// Эволюционировавшая версия модели
[AvroSchema(@"{
  ""type"": ""record"",
  ""name"": ""OrderCreated"",
  ""fields"": [
    { ""name"": ""OrderId"", ""type"": ""string"" },
    { ""name"": ""CustomerId"", ""type"": ""string"" },
    { ""name"": ""TotalAmount"", ""type"": ""double"" },
    { ""name"": ""CreatedAt"", ""type"": [""null"", ""string""], ""default"": null }
  ]
}")]
public class OrderCreated
{
  public string OrderId { get; set; }
  public string CustomerId { get; set; }
  public double TotalAmount { get; set; }
  public string CreatedAt { get; set; }
}

public class ResilientKafkaService
{
  private readonly ILogger<ResilientKafkaService> _logger;
  private readonly IProducer<string, string> _producer;
  private readonly AsyncCircuitBreakerPolicy _circuitBreaker;
  private readonly AsyncRetryPolicy _retryPolicy;
  
  public ResilientKafkaService(
      ProducerConfig producerConfig,
      ILogger<ResilientKafkaService> logger)
  {
      _logger = logger;
      
      // Политика повторных попыток
      _retryPolicy = Policy
          .Handle<KafkaException>(ex => ex.Error.Code == ErrorCode.NetworkException)
          .Or<KafkaException>(ex => ex.Error.Code == ErrorCode.UnknownServerError)
          .WaitAndRetryAsync(
              retryCount: 3,
              sleepDurationProvider: attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)),
              onRetry: (ex, delay, retryCount, _) =>
              {
                  _logger.LogWarning(ex, 
                      "Ошибка при отправке сообщения в Kafka. Повторная попытка {RetryCount} через {Delay}",
                      retryCount, delay);
              });
      
      // Реализация Circuit Breaker для предотвращения каскадных отказов
      _circuitBreaker = Policy
          .Handle<KafkaException>()
          .CircuitBreakerAsync(
              exceptionsAllowedBeforeBreaking: 5,
              durationOfBreak: TimeSpan.FromMinutes(1),
              onBreak: (ex, breakDuration) =>
              {
                  _logger.LogError(ex, 
                      "Circuit Breaker разомкнут на {BreakDuration} после повторяющихся ошибок",
                      breakDuration);
              },
              onReset: () =>
              {
                  _logger.LogInformation("Circuit Breaker восстановлен, соединение с Kafka возобновлено");
              });
      
      _producer = new ProducerBuilder<string, string>(producerConfig).Build();
  }
  
  public async Task<DeliveryResult<string, string>> ProduceWithResilienceAsync(
      string topic, string key, string value)
  {
      // Применяем обе политики
      return await _retryPolicy
          .WrapAsync(_circuitBreaker)
          .ExecuteAsync(async () =>
          {
              try
              {
                  return await _producer.ProduceAsync(topic, new Message<string, string>
                  {
                      Key = key,
                      Value = value
                  });
              }
              catch (ProduceException<string, string> ex)
              {
                  _logger.LogError(ex, "Ошибка при отправке. Код: {ErrorCode}, Причина: {Reason}",
                      ex.Error.Code, ex.Error.Reason);
                  throw;
              }
          });
  }
}

public class RecoverableConsumerService : IHostedService
{
  private readonly ILogger<RecoverableConsumerService> _logger;
  private readonly ConsumerConfig _consumerConfig;
  private IConsumer<string, string> _consumer;
  private Task _consumeTask;
  private CancellationTokenSource _cancellationTokenSource;
  
  public RecoverableConsumerService(
      IConfiguration config,
      ILogger<RecoverableConsumerService> logger)
  {
      _logger = logger;
      _consumerConfig = new ConsumerConfig
      {
          BootstrapServers = config["Kafka:BootstrapServers"],
          GroupId = config["Kafka:GroupId"],
          AutoOffsetReset = AutoOffsetReset.Earliest,
          EnableAutoCommit = false,
          // Важные настройки для восстановления:
          SessionTimeoutMs = 10000,             // Время обнаружения сбоя клиента
          HeartbeatIntervalMs = 3000,           // Частота отправки heartbeat
          MaxPollIntervalMs = 300000,           // Макс. время между poll() вызовами
          SocketConnectionSetupTimeoutMs = 10000 // Таймаут соединения
      };
  }
  
  public Task StartAsync(CancellationToken cancellationToken)
  {
      _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
      _consumeTask = Task.Run(() => ConsumeWithRecoveryAsync(_cancellationTokenSource.Token));
      return Task.CompletedTask;
  }
  
  private async Task ConsumeWithRecoveryAsync(CancellationToken cancellationToken)
  {
      while (!cancellationToken.IsCancellationRequested)
      {
          try
          {
              // Создаём новый консьюмер при каждой попытке восстановления
              using (_consumer = new ConsumerBuilder<string, string>(_consumerConfig).Build())
              {
                  _consumer.Subscribe("my-topic");
                  
                  _logger.LogInformation("Потребитель запущен и подписан на топик");
                  
                  // Основной цикл потребления
                  while (!cancellationToken.IsCancellationRequested)
                  {
                      try
                      {
                          var result = _consumer.Consume(TimeSpan.FromMilliseconds(100));
                          
                          if (result != null)
                          {
                              await ProcessMessageAsync(result);
                              
                              // Ручной коммит после успешной обработки
                              _consumer.Commit(result);
                          }
                      }
                      catch (ConsumeException ex) when (IsTransientError(ex))
                      {
                          _logger.LogWarning(ex, "Временная ошибка при потреблении, продолжаем...");
                          await Task.Delay(1000, cancellationToken);
                      }
                  }
              }
          }
          catch (Exception ex)
          {
              _logger.LogError(ex, "Критическая ошибка потребителя. Повторное подключение через 5 секунд...");
              
              // Даём системе время на восстановление перед следующей попыткой
              try
              {
                  await Task.Delay(5000, cancellationToken);
              }
              catch (OperationCanceledException)
              {
                  break;
              }
          }
      }
  }
  
  private bool IsTransientError(Exception ex)
  {
      if (ex is KafkaException kafkaEx)
      {
          return kafkaEx.Error.IsRetriable ||
                 kafkaEx.Error.Code == ErrorCode.NetworkException ||
                 kafkaEx.Error.Code == ErrorCode.NotCoordinatorForGroup;
      }
      return false;
  }
  
  private Task ProcessMessageAsync(ConsumeResult<string, string> result)
  {
      _logger.LogInformation("Обработка сообщения из топика {Topic}, раздел {Partition}, смещение {Offset}",
          result.Topic, result.Partition, result.Offset);
      
      // Бизнес-логика обработки...
      
      return Task.CompletedTask;
  }
  
  public async Task StopAsync(CancellationToken cancellationToken)
  {
      _cancellationTokenSource?.Cancel();
      
      try
      {
          if (_consumeTask != null)
              await Task.WhenAny(_consumeTask, Task.Delay(5000, cancellationToken));
      }
      catch
      {
          // Игнорируем ошибки при остановке
      }
      
      _consumer?.Close();
  }
}

var producerConfig = new ProducerConfig
{
  BootstrapServers = "localhost:9092",
  // Увеличение размера буфера памяти для пакетной отправки
  BatchSize = 16384 * 4,  // 64 КБ (по умолчанию 16 КБ)
  LingerMs = 5,           // Задержка отправки для накопления сообщений
  
  // Увеличение пропускной способности сети
  SocketSendBufferBytes = 1024 * 1024,  // 1 МБ
  
  // Оптимизация под ack-never для нескритичных данных
  // (для критичных использовать Acks.All и минимум 2 реплики)
  Acks = Acks.None,     
  
  // Ограничение памяти
  BufferingMaxKbytes = 128 * 1024,  // 128 МБ
  
  // Настройка сжатия
  CompressionType = CompressionType.Snappy  // Хороший баланс CPU/сжатие
};

public class BatchProducerService<TKey, TValue>
{
  private readonly IProducer<TKey, TValue> _producer;
  private readonly BlockingCollection<Message<TKey, TValue>> _messageQueue;
  private readonly Thread _backgroundThread;
  private readonly int _batchSize;
  private readonly int _flushIntervalMs;
  private readonly string _topic;
  private readonly CancellationTokenSource _cancellationTokenSource;
  private readonly ILogger _logger;
  
  public BatchProducerService(
      ProducerConfig config, 
      string topic,
      int batchSize = 100,
      int flushIntervalMs = 100,
      ILogger logger = null)
  {
      _producer = new ProducerBuilder<TKey, TValue>(config).Build();
      _messageQueue = new BlockingCollection<Message<TKey, TValue>>(new ConcurrentQueue<Message<TKey, TValue>>());
      _batchSize = batchSize;
      _flushIntervalMs = flushIntervalMs;
      _topic = topic;
      _logger = logger;
      
      _cancellationTokenSource = new CancellationTokenSource();
      _backgroundThread = new Thread(ProcessQueueAsync) { IsBackground = true };
      _backgroundThread.Start();
  }
  
  public void Enqueue(TKey key, TValue value)
  {
      _messageQueue.Add(new Message<TKey, TValue> { Key = key, Value = value });
  }
  
  private void ProcessQueueAsync()
  {
      var batch = new List<Message<TKey, TValue>>(_batchSize);
      var sw = Stopwatch.StartNew();
      
      while (!_cancellationTokenSource.IsCancellationRequested)
      {
          try
          {
              // Набираем батч или ждём макс. интервал
              while (batch.Count < _batchSize && sw.ElapsedMilliseconds < _flushIntervalMs)
              {
                  if (_messageQueue.TryTake(out var message, 
                          Math.Max(1, _flushIntervalMs - (int)sw.ElapsedMilliseconds)))
                  {
                      batch.Add(message);
                  }
              }
              
              if (batch.Count > 0)
              {
                  // Отправляем батч
                  foreach (var message in batch)
                  {
                      _producer.Produce(_topic, message, DeliveryReportHandler);
                  }
                  
                  // Обеспечиваем отправку всех накопленных сообщений
                  _producer.Flush(TimeSpan.FromMilliseconds(1000));
                  
                  _logger?.LogDebug("Отправлен батч из {Count} сообщений", batch.Count);
                  batch.Clear();
              }
              
              sw.Restart();
          }
          catch (Exception ex)
          {
              _logger?.LogError(ex, "Ошибка при обработке пакета сообщений");
              
              // Кратковременная пауза перед следующей попыткой
              Thread.Sleep(1000);
          }
      }
  }
  
  private void DeliveryReportHandler(DeliveryReport<TKey, TValue> report)
  {
      if (report.Error.IsError)
      {
          _logger?.LogWarning("Ошибка доставки: {Reason}", report.Error.Reason);
      }
  }
  
  public void Dispose()
  {
      _cancellationTokenSource.Cancel();
      _backgroundThread.Join(5000);
      _producer.Dispose();
      _messageQueue.Dispose();
  }
}

var consumerConfig = new ConsumerConfig
{
  BootstrapServers = "localhost:9092",
  GroupId = "performance-group",
  
  // Увеличение размера пакета чтения
  FetchMaxBytes = 52428800,  // 50 МБ
  MaxPartitionFetchBytes = 1048576,  // 1 МБ
  
  // Оптимизация коммитов оффсетов
  AutoCommitIntervalMs = 5000,  // 5 сек между автокоммитами
  
  // Оптимизация сессий
  SessionTimeoutMs = 30000,    // 30 сек
  HeartbeatIntervalMs = 10000, // 10 сек
  
  // Оптимизация перебалансировок
  MaxPollIntervalMs = 300000,  // 5 минут
  
  // Оптимизация буферов
  SocketReceiveBufferBytes = 1024 * 1024  // 1 МБ
};

public class HighThroughputConsumerService : BackgroundService
{
  private readonly IConsumer<string, string> _consumer;
  private readonly IServiceProvider _serviceProvider;
  private readonly ILogger<HighThroughputConsumerService> _logger;
  private readonly SemaphoreSlim _maxConcurrencySemaphore;
  private readonly ConcurrentDictionary<TopicPartition, SemaphoreSlim> _partitionSemaphores = new();
  
  public HighThroughputConsumerService(
      ConsumerConfig config,
      IServiceProvider serviceProvider,
      ILogger<HighThroughputConsumerService> logger,
      int maxConcurrency = 100)
  {
      _consumer = new ConsumerBuilder<string, string>(config).Build();
      _serviceProvider = serviceProvider;
      _logger = logger;
      _maxConcurrencySemaphore = new SemaphoreSlim(maxConcurrency);
  }
  
  protected override async Task ExecuteAsync(CancellationToken stoppingToken)
  {
      _consumer.Subscribe("high-volume-topic");
      
      try
      {
          while (!stoppingToken.IsCancellationRequested)
          {
              try
              {
                  var result = _consumer.Consume(TimeSpan.FromMilliseconds(100));
                  
                  if (result == null) continue;
                  
                  // Ограничение общей параллельности
                  await _maxConcurrencySemaphore.WaitAsync(stoppingToken);
                  
                  // Получаем семафор для конкретной партиции
                  var partitionSemaphore = _partitionSemaphores.GetOrAdd(
                      result.TopicPartition, _ => new SemaphoreSlim(1));
                  
                  // Запускаем обработку, сохраняя порядок внутри партиции
                  // но разрешая параллельную обработку разных партиций
                  _ = Task.Run(async () =>
                  {
                      try
                      {
                          await partitionSemaphore.WaitAsync();
                          
                          try
                          {
                              // Обработка сообщения
                              using var scope = _serviceProvider.CreateScope();
                              var handler = scope.ServiceProvider.GetRequiredService<IMessageHandler>();
                              await handler.HandleAsync(result.Message.Key, result.Message.Value);
                              
                              // Ручной коммит после успешной обработки
                              _consumer.Commit(result);
                              
                              _logger.LogDebug("Сообщение обработано: {Partition}@{Offset}", 
                                  result.Partition, result.Offset);
                          }
                          finally
                          {
                              partitionSemaphore.Release();
                          }
                      }
                      catch (Exception ex)
                      {
                          _logger.LogError(ex, "Ошибка при обработке сообщения");
                      }
                      finally
                      {
                          _maxConcurrencySemaphore.Release();
                      }
                  });
              }
              catch (OperationCanceledException)
              {
                  break;
              }
              catch (Exception ex)
              {
                  _logger.LogError(ex, "Ошибка при получении сообщения");
                  await Task.Delay(1000, stoppingToken);
              }
          }
      }
      finally
      {
          _consumer.Close();
      }
  }
}

// В Program.cs
var builder = WebApplication.CreateBuilder(args);
 
// Настройка пула потоков
ThreadPool.SetMinThreads(100, 100);
 
// Настройка Kestrel для повышения пропускной способности
builder.WebHost.ConfigureKestrel(options =>
{
  options.Limits.MaxConcurrentConnections = 1000;
  options.Limits.MaxRequestBodySize = 20 * 1024 * 1024; // 20 MB
  options.Limits.MinRequestBodyDataRate = null;
  
  // Настройка HTTP/2
  options.ListenLocalhost(5001, listenOptions =>
  {
      listenOptions.Protocols = HttpProtocols.Http2;
  });
});
 
// Увеличение времени жизни HttpClient для соединений с Kafka REST Proxy
builder.Services.AddHttpClient("KafkaRestClient")
  .SetHandlerLifetime(TimeSpan.FromMinutes(5))
  .ConfigurePrimaryHttpMessageHandler(() => new SocketsHttpHandler
  {
      PooledConnectionLifetime = TimeSpan.FromMinutes(10),
      MaxConnectionsPerServer = 100
  });
 
// Оптимизация сериализации JSON для сообщений Kafka
builder.Services.Configure<JsonSerializerOptions>(options =>
{
  options.DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull;
  options.PropertyNamingPolicy = JsonNamingPolicy.CamelCase;
  options.Converters.Add(new JsonStringEnumConverter());
});

public class OrderServiceTests
{
  [Fact]
  public async Task ProcessOrder_ShouldSendToKafka_WhenOrderIsValid()
  {
      // Arrange
      var mockProducer = new Mock<IKafkaProducerService>();
      mockProducer
          .Setup(x => x.ProduceAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>()))
          .ReturnsAsync(new DeliveryResult<string, string>());
      
      var orderService = new OrderService(mockProducer.Object);
      var order = new Order { Id = "12345", CustomerId = "C789", Amount = 100.0m };
      
      // Act
      await orderService.ProcessAsync(order);
      
      // Assert
      mockProducer.Verify(
          x => x.ProduceAsync(
              "orders",
              It.Is<string>(k => k == order.Id),
              It.IsAny<string>()),
          Times.Once);
  }
}

public class KafkaIntegrationTests : IAsyncLifetime
{
  private readonly KafkaContainer _kafka;
  private IKafkaProducerService _producer;
  private IKafkaConsumerService _consumer;
  private List<string> _receivedMessages = new();
  
  public KafkaIntegrationTests()
  {
      _kafka = new KafkaContainerBuilder()
          .WithImage("confluentinc/cp-kafka:7.3.0")
          .WithZookeeper()
          .Build();
  }
  
  public async Task InitializeAsync()
  {
      // Запуск контейнера с Kafka
      await _kafka.StartAsync();
      
      // Настройка продюсера
      var producerConfig = new ProducerConfig
      {
          BootstrapServers = _kafka.GetBootstrapServers()
      };
      _producer = new KafkaProducerService(producerConfig);
      
      // Настройка потребителя
      var consumerConfig = new ConsumerConfig
      {
          BootstrapServers = _kafka.GetBootstrapServers(),
          GroupId = "test-group",
          AutoOffsetReset = AutoOffsetReset.Earliest
      };
      _consumer = new KafkaConsumerService(consumerConfig);
      
      // Подписка на события
      _consumer.MessageReceived += (sender, args) =>
      {
          _receivedMessages.Add(args.Value);
      };
      
      // Создание тестового топика
      using var adminClient = new AdminClientBuilder(
          new AdminClientConfig { BootstrapServers = _kafka.GetBootstrapServers() })
          .Build();
      await adminClient.CreateTopicsAsync(new TopicSpecification[]
      {
          new TopicSpecification { Name = "test-topic", NumPartitions = 1, ReplicationFactor = 1 }
      });
      
      // Запуск консьюмера
      await _consumer.StartConsumingAsync("test-topic", CancellationToken.None);
  }
  
  [Fact]
  public async Task ProducedMessage_ShouldBeConsumed()
  {
      // Arrange
      var testMessage = $"Test message {DateTime.UtcNow}";
      var testKey = Guid.NewGuid().ToString();
      
      // Act
      await _producer.ProduceAsync("test-topic", testKey, testMessage);
      
      // Assert
      // Ждём получения сообщения с таймаутом
      await WaitForConditionAsync(
          () => _receivedMessages.Contains(testMessage),
          TimeSpan.FromSeconds(10));
      
      Assert.Contains(testMessage, _receivedMessages);
  }
  
  private static async Task WaitForConditionAsync(Func<bool> condition, TimeSpan timeout)
  {
      var sw = Stopwatch.StartNew();
      while (!condition() && sw.Elapsed < timeout)
      {
          await Task.Delay(100);
      }
      
      if (!condition())
          throw new TimeoutException($"Тайм-аут ожидания условия после {timeout}");
  }
  
  public async Task DisposeAsync()
  {
      await _consumer.StopConsumingAsync();
      await _kafka.StopAsync();
  }
}

public class KafkaStressTest
{
  [Fact]
  public async Task ShouldHandleHighThroughput()
  {
      // Настройка тестового окружения
      var kafkaHelper = new KafkaTestHelper(); // Вспомогательный класс для тестов
      await kafkaHelper.InitializeAsync();
      
      try
      {
          // Подготовка тестовых данных
          const int messageCount = 100_000;
          var messages = Enumerable.Range(1, messageCount)
              .Select(i => new TestMessage { Id = i, Text = $"Message {i}" })
              .ToList();
          
          // Запуск параллельной отправки с измерением времени
          var sw = Stopwatch.StartNew();
          
          await Task.WhenAll(
              Partitioner.Create(messages).GetPartitions(8)
              .Select(partition => Task.Run(async () =>
              {
                  foreach (var msg in partition)
                  {
                      await kafkaHelper.Producer.ProduceAsync(
                          "stress-test", 
                          msg.Id.ToString(), 
                          JsonSerializer.Serialize(msg));
                  }
              }))
          );
          
          sw.Stop();
          var throughput = messageCount / sw.Elapsed.TotalSeconds;
          
          // Верификация результатов
          _output.WriteLine($"Throughput: {throughput:N0} messages/sec");
          
          // Проверка, что все сообщения были доставлены
          var topicInfo = await kafkaHelper.GetTopicInfoAsync("stress-test");
          var messageCountInTopic = topicInfo.Partitions.Sum(p => p.HighWatermark);
          
          Assert.Equal(messageCount, messageCountInTopic);
      }
      finally
      {
          await kafkaHelper.CleanupAsync();
      }
  }
  
  [Fact]
  public async Task ShouldRecoverFromBrokerFailure()
  {
      // Имитация отказа брокера путём остановки контейнера
      // и проверка механизмов восстановления
      var kafkaCluster = new KafkaContainerCluster(3); // 3-нодовый кластер
      await kafkaCluster.StartAsync();
      
      try
      {
          // Настройка топика с репликацией
          await kafkaCluster.CreateTopicAsync("resilience-test", 
              partitions: 4, replicationFactor: 3);
          
          // Отправка тестовых сообщений
          var producer = new KafkaProducerService(new ProducerConfig 
          { 
              BootstrapServers = kafkaCluster.BootstrapServers,
              Acks = Acks.All // Требуем подтверждения от всех ISR
          });
          
          for (int i = 0; i < 1000; i++)
          {
              await producer.ProduceAsync("resilience-test", i.ToString(), $"Value-{i}");
          }
          
          // Остановка одного брокера
          await kafkaCluster.StopBrokerAsync(1);
          _output.WriteLine("Брокер 1 остановлен");
          
          // Проверка, что система всё ещё функционирует
          for (int i = 1000; i < 2000; i++)
          {
              await producer.ProduceAsync("resilience-test", i.ToString(), $"Value-{i}");
          }
          
          // Восстановление брокера
          await kafkaCluster.StartBrokerAsync(1);
          _output.WriteLine("Брокер 1 восстановлен");
          
          // Верификация всех данных
          var consumer = new KafkaConsumerService(new ConsumerConfig
          {
              BootstrapServers = kafkaCluster.BootstrapServers,
              GroupId = "test-verification",
              AutoOffsetReset = AutoOffsetReset.Earliest
          });
          
          var messages = await consumer.ConsumeAllAsync("resilience-test", 
              timeout: TimeSpan.FromMinutes(1));
          
          Assert.Equal(2000, messages.Count);
      }
      finally
      {
          await kafkaCluster.StopAsync();
      }
  }
}