Build event-driven triggers with Kafka and KEDA autoscaling

Introduction

When you run a fixed-schedule cron pipeline against bursty document ingestion, you waste compute during quiet hours and leave thousands of documents queued during traffic spikes. Event-driven pipelines triggered by Kafka and scaled by KEDA fix both ends: workers wake up when messages arrive and scale back to zero when the topic is idle. Getting this wrong means either burning budget on always-on consumers or missing the SLA when a product launch drives ingest from 50 to 5,000 messages per second. By the end of this lesson you'll be able to wire a Kafka consumer to Argo Workflow submission and configure a KEDA ScaledObject that scales the consumer fleet from zero to N replicas based on consumer-group lag.

Key Terminology

• KEDA (Kubernetes Event-Driven Autoscaling) — a Kubernetes add-on that scales workloads based on external event-source metrics (Kafka lag, queue depth, etc.); the engine that gives this pipeline its scale-to-zero property.
• Consumer group lag — the count of unprocessed messages across partitions assigned to a Kafka consumer group; KEDA reads this number to decide how many consumer pods to run.
• confluent_kafka — the production-grade Python Kafka client used here to read events with manual offset commits, the foundation for at-least-once processing semantics.
• lagThreshold — KEDA Kafka-scaler setting that defines target messages per replica; set it to match your batch size so each pod processes a meaningful workload before another replica is added.
• cooldownPeriod — KEDA setting controlling how long lag must stay below threshold before scaling down; prevents oscillation during bursty traffic.

Concepts

Event-driven trigger architecture

The pipeline has three layers: a Kafka topic receives document.created events from upstream producers; a Python consumer reads events in batches and submits an Argo Workflow per batch; KEDA monitors consumer-group lag and adjusts replica count to keep lag below a target threshold. The consumer never polls on a schedule — it wakes up only when messages exist, and KEDA shuts it down when the topic drains (see Code Walkthrough).

Scale-to-zero and lag-driven scaling

KEDA's Kafka scaler queries consumer-group lag and divides it by lagThreshold to compute desired replica count, capped by maxReplicaCount. When lag is zero, replicas drop to zero; the next message triggers a cold start of one pod, and load grows from there. Set lagThreshold equal to your batch size (e.g. 50) so each pod processes a full batch before another replica is justified, and set cooldownPeriod to at least 60 seconds to prevent flap during bursty traffic.

At-least-once delivery via manual offset commits

Disable Kafka auto-commit and commit offsets only after the Argo submission succeeds. If the consumer crashes between reading and submitting, the unacknowledged messages are redelivered on restart — the workflow may run twice for a batch, but no document is lost. Workflow-layer idempotency (dedupe by document_id) handles the duplicate case.

Code Walkthrough

Now that you have seen the concepts above, the walkthrough below turns them into working code.

The snippet below demonstrates the three concepts together: it builds a confluent_kafka consumer with manual commits, reads a batch, submits an Argo Workflow, and commits offsets only on success. KEDA scaling is configured separately via a ScaledObject manifest that references the same group.id.

Code snippetpython
1from confluent_kafka import Consumer, KafkaError
2import json, logging, os, requests
3
4logger = logging.getLogger(__name__)
5
6def create_consumer(group_id: str = "pipeline-trigger") -> Consumer:
7    return Consumer({
8        "bootstrap.servers": os.environ["KAFKA_BROKERS"],
9        "group.id": group_id,
10        "auto.offset.reset": "earliest",
11        "enable.auto.commit": False,
12        "max.poll.interval.ms": 300000,
13        "session.timeout.ms": 45000,
14    })
15
16def consume_and_trigger(
17    consumer: Consumer,
18    topic: str = "document-events",
19    batch_size: int = 50,
20    argo_api: str = "http://argo-server:2746",
21) -> int:
22    consumer.subscribe([topic])
23    messages = consumer.consume(num_messages=batch_size, timeout=10.0)
24
25    valid_docs = []
26    for msg in messages:
27        if msg.error():
28            if msg.error().code() != KafkaError._PARTITION_EOF:
29                logger.error("Kafka error: %s", msg.error())
30            continue
31        event = json.loads(msg.value().decode("utf-8"))
32        if event.get("event_type") == "document.created":
33            valid_docs.append(event["document_id"])
34
35    if not valid_docs:
36        return 0
37
38    resp = requests.post(
39        f"{argo_api}/api/v1/workflows/pipelines",
40        json={"workflow": build_batch_workflow(valid_docs)},
41        timeout=30,
42    )
43    resp.raise_for_status()
44    consumer.commit(asynchronous=False)
45    logger.info("Triggered workflow for %d documents", len(valid_docs))
46    return len(valid_docs)

enable.auto.commit=False is the lynchpin of at-least-once semantics: the consumer.commit() call after raise_for_status() is the only place offsets advance. max.poll.interval.ms=300000 gives the Argo POST headroom without triggering a consumer-group rebalance. Filtering document.created events explicitly keeps update/delete events on the same topic from fanning out unnecessary workflows.

You'll know it works when, with KEDA installed and a ScaledObject pointing at group.id=pipeline-trigger with lagThreshold: 50, producing 5,000 events causes pod replicas to climb proportionally and an Argo Workflow appears in argo list for each batch; producing zero events for cooldownPeriod seconds drives replicas back to zero.

Do's and Don'ts

Now that you have worked through the implementation, the practices below separate a durable approach from a fragile one.

Do's

1. ✓Do disable Kafka auto-commit and commit after Argo submission succeeds — guarantees at-least-once delivery so a consumer crash never silently drops a document batch.
2. ✓Do set lagThreshold to your batch size — keeps each pod processing a meaningful batch and prevents fragmentation into many under-utilized replicas.
3. ✓Do filter event types in the consumer before submitting workflows — avoids triggering pipeline runs for irrelevant events (updates, deletes) sharing the topic.

Don'ts

1. ✗Don't set cooldownPeriod below 60 seconds — short cooldowns cause scale-up/scale-down oscillation during bursty traffic and thrash the Kubernetes scheduler.
2. ✗Don't leave max.poll.interval.ms at the default — the default is too short for synchronous Argo API calls; the consumer group will rebalance mid-batch and lose partition ownership.
3. ✗Don't run KEDA with an unbounded maxReplicaCount — a poison-pill producer can blow up the cluster; cap replicas to a known cost ceiling.