کانفیگ کردن اصولی Kafka در محیط عملیاتی

private void ConfigKafka(KafkaConfig config)
{
    var consumerConfig = new ConsumerConfig
    {
        BootstrapServers = GetBootstrapServers(config.BootstrapServers),
        AutoOffsetReset = AutoOffsetReset.Earliest,
        GroupId = config.GroupId,
        ClientId = config.ClientId,
        SaslUsername = config.Username,
        SaslPassword = config.Password,
        SecurityProtocol = (SecurityProtocol?)config.SecurityProtocol,
        SaslMechanism = (SaslMechanism?)config.SaslMechanism,
        SessionTimeoutMs = 6000,
        EnableAutoCommit = false,
        EnablePartitionEof = true,
    };

    _consumer = new ConsumerBuilder<Ignore, string>(consumerConfig).Build();
}

private void ConfigKafkaPublisher(KafkaConfig config)
{
    var producerConfig = new ProducerConfig
    {
        BootstrapServers = GetBootstrapServers(config.BootstrapServers),
        SaslUsername = config.ProducerUsername,
        SaslPassword = config.ProducerPassword,
        ClientId = config.ClientId,
        SecurityProtocol = (SecurityProtocol?)config.SecurityProtocol,
        SaslMechanism = (SaslMechanism?)config.SaslMechanism,
        EnableIdempotence = true,
        Acks = Acks.All,
    };

    _producer = new ProducerBuilder<Null, string>(producerConfig).Build();
}

private string GetBootstrapServers(List<string> l)
{
    return l.Count == 1 ? l.First() : string.Join(",", l);
}

"Kafka": {
    "BootstrapServers": ["server-alpha-1:9043", "server-alpha-2:9043"],
    "TopicName": "topic",
    "ProducerTopic": "dead-letter",
    "GroupId": "group",
    "ClientId": "client",
    "Username": "kafka",
    "Password": "pass",
    "ProducerUsername": "kafka",
    "ProducerPassword": "pass",
    "SecurityProtocol": "SaslPlaintext",
    "SaslMechanism": "Plain"
}

private void Do(CancellationToken stoppingToken)
{
    IsLastMessage = false;

    _consumer.Subscribe(_kafkaConfig.TopicName);

    while ((!stoppingToken.IsCancellationRequested && _shouldWork) || !IsLastMessage)
    {
        try
        {
            var consumeResult = _consumer.Consume();

            if (consumeResult != null)
            {
                if (consumeResult.Message != null)
                {
                    using (Tracing.ActivitySource.StartActivity())
                    {
                        try
                        {
                            Activity.Current?.SetTag("Offset", consumeResult.Offset.Value);

                            OnMessageReceived?.Invoke(this, new MessageReceivedEventArgs(consumeResult.Message.Value));

                            _consumer.Commit();
                        }
                        catch (TimeoutException ex)
                        {
                            _logger.LogCritical(ex, $"Time out Invoke Kafka message: {ex.Message}");
                        }
                        catch (InvalidOperationException ex)
                        {
                            _logger.LogCritical(ex, $"Time out Invoke Kafka message: {ex.Message}");
                        }
                        catch (TaskCanceledException)
                        {
                            _logger.LogInformation("Task Cancellation");
                        }
                        catch (OperationCanceledException)
                        {
                            _logger.LogInformation("Task Cancellation");
                        }
                        catch (Exception ex)
                        {
                            _logger.LogCritical(ex, $"Error Invoke Kafka message: {ex.Message}");

                            _producer.Produce(_kafkaConfig.ProducerTopic, new Message<Null, string>
                            {
                                Value = consumeResult.Message.Value
                            });

                            _consumer.Commit();
                        }
                    }
                }

                IsLastMessage = consumeResult.IsPartitionEOF;

                if (IsLastMessage)
                {
                    _logger.LogInformation($"LastMessage, Offset: {consumeResult.Offset.Value}");
                }
            }
            else
            {
                Thread.Sleep(100);
            }
        }
        catch (OperationCanceledException)
        {
            _logger.LogInformation("Task Cancellation");

            Thread.Sleep(100);
        }
        catch (Exception ex)
        {
            _logger.LogCritical(ex, $"Error processing Kafka message: {ex.Message}");
        }
    }
}