Skip to main content

Kafka Producer Consumer Loop

This distributed use case simulates event-driven behavior across API, producer, and consumer services while FlowLayer keeps startup dependencies explicit.

Server config for a Kafka event loop​

{
  "session": {
    "bind": "127.0.0.1:6999",
    "token": "dev-token-kafka"
  },
  "services": {
    "postgres": {
      "cmd": ["postgres", "-D", "./var/postgres"],
      "port": 5432,
      "ready": {
        "type": "tcp"
      }
    },
    "redis": {
      "cmd": ["redis-server", "./infra/redis.conf"],
      "port": 6379,
      "ready": {
        "type": "tcp"
      }
    },
    "kafka": {
      "cmd": ["kafka-server-start.sh", "./infra/kafka/server.properties"],
      "port": 9092,
      "ready": {
        "type": "tcp"
      }
    },
    "api": {
      "cmd": ["pnpm", "--dir", "services/api", "dev"],
      "dependsOn": ["postgres", "redis", "kafka"],
      "port": 3000,
      "ready": {
        "type": "http",
        "url": "http://127.0.0.1:3000/health"
      }
    },
    "billing-producer": {
      "cmd": ["pnpm", "--dir", "services/billing-producer", "dev"],
      "dependsOn": ["api", "kafka"]
    },
    "billing-consumer": {
      "cmd": ["pnpm", "--dir", "services/billing-consumer", "dev"],
      "dependsOn": ["kafka", "postgres", "redis"]
    }
  }
}

Start and operate the session​

flowlayer-server -c ./flowlayer.jsonc
flowlayer-client-tui -addr 127.0.0.1:6999 -token dev-token-kafka

Operational ideas:

  • restart billing-consumer after topic schema changes
  • stop only billing-producer to inspect consumer lag behavior
  • compare API and consumer logs during event bursts

Optional protocol-level log pull​

When you need scripted validation, use get_logs over WebSocket:

{
  "type": "command",
  "id": "logs-consumer-1",
  "name": "get_logs",
  "payload": {
    "service": "billing-consumer",
    "limit": 300
  }
}

See Logs and Events for message details.

Security and scope​

  • Keep session tokens out of version control.
  • Run this loop in local dev, CI sandboxes, or dedicated test hosts.
  • Do not aim this setup at production Kafka clusters.

Boundaries​

  • FlowLayer helps run and observe development process topologies.
  • It does not replace Kafka platform operations, Kubernetes controllers, Docker Compose stacks, PM2, or systemd.