Configurando DLQ no Payment

Objetivos

• Materializar trilha de falha no Kafka.
• Salvar contexto completo para análise posterior.
• Isolar falhas sem parar o consumo principal.

Pré-requisitos

• Aula poison-messages-dlq concluída.
• Kafka acessível e usuário com permissão de escrita em tópico.
• payment-service usando segmentio/kafka-go.

Conceito

DLQ é um tópico dedicado para mensagens não processáveis. O payload original deve ser preservado e enriquecido com headers: motivo, serviço de origem, tentativa e timestamp.

Estrutura de arquivos

infra/kafka/topics/
└── topics.yaml
services/payment-service/internal/
├── dlq/publisher.go
└── consumer/payment_consumer.go

Passo a passo

1. Declare tópico no provisionamento (infra/kafka/topics/topics.yaml):

topics:
  - name: payment.events
    partitions: 3
    replicationFactor: 1
  - name: payment.dlq
    partitions: 3
    replicationFactor: 1

2. Crie publisher em dlq/publisher.go:

package dlq

import (
  "context"
  "fmt"
  "time"
  "github.com/segmentio/kafka-go"
)

type Publisher struct {
  writer *kafka.Writer
}

func NewPublisher(broker string) *Publisher {
  return &Publisher{
    writer: &kafka.Writer{
      Addr:     kafka.TCP(broker),
      Topic:    "payment.dlq",
      Balancer: &kafka.Hash{},
    },
  }
}

func (p *Publisher) Publish(ctx context.Context, original kafka.Message, reason error, attempts int) error {
  msg := kafka.Message{
    Key:   original.Key,
    Value: original.Value,
    Headers: []kafka.Header{
      {Key: "x-error-reason", Value: []byte(reason.Error())},
      {Key: "x-origin-topic", Value: []byte("payment.events")},
      {Key: "x-attempts", Value: []byte(fmt.Sprintf("%d", attempts))},
      {Key: "x-failed-at", Value: []byte(time.Now().UTC().Format(time.RFC3339))},
      {Key: "x-service", Value: []byte("payment-service")},
    },
  }
  return p.writer.WriteMessages(ctx, msg)
}

3. Chame no consumer:

if errors.IsPoison(err) {
  if publishErr := c.dlqPublisher.Publish(ctx, msg, err, attempts); publishErr != nil {
    c.logger.Error("falha ao enviar para dlq", "err", publishErr)
    return publishErr
  }
  c.logger.Warn("evento enviado para payment.dlq", "key", string(msg.Key))
  return nil
}

Como testar

docker compose up -d kafka payment-service
echo '{"type":"payment.requested","payload":{"orderId":"o-55","amountCents":-10}}' \
  | kcat -P -b localhost:9092 -t payment.events
kcat -C -b localhost:9092 -t payment.dlq -o beginning -e -f 'Key=%k Headers=%h Value=%s\n'

Dicas de projeto

• Preserve payload original sempre.
• Inclua headers padronizados para todas as DLQs do monorepo.
• Versione contrato de erro se o payload evoluir.

Erros comuns

• Publicar na DLQ sem eventId.
• Alterar payload durante envio à DLQ.
• Não registrar número de tentativas.

Resumo

Você configurou a DLQ do payment-service com tópico dedicado e headers úteis para triagem e auditoria.