spoud/kafka-datagen repository overview

⁠kafka-datagen

This is a reimplementation of the kafka-datagen⁠ project in Quarkus.

The same data streams can be generated with much lower startup time and memory footprint.

⁠Container quick start

Run the published image with the minimum required settings:

docker run --rm \
  -e KAFKA_BOOTSTRAP_SERVERS=broker:9092 \
  -e TOPIC=topic \
  -e SCHEMA_FILE=clickstream_users_schema.avro \
  -e SCHEMA_KEYFIELD=username \
  spoud/kafka-datagen:latest

The container:

• listens on port 8080
• runs as a non-root user (185:0)
• reads schemas from the image classpath or from a mounted file path
• exposes metrics on /q/metrics
• needs outbound access to Kafka and, for Avro production, Schema Registry

⁠What an agent or operator needs to know

If you want an agent to deploy or operate this image safely, these are the facts it needs up front:

1. Kafka connectivity

   • bootstrap servers
   • whether TLS is required
   • whether SASL is required
   • where credentials come from

2. Data contract

   • topic name
   • schema file name or mounted schema path
   • output format (json or avro)
   • schema key field, if any

3. Runtime behavior

   • target rate
   • whether generation is bounded with MAX_RECORDS
   • whether late events, duplicates, or poison pills are enabled

4. Infrastructure assumptions

   • whether the platform allows a writable /tmp
   • whether the platform allows outbound access to Kafka and Schema Registry
   • CPU and memory limits

5. Security posture

   • whether /q/metrics should be internal only
   • how secrets are injected
   • whether the deployment should use a read-only root filesystem

The repository does not know your Kafka auth model, your schema distribution approach, or your production network boundaries. An agent still needs those deployment-specific inputs.

⁠Runtime contract

⁠Configuration

Environment variables are mapped to Quarkus config using the usual MicroProfile naming rules.

Core settings:

• KAFKA_BOOTSTRAP_SERVERS: The Kafka bootstrap servers to connect to. Default: localhost:9092
• TOPIC: The Kafka topic to produce to. Default: test
• SCHEMA_FILE: The Avro schema file to use. Default: clickstream_users_schema.avro
• SCHEMA_KEYFIELD: The field in the Avro schema to use as the key. Default: null
• RATE: The number of records to produce per second. Default: 10
• MAX_RECORDS: The maximum number of records to produce. Default: 0 (unlimited)
• FORMAT: json or avro. Default: json
• SCHEMA_REGISTRY_URL: The URL of the Confluent Schema Registry

Late-event settings:

• LATE_EVENTS_PERCENTAGE: Percentage of generated records whose event timestamp should be shifted into the past. Decimal values such as 12.5 are supported. Default: 0
• LATE_EVENTS_MIN_MS: Minimum lateness in milliseconds. Default: 0
• LATE_EVENTS_MAX_MS: Maximum lateness in milliseconds. Default: 0
• LOG_LATE_EVENT: If set to true, log late-event details without requiring special Quarkus logger-category configuration. Default: false

Poison-pill settings:

• POISON_PILL_ENABLED: Enable poison-pill injection. Default: false
• POISON_PILL_COUNT: Number of poison pills to inject. Use -1 to repeat forever. Default: -1
• POISON_PILL_INTERVAL: Delay between poison pills as an ISO-8601 duration. Default: PT5M
• POISON_PILL_CLEAN_RECORDS: Minimum number of normal records to emit before poison pills are allowed. Default: 0
• POISON_PILL_CLEAN_DURATION: Minimum startup time before poison pills are allowed, as an ISO-8601 duration. Default: PT0S
• POISON_PILL_TYPE: Supported values: invalid-string, random-bytes. Default: invalid-string
• LOG_POISON_PILL: If set to true, log poison-pill details without requiring special Quarkus logger-category configuration. Default: false

Duplicate-event settings:

• DUPLICATE_EVENT_ENABLED: Enable duplicate-event replay. Default: false
• DUPLICATE_EVENT_COUNT: Number of duplicate events to inject. Use -1 to repeat forever. Default: -1
• DUPLICATE_EVENT_INTERVAL: Delay between duplicate events as an ISO-8601 duration. Default: PT5M
• DUPLICATE_EVENT_CLEAN_RECORDS: Minimum number of normal records to emit before duplicates are allowed. Default: 0
• DUPLICATE_EVENT_CLEAN_DURATION: Minimum startup time before duplicates are allowed, as an ISO-8601 duration. Default: PT0S
• LOG_DUPLICATE_EVENT: If set to true, log duplicate-event details without requiring special Quarkus logger-category configuration. Default: false

Feature-specific headers are configured with these property prefixes:

late.events.headers.<header-name>=<value>
poison.pill.headers.<header-name>=<value>
duplicate.event.headers.<header-name>=<value>

Important behavior notes:

• When using the Avro format, configure the Kafka value serializer accordingly:

  mp.messaging.outgoing.generated.value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
  

  Copy

• The JSON generator still requires SCHEMA_FILE, because the schema drives field generation.

• Late events require exactly one timestamp field in the Avro schema. The generator detects Avro logical timestamp fields (timestamp-millis or timestamp-micros) and fails fast when late events are enabled but the schema has none or more than one matching field.

• late.events.percentage accepts decimal values in the range 0..100.

• Duplicate events replay an earlier generated event with the original key and payload. Only headers are overridden for the duplicate record.

