Piping generated data to Kafka

This approach uses the avro-datagen CLI to generate JSON lines and pipes them into a Kafka console producer. No Python Kafka client needed — just the CLI and a running Kafka broker.

Architecture

┌──────────────┐   stdout    ┌──────────────────────┐         ┌─────────────┐
│              │   (pipe)    │ kafka-console-        │         │             │
│avro-datagen├────────────>│ producer              ├────────>│ Kafka topic │
│              │  JSON lines │ (runs in container)   │         │             │
└──────────────┘             └──────────────────────┘         └─────────────┘

Each JSON line becomes one Kafka message. No serialization overhead — the consumer deserializes JSON on the other side.

Quick start with docker-compose

1. Start Kafka

# docker-compose.yml
services:
  kafka:
    image: confluentinc/cp-kafka:7.9.0
    hostname: kafka
    ports:
      - "9092:9092"
      - "9093:9093"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:29093
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:29093,EXTERNAL://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,EXTERNAL://localhost:9093
      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
      CLUSTER_ID: "txn-aggregator-local"

docker compose up -d kafka

2. Create the topic

docker compose exec kafka \
  kafka-topics --create \
    --topic txn.raw \
    --bootstrap-server kafka:9092 \
    --partitions 3 \
    --replication-factor 1

3. Generate and pipe

From your host machine (Kafka exposed on localhost:9093):

# Generate 100 transactions and produce to Kafka
avro-datagen -s schemas/transaction.avsc -c 100 -r 1 \
  | docker exec -i kafka /opt/kafka/bin/kafka-console-producer.sh \
        --topic txn.raw \
        --bootstrap-server kafka:9092

Infinite mode (Ctrl+C to stop):

avro-datagen -s schemas/transaction.avsc -c 0 \
  | docker exec kafka \
      kafka-console-producer \
        --topic txn.raw \
        --bootstrap-server kafka:9092

Important: Use exec -T (no TTY) so stdin piping works correctly.

4. Verify messages landed

docker compose exec kafka \
  kafka-console-consumer \
    --topic txn.raw \
    --bootstrap-server kafka:9092 \
    --from-beginning \
    --max-messages 5

Rate limiting

Use the --rate flag to control records per second:

avro-datagen -s schemas/transaction.avsc -c 0 --rate 10 \
  | docker compose exec -T kafka \
      kafka-console-producer \
        --topic txn.raw \
        --bootstrap-server kafka:9092

This produces 10 records per second. Omit --rate for maximum throughput.

Using kcat (kafkacat) instead

kcat is lighter than the Kafka console producer and easier to pipe into:

# Install: brew install kcat
avro-datagen -s schemas/transaction.avsc -c 100 \
  | kcat -P -b localhost:9093 -t txn.raw

Seeded data for reproducible testing

# Always produces the same 50 records
avro-datagen -s schemas/transaction.avsc -c 50 --seed 42 \
  | kcat -P -b localhost:9093 -t txn.raw

This is useful for integration tests where you need deterministic data in the topic.

Tradeoffs vs. the Python producer app

	Pipe approach	Python producer app
Dependencies	None (CLI + Kafka container)	confluent-kafka
Rate control	--rate flag	--rate flag + producer tuning
Error handling	None — pipe breaks silently	Delivery callbacks, retries
Idempotent produce	No	Yes (enable.idempotence=true)
Headers	Not possible	Can set correlationId in Kafka headers
Best for	Quick local testing, demos	Continuous load simulation, staging

For production-like load testing, use the Python producer app approach described in kafka-producer-app.md. The pipe approach is ideal for quick local development and verifying the consumer works.