EAC.Infrastructure.Messaging.Kafka 0.1.0-rc.4

EAC.Infrastructure.Messaging.Kafka

Adaptador Kafka del núcleo neutral EAC.Infrastructure.Messaging, empaquetado como EAC.Infrastructure.Messaging.Kafka para .NET 10.

KAFKA-001 establece la base técnica; KAFKA-002 incorpora el producer y KAFKA-003 el consumo gobernado con Inbox, offsets manuales y cuarentena durable. El paquete no registra servicios implícitamente.

Desarrollo local

./scripts/ci.sh

La secuencia ejecuta validación estructural y documental, restore bloqueado, formato, build Release y pruebas sobre los mismos binarios. Durante el desarrollo coordinado, el adaptador usa una referencia de proyecto determinista a EAC.Infrastructure.Messaging; el retorno a PackageReference y la publicación pertenecen al cierre de release. Confluent.Kafka 2.15.0 permanece como dependencia externa.

El repositorio no contiene Pipelines ni Tasks. Su binding mínimo .tekton fija el perfil reutilizable eac-nuget-ci con integration-profile: kafka del Pipeline Catalog v0.4.4; el perfil aporta Kafka efímero sin privilegios y las pruebas también admiten KAFKA_BOOTSTRAP_SERVERS.

• Índice documental
• Plan de implementación
• Diseño
• Guía de consumo y configuración
• Arquitectura transversal

Producer Kafka actual

El transporte productor se compone explícitamente; construirlo no ocurre al referenciar el paquete:

var channels = new KafkaChannelMap([
    new KafkaChannelMapping(
        new LogicalChannel("sample.order.events"),
        "dev.sample.order.events")
]);

using var transport = new KafkaMessageTransport(
    channels,
    new KafkaProducerOptions(
        "kafka:9093",
        "order-service",
        KafkaSecurityOptions.Tls("/var/run/secrets/kafka/ca.pem")));

El mapa rechaza canales duplicados y topics inválidos. En composición, cada canal declarado usa su propio nombre lógico como topic físico salvo que Eac:Messaging:Kafka:Channels declare un override válido para ese canal. El producer es único por instancia del transporte, fuerza idempotencia y acks=all, conserva orderingKey —o messageId cuando no existe— y publica el payload sin reserializarlo. Los headers usan nombres contractuales y valores UTF-8; no incluyen nombres CLR, payloads duplicados ni secretos.

La API de bajo nivel permanece disponible. AddEacKafka aporta composición DI explícita, binding validado desde appsettings y activación independiente de producer/consumers; no inicia workers ni crea topics.

Consumer Kafka actual

KafkaMessageConsumer recibe un IncomingMessageProcessor del núcleo neutral y un IKafkaDeadLetterStore implementado por la persistencia elegida por el host o por KafkaTopicDeadLetterStore del paquete. La sobrecarga AddEacKafkaConsumer<TDispatcher> selecciona este último mediante Consumers:{nombre}:DeadLetterTopic; la sobrecarga AddEacKafkaConsumer<TDispatcher, TDeadLetterStore> conserva proveedores alternativos. EnableAutoCommit y EnableAutoOffsetStore permanecen desactivados: un resultado Acknowledge confirma el siguiente offset, un resultado Retry reposiciona el mismo record y un resultado DeadLetter persiste primero la cuarentena. Cada instancia posee su propio consumer y un único endpoint explícito. La publicación DLT y el commit del origen son operaciones sucesivas: un fallo entre ambas puede producir duplicados al reentregar el origen.

La API de bajo nivel procesa un record por llamada mediante ConsumeOnceAsync. KAFKA-004 incorpora el scheduler interno que conserva orden dentro de cada partition, permite progreso paralelo entre partitions y emite señales específicas cuando una cola acotada requiere Pause o recupera capacidad para Resume; no cambia la semántica de Inbox ni adelanta offsets.

KAFKA-005 exige un KafkaSecurityOptions explícito: Tls y SaslScram son los perfiles de runtime; InsecureDevelopment declara de forma visible el único uso permitido de texto plano en pruebas aisladas. Assignment, revocation y loss actualizan ownership antes de cualquier commit tardío, los consumers ejecutan Close antes de Dispose y KafkaDiagnostics publica únicamente coordenadas y resultados seguros.

KAFKA-006 ejecuta el kit completo contra confluentinc/cp-kafka:7.7.1: publicación, orden, consumo, duplicados Inbox, retry sin commit, reasignación del mismo group a otra instancia y cuarentena durable.

El adaptador emite messaging.send como hijo de messaging.publish y propaga el traceparent W3C resultante. En recepción crea messaging.receive; el procesador Inbox del núcleo lo continúa con messaging.process, evitando dos spans Consumer paralelos. Los spans usan el canal lógico como destino y no incluyen topic físico, key, payload, identificadores ni valores de headers.

El meter EAC.Infrastructure.Messaging.Kafka publica eac.kafka.operations, eac.kafka.operation.duration en segundos y eac.kafka.errors, además de las observaciones de consumo y rebalanceo. Las métricas usan sólo operation, result y message.kind; topics, partitions, offsets, IDs y errores no se convierten en labels.

Composición pública

services.AddEacKafka(configuration, kafka => kafka
    .MapChannel("sample.order.events")
    .UseProducer()
    .AddConsumer("orders", "sample.order.events"));

La configuración, variables, secretos, overrides, errores y compatibilidad de publicación están definidos en la guía canónica. El paquete soporta trimming verificado; no declara Native AOT porque el cliente Confluent.Kafka 2.15.0 usa reflexión para cargar los entry points nativos.