Reprocessamento e monitoramento

Objetivos

• Transformar DLQ em mecanismo de recuperação, não depósito.
• Reenviar mensagens corrigidas para o tópico principal.
• Instrumentar métricas para alertas de operação.

Pré-requisitos

• Tópico payment.dlq configurado.
• Acesso de leitura/escrita no Kafka.
• Prometheus e Grafana no ambiente local.

Conceito

Reprocessamento deve ser controlado, em lote e com auditoria. Sem isso, você pode repetir o incidente em escala. O ideal é corrigir a causa raiz, executar dry-run e só depois reenviar.

Estrutura de arquivos

services/payment-service/
├── cmd/reprocess-dlq/main.go
└── internal/metrics/dlq_metrics.go
docs/runbooks/
└── dlq-reprocessamento.md

Passo a passo

1. Crie comando de reprocessamento (cmd/reprocess-dlq/main.go):

package main

import (
  "context"
  "log"
  "os"
  "github.com/segmentio/kafka-go"
)

func main() {
  broker := os.Getenv("KAFKA_BROKER")
  dryRun := os.Getenv("DRY_RUN") == "true"

  dlqReader := kafka.NewReader(kafka.ReaderConfig{
    Brokers: []string{broker},
    Topic:   "payment.dlq",
    GroupID: "payment-dlq-reprocessor",
  })
  defer dlqReader.Close()

  writer := &kafka.Writer{
    Addr:     kafka.TCP(broker),
    Topic:    "payment.events",
    Balancer: &kafka.Hash{},
  }
  defer writer.Close()

  for {
    msg, err := dlqReader.ReadMessage(context.Background())
    if err != nil {
      log.Fatal(err)
    }

    if dryRun {
      log.Printf("[DRY-RUN] key=%s headers=%v", string(msg.Key), msg.Headers)
      continue
    }

    if err := writer.WriteMessages(context.Background(), kafka.Message{
      Key:   msg.Key,
      Value: msg.Value,
    }); err != nil {
      log.Printf("falha no reenvio key=%s err=%v", string(msg.Key), err)
      continue
    }
    log.Printf("reprocessado key=%s", string(msg.Key))
  }
}

2. Exponha métricas DLQ (internal/metrics/dlq_metrics.go):

var (
  DlqReprocessedTotal = promauto.NewCounter(prometheus.CounterOpts{
    Name: "payment_dlq_reprocessed_total",
    Help: "Total de eventos reprocessados da DLQ",
  })
  DlqDepthGauge = promauto.NewGauge(prometheus.GaugeOpts{
    Name: "payment_dlq_depth",
    Help: "Quantidade estimada de mensagens pendentes na DLQ",
  })
)

3. Documente runbook (docs/runbooks/dlq-reprocessamento.md):

1. Corrigir bug que gerou falha.
2. Rodar reprocessador em dry-run.
3. Validar amostragem de 10 mensagens.
4. Executar reprocessamento real.
5. Monitorar lag e taxa de erro por 15 minutos.

Como testar

docker compose up -d kafka payment-service
DRY_RUN=true KAFKA_BROKER=localhost:9092 go run services/payment-service/cmd/reprocess-dlq/main.go
DRY_RUN=false KAFKA_BROKER=localhost:9092 go run services/payment-service/cmd/reprocess-dlq/main.go
curl -s http://localhost:8083/metrics | rg "payment_dlq_(reprocessed_total|depth)"

Dicas de projeto

• Reprocesse por lotes pequenos.
• Mantenha dry-run obrigatório em produção.
• Registre operador e janela do reprocessamento.

Erros comuns

• Reenviar sem corrigir causa raiz.
• Rodar reprocessamento concorrente sem controle.
• Não monitorar lag após reenvio.

Resumo

Você implementou reprocessamento seguro e monitoramento de DLQ, fechando o ciclo operacional de recuperação no payment-service.