1. Kafka no Compose

Neste passo você substitui o seu docker-compose.yaml pelo arquivo completo deste lab, sobe os quatro novos contêineres de infraestrutura e conhece o Kafka pela linha de comando. Todos os comandos rodam na raiz do seu repositório boxoffice.

Contêiner	Imagem	Porta no host	Papel
`kafka`	`apache/kafka:4.3.1`	9094	um broker que também é o próprio controlador (KRaft)
`kafka-init`	`apache/kafka:4.3.1`	—	tarefa única: cria os tópicos e termina
`kafka-ui`	`kafbat/kafka-ui:v1.5.0`	8089	navegar por tópicos, mensagens, headers e lag dos consumidores
`mailpit`	`axllent/mailpit:v1.31`	8025 (web), 1025 (SMTP)	provedor de e-mail falso, com caixa de entrada e API

Kafka em modo KRaft

O Kafka guarda os próprios metadados (tópicos, partições, líderes) em um quórum Raft de controladores — o KRaft. O ZooKeeper não existe mais desde o Kafka 4.0. Aqui um único nó faz os dois papéis, broker,controller. Esta é a parte nova do docker-compose.yaml:

docker-compose.yaml (excerpt)

  kafka:
    image: apache/kafka:4.3.1
    hostname: kafka
    ports:
      - "9094:9094"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,EXTERNAL://localhost:9094
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
    healthcheck:
      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 > /dev/null 2>&1"]
      interval: 10s
      timeout: 10s
      start_period: 60s      # a busy laptop can take a minute to start the broker
      retries: 20

  kafka-init:
    image: apache/kafka:4.3.1
    depends_on:
      kafka:
        condition: service_healthy
    entrypoint: ["/bin/sh", "-c"]
    command:
      - |
        /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists \
          --topic order-events --partitions 6 --replication-factor 1 &&
        /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists \
          --topic order-events-dlt --partitions 6 --replication-factor 1
    restart: "no"

  kafka-ui:
    image: kafbat/kafka-ui:v1.5.0
    depends_on:
      kafka:
        condition: service_healthy
    ports:
      - "8089:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: boxoffice
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092

  mailpit:
    image: axllent/mailpit:v1.31
    ports:
      - "8025:8025"
      - "1025:1025"

Pegue o arquivo completo

Rode na raiz do seu repositório boxoffice (substitui a versão do lab 6; guarde uma cópia se você a alterou):

cp docker-compose.yaml docker-compose.lab6.yaml
curl -fsSL -o docker-compose.yaml https://insper.github.io/platform/2027.1/hands-on/7/code/docker-compose.yaml
docker compose config --services

Você deve ver os serviços do lab 6 mais kafka, kafka-init, kafka-ui, mailpit, order e notification. order e notification são construídos nos passos 2 e 3; até lá, suba só os serviços citados em cada comando. Apague docker-compose.lab6.yaml antes de commitar: o Git já guarda a versão do lab 6 (git show lab6:docker-compose.yaml).

docker-compose.yaml — o arquivo completo ao fim deste lab

docker-compose.yaml

name: boxoffice

# settings shared by the three catalog replicas (a YAML anchor)
x-catalog: &catalog
  build: ./services/catalog
  image: boxoffice-catalog
  environment:
    DB_HOST: db
    DB_PORT: 5432
    DB_NAME: ${DB_NAME:-boxoffice}
    DB_USER: ${DB_USER:-boxoffice}
    DB_PASSWORD: ${DB_PASSWORD:-boxoffice}
    DB_POOL_SIZE: ${DB_POOL_SIZE:-10}
    REDIS_HOST: redis
    REDIS_PORT: 6379
    CACHE_ENABLED: ${CACHE_ENABLED:-true}
    CACHE_TTL_EVENT: ${CACHE_TTL_EVENT:-60s}
    CACHE_TTL_LIST: ${CACHE_TTL_LIST:-10s}
    CACHE_STAMPEDE_PROTECTION: ${CACHE_STAMPEDE_PROTECTION:-false}
  depends_on:
    db:
      condition: service_healthy
    redis:
      condition: service_healthy
  deploy:
    resources:
      limits:
        cpus: "1.0"
        memory: 768M
  healthcheck:
    test: ["CMD-SHELL", "wget -qO- http://localhost:8082/actuator/health/readiness || exit 1"]
    interval: 10s
    timeout: 3s
    start_period: 40s
    retries: 3

