Implementando customer.created ponta a ponta

Objetivos

• Publicar customer.created.v1 após criação de cliente no Spring
• Consumir evento no FastAPI com processamento e logs
• Validar o fluxo fim a fim com comandos reproduzíveis

Pré-requisitos

• Kafka em execução no Compose
• APIs de Customer e Order prontas
• Contrato JSON de evento criado

Conceito

Esta aula transforma conceito em integração real entre stacks diferentes. O customer-service (Java) publica o evento e o order-service (Python) reage. Isso prova que o sistema está desacoplado temporalmente e pronto para crescer sem chamadas HTTP em cascata.

O ponto central é contrato explícito. Quando producer e consumer compartilham um schema versionado, você reduz o risco de interpretação divergente do payload e facilita evolução futura (v2, v3) sem quebra.

Além de publicar e consumir, você vai testar de verdade com curl e logs. Sem validação operacional, fluxo assíncrono vira "fé" em vez de engenharia.

Estrutura de arquivos

contracts/events/customer-created.v1.json
services/customer-service/src/main/java/com/aprendi/customer/events/KafkaCustomerPublisher.java
services/customer-service/src/main/java/com/aprendi/customer/service/CustomerService.java
services/order-service/app/consumers/customer_created_consumer.py
services/order-service/app/main.py

Passo a passo

1. Definir contrato contracts/events/customer-created.v1.json

{
  "title": "customer.created.v1",
  "type": "object",
  "required": ["eventId", "occurredAt", "customerId", "fullName", "email"],
  "properties": {
    "eventId": { "type": "string" },
    "occurredAt": { "type": "string", "format": "date-time" },
    "customerId": { "type": "string" },
    "fullName": { "type": "string" },
    "email": { "type": "string", "format": "email" }
  }
}

2. Implementar publisher Kafka no Spring

package com.aprendi.customer.events;

import com.fasterxml.jackson.databind.ObjectMapper;
import java.time.Instant;
import java.util.Map;
import java.util.UUID;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class KafkaCustomerPublisher {
  private final KafkaTemplate<String, String> kafkaTemplate;
  private final ObjectMapper objectMapper = new ObjectMapper();

  public KafkaCustomerPublisher(KafkaTemplate<String, String> kafkaTemplate) {
    this.kafkaTemplate = kafkaTemplate;
  }

  public void publishCreated(String customerId, String fullName, String email) throws Exception {
    String eventId = UUID.randomUUID().toString();
    Map<String, Object> event = Map.of(
      "eventId", eventId,
      "occurredAt", Instant.now().toString(),
      "customerId", customerId,
      "fullName", fullName,
      "email", email
    );
    kafkaTemplate.send("customer.created.v1", customerId, objectMapper.writeValueAsString(event));
  }
}

3. Chamar publisher após criar cliente

// CustomerService.java
public Map<String, String> create(String fullName, String email) {
  String id = UUID.randomUUID().toString();
  repository.save(id, fullName, email);
  publisher.publishCreated(id, fullName, email);
  return Map.of("id", id, "fullName", fullName, "email", email);
}

4. Implementar consumer no Order (Python)

import json
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    "customer.created.v1",
    bootstrap_servers="kafka:9092",
    group_id="order-service-customer-group",
    auto_offset_reset="earliest",
    enable_auto_commit=True,
)

for msg in consumer:
    event = json.loads(msg.value.decode("utf-8"))
    print(f"[order-service] customer.created recebido: customerId={event['customerId']} offset={msg.offset}")

Como testar

1. Subir os serviços necessários:

docker compose -f infra/docker-compose.yml up --build -d kafka customer-service order-service

2. Criar cliente via HTTP:

curl -s -X POST http://localhost:8081/customers \
  -H "Content-Type: application/json" \
  -d '{"fullName":"Carla Mendes","email":"carla@aprendi.dev"}'

3. Conferir logs de publish/consume:

docker compose -f infra/docker-compose.yml logs -f customer-service order-service

Saída esperada:

customer-service | published customer.created.v1 key=<customerId>
order-service    | customer.created recebido: customerId=<customerId> offset=<n>

4. Validar tópico no broker:

docker compose -f infra/docker-compose.yml exec kafka \
  kafka-topics.sh --describe --topic customer.created.v1 --bootstrap-server kafka:9092

Dicas de projeto

• Sempre versione o nome do evento (v1) desde o começo.
• Use chave de partição com identidade de negócio (customerId).
• Registre eventId e offset nos logs.
• Centralize serialização JSON para reduzir erro de contrato.

Erros comuns

• Publicar evento antes de persistir cliente.
• Não tratar exceção de serialização no producer.
• Consumidor sem group_id explícito.
• Não monitorar logs e achar que o fluxo "sumiu".

Resumo

Você implementou o primeiro fluxo assíncrono real do curso com contrato versionado, publisher Java e consumer Python. O monorepo agora está pronto para encadear eventos de pedido e pagamento.