Hellnet.Kafka 1.4.13

Hellnet.Kafka

Biblioteca opinionada de integração Kafka para microsserviços .NET.
Construída sobre Confluent.Kafka com resiliência via Polly.

dotnet add package Hellnet.Kafka

• Quickstart
• Serialização
• Resiliência
• Topic naming
• Dead Letter Queue
• MessageHandlerAttribute
• Configuração
• Arquitetura
• ADR
• Dependências

Quickstart

// Program.cs — única linha
builder.Services.AddHellnetKafka();

# .env — só o que muda por serviço
HELLNET_KAFKA_CONSUMER_GROUP=hellnet.meu-app.orders
HELLNET_KAFKA_SASL_PASSWORD=hellnet2026

Producer

public sealed record OrderCreated : IMessage
{
    public string MessageType => "order.created.v1";
    public string OrderId { get; init; } = "";
    public double Total { get; init; }
}

public class MeuService(IMessageBus bus)
{
    public async Task Criar() =>
        await bus.PublishAsync(new OrderCreated
        {
            OrderId = "123",
            Total = 299.90
        });
}

Consumer

public class OrderHandler : IMessageHandler<OrderCreated>
{
    private readonly ILogger<OrderHandler> _logger;

    public OrderHandler(ILogger<OrderHandler> logger) => _logger = logger;

    public Task HandleAsync(OrderCreated msg, IMessageContext ctx, CancellationToken ct)
    {
        _logger.LogInformation("Pedido {OrderId} recebido (partition {P}, offset {O})",
            msg.OrderId, ctx.Partition, ctx.Offset);
        return Task.CompletedTask;
    }
}

O handler é auto-descoberto via DI — apenas defina a classe que a biblioteca registra e inicia o consumer automaticamente.

Serialização

JSON

HELLNET_KAFKA_DEFAULT_SERIALIZER=json

Usa System.Text.Json com SnakeCaseLower. Não precisa de Schema Registry.
Recomendado para desenvolvimento local e mensagens simples.

Avro

HELLNET_KAFKA_DEFAULT_SERIALIZER=avro
HELLNET_KAFKA_SCHEMA_REGISTRY_URL=https://schema.hellnet.com.br

Usa AvroSerializer<T> do Confluent com Schema Registry.
Requer tipos Avro (ISpecificRecord) gerados de arquivos .avsc.

[AvroSchema(EmbeddedResource = "Schemas.order-created.avsc")]
public sealed record OrderCreated : IMessage
{
    public string MessageType => "order.created.v1";
}

Protobuf

HELLNET_KAFKA_DEFAULT_SERIALIZER=protobuf

Usa ProtobufSerializer<T> do Confluent com Schema Registry.

Resiliência

Todas as operações são envolvidas por pipelines Polly compostos por Timeout → Retry → Circuit Breaker:

Configurável por env vars (HELLNET_KAFKA_RETRY_DELAY_MS, HELLNET_KAFKA_TIMEOUT_PRODUCE_MS, HELLNET_KAFKA_CIRCUIT_BREAKER_COUNT, etc).

Fluxo de falha (Produce)

ProduceAsync falha (broker timeout)
  → Timeout 30s → estoura
  → Retry 1 (200ms + jitter)
  → Retry 2 (400ms + jitter)
  → Circuit breaker conta +1 (5/5 → OPEN)
  → Próximos produce por 30s: falham instantaneamente
  → Após 30s: half-open → testa → se ok → CLOSED

Topic naming

{prefix}.{messageType}
  │         └── IMessage.MessageType
  └── HELLNET_KAFKA_TOPIC_PREFIX (default: "hellnet")

Consumer group segue o padrão hellnet.{app}.{domain}.{event}.

Dead Letter Queue

Quando um handler esgota os retries, a mensagem é publicada no topic {topic}.dlq:

MessageHandlerAttribute

[MessageHandler(Topic = "custom.topic.v1", ConsumerGroup = "hellnet.custom", MaxRetries = 5)]
public class MeuHandler : IMessageHandler<MinhaMsg> { }

Sobrescreve o topic, consumer group e número de retries por handler.

Configuração

Env vars

Com AddHellnetKafka(), apenas HELLNET_KAFKA_CONSUMER_GROUP e HELLNET_KAFKA_SASL_PASSWORD são obrigatórios por serviço.

Config sem defaults Hellnet

services.AddHellnetKafka(new HellnetKafkaOptions
{
    Brokers = "localhost:9092",
    SecurityProtocol = "plaintext",
});

Arquitetura

App
 │
 ├── IMessageBus.PublishAsync<T>()     → Producer (SASL_SSL)
 │                                        resiliência via Polly
 │
 └── KafkaConsumerHost (BackgroundService)
      │
      └── IMessageHandler<T>.HandleAsync()
           │
           ├── Polly Retry Pipeline (expo backoff + jitter)
           │
           └── DeadLetterService (DLQ topic)

Fluxo de consumo

1. KafkaConsumerHost descobre todos os IMessageHandler<T> no assembly (auto-discover)
2. Cada handler ganha um consumer próprio com consumer group dedicado
3. Mensagem recebida → deserializa → executa HandleAsync
4. Se lançar exception → Polly retenta (exponential backoff com jitter)
5. Se esgotar retries → DeadLetterService publica no topic .dlq (com resiliência)
6. Offset é commitado apenas após sucesso ou DLQ

Estrutura do projeto

app/
├── Hellnet.Kafka.slnx
├── src/Hellnet.Kafka/
│   ├── Abstractions/         # Interfaces: IMessage, IMessageBus, IMessageHandler
│   ├── Configuration/        # Options, defaults, env binder, config builder
│   ├── Internal/             # KafkaMessageBus, KafkaConsumerHost, RetryEngine, DLQ
│   ├── Serialization/        # IMessageSerializer, Json, Avro
│   └── DependencyInjection.cs
├── tests/Hellnet.Kafka.UnitTests/
└── tests/Hellnet.Kafka.IntegrationTests/

ADR

As decisões arquiteturais da biblioteca estão documentadas no formato ADR (Architecture Decision Record).

Dependências

Licença

Apache 2.0 © 2026 Hellnet