services:

  db:
    image: postgres:17
    command: ["postgres", "-c", "max_connections=${DB_MAX_CONNECTIONS:-100}"]
    environment:
      POSTGRES_DB: ${DB_NAME:-boxoffice}
      POSTGRES_USER: ${DB_USER:-boxoffice}
      POSTGRES_PASSWORD: ${DB_PASSWORD:-boxoffice}
      REPLICATION_PASSWORD: ${REPLICATION_PASSWORD:-replicator}
    ports:
      - "5432:5432"
    volumes:
      - db-data:/var/lib/postgresql/data
      # runs only when the volume is new: creates the "replicator" user
      - ./infra/postgres/10-replication.sh:/docker-entrypoint-initdb.d/10-replication.sh:ro
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB}"]
      interval: 5s
      timeout: 3s
      retries: 10

  # read-only streaming replica of db (hot standby)
  db-replica:
    image: postgres:17
    user: postgres
    entrypoint: ["/usr/local/bin/replica-entrypoint.sh"]
    environment:
      PGPASSWORD: ${REPLICATION_PASSWORD:-replicator}     # used by pg_basebackup
      REPLICA_DELAY: ${REPLICA_DELAY:-0}                   # e.g. 2s: artificial lag
    ports:
      - "5433:5432"
    volumes:
      - replica-data:/var/lib/postgresql/data
      - ./infra/postgres/replica-entrypoint.sh:/usr/local/bin/replica-entrypoint.sh:ro
    depends_on:
      db:
        condition: service_healthy
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U boxoffice -d boxoffice"]
      interval: 5s
      timeout: 3s
      retries: 20

  pgbouncer:
    image: edoburu/pgbouncer:v1.26.0-p0
    ports:
      - "6432:6432"
    volumes:
      - ./infra/pgbouncer/pgbouncer.ini:/etc/pgbouncer/pgbouncer.ini:ro
      - ./infra/pgbouncer/userlist.txt:/etc/pgbouncer/userlist.txt:ro
    depends_on:
      db:
        condition: service_healthy

  redis:
    image: redis:8-alpine
    # a pure cache: bounded memory, evict least-recently-used keys, no persistence
    command: ["redis-server", "--maxmemory", "256mb", "--maxmemory-policy", "allkeys-lru",
              "--save", "", "--appendonly", "no"]
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 10

  # three identical replicas; no "ports:" — only the edge is reachable from outside
  catalog-1:
    <<: *catalog
    hostname: catalog-1
  catalog-2:
    <<: *catalog
    hostname: catalog-2
  catalog-3:
    <<: *catalog
    hostname: catalog-3

  inventory:
    build: ./services/inventory
    environment:
      DB_HOST: ${INVENTORY_DB_HOST:-pgbouncer}      # writes go through PgBouncer...
      DB_PORT: ${INVENTORY_DB_PORT:-6432}
      DB_MIGRATION_HOST: db                         # ...migrations straight to the primary
      DB_REPLICA_HOST: ${DB_REPLICA_HOST:-db-replica}
      DB_REPLICA_PORT: 5432
      DB_NAME: ${DB_NAME:-boxoffice}
      DB_USER: ${DB_USER:-boxoffice}
      DB_PASSWORD: ${DB_PASSWORD:-boxoffice}
      DB_POOL_SIZE: ${INVENTORY_POOL_SIZE:-20}
      INVENTORY_LOCKING: ${INVENTORY_LOCKING:-pessimistic}
      INVENTORY_HOLD_TTL: ${INVENTORY_HOLD_TTL:-PT10M}
    ports:
      - "8083:8083"            # published for experiments; clients go through the edge
    depends_on:
      db:
        condition: service_healthy
      db-replica:
        condition: service_healthy
      pgbouncer:
        condition: service_started
    deploy:
      resources:
        limits:
          cpus: "1.0"
          memory: 768M
    healthcheck:
      test: ["CMD-SHELL", "wget -qO- http://localhost:8083/actuator/health/readiness || exit 1"]
      interval: 10s
      timeout: 3s
      start_period: 40s
      retries: 3

  auth:
    build: ./services/auth
    environment:
      DB_HOST: db
      DB_PORT: 5432
      DB_NAME: ${DB_NAME:-boxoffice}
      DB_USER: ${DB_USER:-boxoffice}
      DB_PASSWORD: ${DB_PASSWORD:-boxoffice}
      JWT_ISSUER: ${JWT_ISSUER:-boxoffice-auth}
      JWT_TTL: ${JWT_TTL:-PT1H}
      ADMIN_EMAIL: ${ADMIN_EMAIL:-admin@boxoffice.dev}
      ADMIN_PASSWORD: ${ADMIN_PASSWORD:-boxoffice-admin}
    depends_on:
      db:
        condition: service_healthy
    deploy:
      resources:
        limits:
          cpus: "1.0"
          memory: 768M
    healthcheck:
      test: ["CMD-SHELL", "wget -qO- http://localhost:8081/actuator/health/readiness || exit 1"]
      interval: 10s
      timeout: 3s
      start_period: 40s
      retries: 3

  gateway:
    build: ./services/gateway
    environment:
      AUTH_URL: http://auth:8081
      INVENTORY_URL: http://inventory:8083
      AUTH_JWKS_URI: http://auth:8081/auth/.well-known/jwks.json
      JWT_ISSUER: ${JWT_ISSUER:-boxoffice-auth}
    ports:
      - "8080:8080"          # published for experiments; clients use the edge (port 80)
    depends_on:
      - auth
      - inventory
      - catalog-1
      - catalog-2
      - catalog-3
    deploy:
      resources:
        limits:
          cpus: "1.0"
          memory: 768M
    healthcheck:
      test: ["CMD-SHELL", "wget -qO- http://localhost:8080/actuator/health/readiness || exit 1"]
      interval: 10s
      timeout: 3s
      start_period: 40s
      retries: 3

  edge:
    image: nginx:1.30-alpine
    ports:
      - "80:80"
    volumes:
      - ./infra/nginx/nginx.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - gateway

  # ---- lab 7: Kafka (KRaft), Kafka UI, Mailpit, order and notification ----

  kafka:
    image: apache/kafka:4.3.1
    hostname: kafka
    ports:
      - "9094:9094"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,EXTERNAL://localhost:9094
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
    healthcheck:
      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 > /dev/null 2>&1"]
      interval: 10s
      timeout: 10s
      start_period: 60s      # a busy laptop can take a minute to start the broker
      retries: 20

  kafka-init:
    image: apache/kafka:4.3.1
    depends_on:
      kafka:
        condition: service_healthy
    entrypoint: ["/bin/sh", "-c"]
    command:
      - |
        /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists \
          --topic order-events --partitions 6 --replication-factor 1 &&
        /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --if-not-exists \
          --topic order-events-dlt --partitions 6 --replication-factor 1
    restart: "no"

  kafka-ui:
    image: kafbat/kafka-ui:v1.5.0
    depends_on:
      kafka:
        condition: service_healthy
    ports:
      - "8089:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: boxoffice
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092

  mailpit:
    image: axllent/mailpit:v1.31
    ports:
      - "8025:8025"
      - "1025:1025"

  order:
    build: ./services/order
    environment:
      DB_HOST: db
      DB_PORT: 5432
      DB_NAME: ${DB_NAME:-boxoffice}
      DB_USER: ${DB_USER:-boxoffice}
      DB_PASSWORD: ${DB_PASSWORD:-boxoffice}
      REDIS_HOST: redis
      KAFKA_BOOTSTRAP_SERVERS: kafka:9092
      NOTIFICATION_URL: http://notification:8086
      NOTIFY_MODE: ${NOTIFY_MODE:-outbox}            # outbox (after) | sync (before)
      ADMIT_PER_SECOND: ${ADMIT_PER_SECOND:-20}      # waiting-room admission rate
    ports:
      - "8084:8084"
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_healthy
      kafka-init:
        condition: service_completed_successfully
    deploy:
      resources:
        limits:
          cpus: "1.0"
          memory: 768M
    healthcheck:
      test: ["CMD-SHELL", "wget -qO- http://localhost:8084/actuator/health/readiness || exit 1"]
      interval: 10s
      timeout: 3s
      start_period: 40s
      retries: 3

  notification:
    build: ./services/notification
    environment:
      DB_HOST: db
      DB_PORT: 5432
      DB_NAME: ${DB_NAME:-boxoffice}
      DB_USER: ${DB_USER:-boxoffice}
      DB_PASSWORD: ${DB_PASSWORD:-boxoffice}
      KAFKA_BOOTSTRAP_SERVERS: kafka:9092
      MAIL_HOST: mailpit
      NOTIFY_CONCURRENCY: ${NOTIFY_CONCURRENCY:-3}   # consumer threads (max useful: 6 partitions)
      EMAIL_LATENCY: ${EMAIL_LATENCY:-50ms}          # simulated e-mail provider latency
      NOTIFY_IDEMPOTENT: ${NOTIFY_IDEMPOTENT:-true}  # false = no dedupe (to see duplicates)
    ports:
      - "8086:8086"
    depends_on:
      db:
        condition: service_healthy
      mailpit:
        condition: service_started
      kafka-init:
        condition: service_completed_successfully
    deploy:
      resources:
        limits:
          cpus: "1.0"
          memory: 768M
    healthcheck:
      test: ["CMD-SHELL", "wget -qO- http://localhost:8086/actuator/health/readiness || exit 1"]
      interval: 10s
      timeout: 3s
      start_period: 40s
      retries: 3