• Poison pills are emitted on a dedicated byte-oriented Kafka producer path so the payload can intentionally violate the normal topic contract.

⁠Schema lookup

SCHEMA_FILE can point to:

• a schema bundled in src/main/resources
• a path mounted into the container filesystem

If you mount your own schemas, make the path explicit:

docker run --rm \
  -v "$PWD/schemas:/schemas:ro" \
  -e KAFKA_BOOTSTRAP_SERVERS=broker:9092 \
  -e TOPIC=topic \
  -e SCHEMA_FILE=/schemas/payroll_bonus.avro \
  spoud/kafka-datagen:latest

⁠Observability

The application exposes:

• metrics on http://<host>:8080/q/metrics

There is currently no built-in authentication layer around that endpoint, so production deployments should normally expose it only on an internal network or through a secured ingress path.

⁠Container hardening and supply chain

The published JVM image now:

• runs as non-root (185:0)
• keeps application files owned by the runtime user and group-accessible for OpenShift-style execution
• removes world write access from /deployments
• uses /tmp as the JVM temp directory

Recommended runtime hardening:

• run with a read-only root filesystem
• mount a writable emptyDir or tmpfs at /tmp
• drop all Linux capabilities
• disallow privilege escalation
• use the default runtime seccomp profile
• avoid exposing /q/metrics publicly

The GitHub Actions image build publishes Docker BuildKit supply-chain metadata for pushed images:

• SBOM
• provenance attestation

⁠Secure deployment example

The example below assumes:

• non-root execution
• read-only root filesystem
• writable /tmp
• secrets injected from Kubernetes
• metrics exposed only inside the cluster

apiVersion: apps/v1
kind: Deployment
metadata:
  name: kafka-datagen
spec:
  replicas: 1
  selector:
    matchLabels:
      app: kafka-datagen
  template:
    metadata:
      labels:
        app: kafka-datagen
    spec:
      containers:
        - name: kafka-datagen
          image: spoud/kafka-datagen:latest
          imagePullPolicy: IfNotPresent
          ports:
            - containerPort: 8080
              name: http
          env:
            - name: KAFKA_BOOTSTRAP_SERVERS
              value: kafka-bootstrap.kafka.svc.cluster.local:9092
            - name: TOPIC
              value: payroll-bonus
            - name: SCHEMA_FILE
              value: /schemas/payroll_bonus.avro
            - name: FORMAT
              value: json
            - name: RATE
              value: "10"
            - name: MAX_RECORDS
              value: "0"
            - name: QUARKUS_HTTP_PORT
              value: "8080"
            - name: MP_MESSAGING_OUTGOING_GENERATED_SECURITY_PROTOCOL
              value: SASL_SSL
            - name: MP_MESSAGING_OUTGOING_GENERATED_SASL_MECHANISM
              value: SCRAM-SHA-512
            - name: MP_MESSAGING_OUTGOING_GENERATED_SASL_JAAS_CONFIG
              valueFrom:
                secretKeyRef:
                  name: kafka-datagen-kafka-auth
                  key: sasl-jaas-config
          volumeMounts:
            - name: schemas
              mountPath: /schemas
              readOnly: true
            - name: tmp
              mountPath: /tmp
          securityContext:
            runAsNonRoot: true
            allowPrivilegeEscalation: false
            readOnlyRootFilesystem: true
            capabilities:
              drop: ["ALL"]
            seccompProfile:
              type: RuntimeDefault
          resources:
            requests:
              cpu: 100m
              memory: 256Mi
            limits:
              cpu: 500m
              memory: 512Mi
      volumes:
        - name: schemas
          configMap:
            name: kafka-datagen-schemas
        - name: tmp
          emptyDir: {}
---
apiVersion: v1
kind: Service
metadata:
  name: kafka-datagen
spec:
  selector:
    app: kafka-datagen
  ports:
    - name: http
      port: 8080
      targetPort: http

Adapt the Kafka security properties to your environment. For Avro output, also set the serializer and Schema Registry properties on the generated outgoing channel.

⁠Local demo stack

The provided docker-compose.yml starts:

• Redpanda
• Redpanda Console
• Prometheus
• kafka-datagen

Start it with:

docker compose up --build

⁠Quarkus Project

This project uses Quarkus, the Supersonic Subatomic Java Framework.

If you want to learn more about Quarkus, please visit its website: https://quarkus.io/⁠.

⁠Running the application in dev mode

You can run your application in dev mode that enables live coding using:

./mvnw compile quarkus:dev

⁠Packaging and running the application

The application can be packaged using:

./mvnw package

It produces the quarkus-run.jar file in the target/quarkus-app/ directory.

The application is then runnable using:

java -jar target/quarkus-app/quarkus-run.jar

If you want to build an über-jar, execute:

./mvnw package -Dquarkus.package.jar.type=uber-jar

⁠Creating a native executable

You can create a native executable using:

./mvnw package -Dnative

Or, if you do not have GraalVM installed:

./mvnw package -Dnative -Dquarkus.native.container-build=true

You can then execute your native executable with:

./target/kafka-datagen-1.0.0-SNAPSHOT-runner

If you want to learn more about building native executables, please consult https://quarkus.io/guides/maven-tooling⁠.

⁠Related Guides

• Apache Kafka Client (guide⁠): Connect to Apache Kafka with its native API
• Messaging - Kafka Connector (guide⁠): Connect to Kafka with Reactive Messaging
• Confluent Schema Registry - Avro (guide⁠): Use Confluent as Avro schema registry