volumes:
  db-data:
  replica-data:

Três detalhes importam:

Dois listeners de cliente. Os contêineres chegam ao broker como kafka:9092 (PLAINTEXT). Um programa no seu notebook (uma execução na IDE, uma CLI) usa localhost:9094 (EXTERNAL). O cliente primeiro se conecta a qualquer endereço, depois recebe o endereço anunciado do líder de cada partição e se reconecta lá — então cada listener precisa anunciar um nome que os seus clientes consigam resolver. Trocar um pelo outro é o clássico bug "conecta e depois dá timeout". Por isso o Kafka é a única exceção à regra "mesma porta dentro e fora" do lab 0: os serviços usam kafka:9092, sua máquina usa localhost:9094.
Os tópicos são criados explicitamente (KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"). A criação automática usa o padrão do broker, 1 partição, o que limitaria em silêncio o paralelismo dos consumidores a 1.
O tópico de mensagens mortas tem tantas partições quanto o principal: o publicador de DLT do Spring grava um registro que falhou no mesmo número de partição em order-events-dlt.

Por que 6 partições?

Uma partição é consumida por no máximo um consumidor do grupo de cada vez; logo, o número de partições é o teto do paralelismo de consumo. Seis permitem que o grupo notification cresça até seis consumidores. Partições são baratas no começo e dolorosas de adicionar depois: aumentar o número de partições muda para qual partição cada chave vai, quebrando a ordem por chave das mensagens em trânsito.

Suba e confira

docker compose up -d kafka kafka-init kafka-ui mailpit ✔ Container boxoffice-mailpit-1 Started
✔ Container boxoffice-kafka-1 Healthy
✔ Container boxoffice-kafka-init-1 Started
✔ Container boxoffice-kafka-ui-1 Starteddocker compose logs kafka-initkafka-init-1 | Created topic order-events.
kafka-init-1 | Created topic order-events-dlt.docker compose exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --describe --topic order-eventsTopic: order-events TopicId: hwjrAir1Q8m_ZV4fyhnctw PartitionCount: 6 ReplicationFactor: 1 Configs: min.insync.replicas=1
Topic: order-events Partition: 0 Leader: 1 Replicas: 1 Isr: 1 Elr: LastKnownElr:
Topic: order-events Partition: 1 Leader: 1 Replicas: 1 Isr: 1 Elr: LastKnownElr:
Topic: order-events Partition: 2 Leader: 1 Replicas: 1 Isr: 1 Elr: LastKnownElr:
Topic: order-events Partition: 3 Leader: 1 Replicas: 1 Isr: 1 Elr: LastKnownElr:
Topic: order-events Partition: 4 Leader: 1 Replicas: 1 Isr: 1 Elr: LastKnownElr:
Topic: order-events Partition: 5 Leader: 1 Replicas: 1 Isr: 1 Elr: LastKnownElr:

Você deve ver kafka, kafka-ui e mailpit rodando e o kafka-init encerrado com código 0 (docker compose ps -a), seis partições em order-events, o Kafka UI em http://localhost:8089 (cluster boxoffice online, tópicos order-events e order-events-dlt) e a caixa vazia do Mailpit em http://localhost:8025. O TopicId muda em cada máquina. Se o kafka ficar em starting por mais de um minuto, veja Solução de problemas na visão geral.

Um passeio de cinco minutos pelo log

Antes que o Spring esconda tudo, sinta como um log difere de uma fila. Abra dois terminais.

No terminal A, inicie um consumidor no grupo tour:

docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka:9092 --topic order-events --group tour \
  --formatter-property print.key=true --formatter-property print.partition=true

No terminal B, produza três mensagens com chave (chave:valor):

docker compose exec -T kafka /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka:9092 --topic order-events \
  --reader-property parse.key=true --reader-property key.separator=: <<'EOF'
order-1:{"hello":"first"}
order-2:{"hello":"second"}
order-1:{"hello":"third"}
EOF

O terminal A imprime cada mensagem com sua partição, por exemplo:

Partition:4 order-1 {"hello":"first"}
Partition:3 order-2 {"hello":"second"}
Partition:4 order-1 {"hello":"third"}

As duas mensagens de order-1 caem na mesma partição — o particionador padrão faz o hash da chave (murmur2) módulo o número de partições — e essa é a única ordem que o Kafka promete: por partição, portanto por chave.

Agora pare o consumidor (Ctrl+C) e olhe o grupo:

docker compose exec kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe --group tour
Consumer group 'tour' has no active members.

GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
tour order-events 0 0 0 0 - - -
tour order-events 1 0 0 0 - - -
tour order-events 2 0 0 0 - - -
tour order-events 3 1 1 0 - - -
tour order-events 4 2 2 0 - - -
tour order-events 5 0 0 0 - - -

As mensagens continuam no log: consumir não apaga. Cada grupo só guarda seu offset por partição (CURRENT-OFFSET), e o lag = LOG-END-OFFSET − CURRENT-OFFSET diz quão atrasado ele está. Volte o grupo para o início e leia tudo de novo:

docker compose exec kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server kafka:9092 \
  --group tour --topic order-events --reset-offsets --to-earliest --execute

Essa capacidade de reprocessar é o que faz do Kafka uma boa espinha dorsal para novos consumidores (o analytics pode ler todo o histórico) — e também o motivo pelo qual todo consumidor precisa tolerar ver uma mensagem duas vezes.

Não apague as mensagens do passeio

As três mensagens de teste não são eventos válidos do BoxOffice. Elas serão úteis depois: no passo 3 elas são exatamente o tipo de mensagem envenenada que o consumidor precisa sobreviver. Deixe-as lá.

Pense primeiro: um grupo de consumidores com 8 instâncias em um tópico de 6 partições

Quantas instâncias recebem mensagens? O que as outras fazem?

Seis. Cada partição é atribuída a um membro do grupo, então duas instâncias ficam ociosas (elas assumem se outro membro morrer). Para usar mais consumidores, o tópico precisa de mais partições.

Próximo: Transactional outbox