Event-Driven AI: How Kafka Streams Changed the Way We Run LLMs in Production

Last October our AI service handled ~50K LLM requests per day. By February that number hit 800K. By April, 2 million. The synchronous HTTP architecture we had shipped in three weeks didn't survive contact with that growth. We were dropping requests, hitting rate limits that cascaded into total outages, and on-call engineers were getting paged at 3AM for timeouts that shouldn't have been possible to timeout.

The fix wasn't bigger timeouts or faster hardware. It was rethinking the execution model entirely — from request-response to event-driven. Here's the full architecture, the Spring AI + Kafka code, the failure modes we hit, and the production numbers that came out the other side.

Why Synchronous LLM Calls Break at Scale

The typical AI feature looks like this: a user does something, your API calls an LLM, the LLM responds, you return the result. Simple, works fine in demos, blows up in production.

LLMs are slow and variable. A fast response is 300ms. A slow one — complex reasoning, long context, throttled provider — is 15–30 seconds. Most HTTP infrastructure timeouts sit at 30 seconds, which means you're living on the edge constantly.

Rate limits cascade. Your provider gives you X requests per minute. Under load, one failed batch causes retries that burn more quota, which causes more failures. Exponential backoff helps but doesn't fix the underlying problem: too many concurrent callers sharing a quota.

No buffering. If the downstream LLM is temporarily rate-limited or degraded, you have no place to hold work. You either drop the request or return an error to the user.

The Event-Driven AI Model

The core idea: instead of making LLM calls synchronously in the API handler, publish work to a Kafka topic, consume it in a dedicated AI worker, and emit results to a results topic. Callers subscribe to results.

The client immediately gets a 202 Accepted with a task ID. The result arrives when the LLM call completes — via WebSocket, SSE, or polling. This separation is what unlocks scale: the HTTP layer and the LLM processing layer evolve independently and fail independently.

[!NOTE] This is the same pattern as async email sending, async PDF generation, or async image processing. It's been standard for I/O-heavy work for a decade. LLMs are just I/O with a 10-second P99.

Topic Design: What Goes Where

Getting the Kafka topic structure right matters more than any individual line of code. Topics are your API contract — they're hard to change once services depend on them.

Design principles we landed on:

One topic per task type, not one monolithic ai-tasks topic. Lets you scale consumers independently, set different retention policies (real-time classification results don't need 7-day retention; completed summarizations do), and gives you clear consumer lag metrics per workload.

Priority tiers as separate consumer groups, not topics. Eight consumer threads watch the high-priority topics; two watch the batch topics. Simple to operate — no complex priority queue logic or message reordering.

Separate DLQ per task type (ai.tasks.classify.dlq, not a shared DLQ). When debugging production failures you already know the task type — a shared DLQ forces you to filter through noise from every other failing type simultaneously.

Results topic as a public contract. Any service can subscribe to ai.results.completed and react to completed AI tasks. Loose coupling at its best — the AI worker doesn't need to know who cares about the results.

@Configuration
public class AITopicConfig {

    public static final String TASKS_SUMMARIZE   = "ai.tasks.summarize";
    public static final String TASKS_CLASSIFY    = "ai.tasks.classify";
    public static final String RESULTS_COMPLETED = "ai.results.completed";
    public static final String RESULTS_FAILED    = "ai.results.failed";
    public static final String DLQ_CLASSIFY      = "ai.tasks.classify.dlq";

    @Bean
    public NewTopic summarizeTopic() {
        return TopicBuilder.name(TASKS_SUMMARIZE)
                .partitions(12)   // max 12 concurrent consumers in this group
                .replicas(3)
                .config(TopicConfig.RETENTION_MS_CONFIG, "604800000") // 7 days
                .build();
    }

    @Bean
    public NewTopic classifyTopic() {
        return TopicBuilder.name(TASKS_CLASSIFY)
                .partitions(24)   // higher throughput — more partitions
                .replicas(3)
                .config(TopicConfig.RETENTION_MS_CONFIG, "86400000")  // 1 day
                .build();
    }

    @Bean
    public NewTopic resultsTopic() {
        return TopicBuilder.name(RESULTS_COMPLETED)
                .partitions(12)
                .replicas(3)
                .config(TopicConfig.RETENTION_MS_CONFIG, "2592000000") // 30 days
                .build();
    }
}

Partition count is a ceiling on parallelism. Kafka consumer group parallelism is bounded by partition count — 12 partitions means you can run at most 12 consumer threads per group before threads sit idle. Size for peak throughput, not current throughput. You can't easily add partitions to a live topic without a migration.

The AI Task Schema

Every message uses the same envelope. Consistency is what lets you build generic retry, routing, and observability tooling without per-task-type special cases.

public record AITask(
    String taskId,          // UUID — idempotency key
    String taskType,        // "summarize" | "classify" | "extract"
    String requesterId,     // service or user that submitted
    String priority,        // "high" | "normal" | "batch"
    Instant submittedAt,
    int    attemptNumber,   // 1-based; consumers use this for backoff strategy
    Map<String, Object> payload,   // task-specific input data
    Map<String, String> metadata   // tracing headers, correlation IDs
) {}

public record AIResult(
    String  taskId,
    String  taskType,
    boolean success,
    Instant completedAt,
    long    processingMs,
    String  model,
    int     inputTokens,
    int     outputTokens,
    Object  output,          // task-specific result — Map, String, etc.
    String  errorMessage     // null on success
) {}

Three design decisions worth calling out:

taskId is the idempotency key. Consumers check whether a task has already been processed before calling the LLM. Kafka at-least-once delivery means you will receive duplicates — consumer rebalancing, network glitches, pod restarts. Plan for it from day one.

attemptNumber travels with the message, not just as a consumer variable. This means retry strategy is encoded in the event itself: first attempt = immediate retry, second = 30s backoff, third = DLQ. Works correctly across pod restarts and consumer rebalancing.

Token counts in every result. Essential for cost tracking. Sum inputTokens + outputTokens per consumer group, per model, per hour. That's your cost dashboard with no extra instrumentation.

Spring AI + Kafka: The Consumer

This is the heart of the system — the consumer that actually processes AI tasks.

@Component
@Slf4j
public class AITaskConsumer {

    private final ChatClient         chatClient;
    private final KafkaTemplate<String, AIResult> resultTemplate;
    private final AITaskRepository   taskRepository;   // idempotency store
    private final MeterRegistry      meterRegistry;
    private final AdaptiveRateLimiter rateLimiter;

    @KafkaListener(
        topics       = { AITopicConfig.TASKS_SUMMARIZE, AITopicConfig.TASKS_CLASSIFY },
        groupId      = "ai-worker-high-priority",
        concurrency  = "8",       // 8 consumer threads per pod
        containerFactory = "aiListenerContainerFactory"
    )
    public void consume(
            @Payload AITask task,
            @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
            Acknowledgment ack) {

        log.info("task={} type={} attempt={}", task.taskId(), task.taskType(), task.attemptNumber());

        // Check idempotency — previous consumer may have already handled this
        if (taskRepository.isCompleted(task.taskId())) {
            log.debug("task={} already completed — skipping (idempotent ack)", task.taskId());
            ack.acknowledge();
            return;
        }

        Timer.Sample timer = Timer.start(meterRegistry);

        try {
            rateLimiter.acquire(); // blocks until a permit is available

            AIResult result = switch (task.taskType()) {
                case "summarize" -> handleSummarize(task);
                case "classify"  -> handleClassify(task);
                default -> throw new IllegalArgumentException("Unknown type: " + task.taskType());
            };

            taskRepository.markComplete(task.taskId(), result); // write idempotency record first
            resultTemplate.send(RESULTS_COMPLETED, task.taskId(), result);
            ack.acknowledge();  // only after result is durably published

            rateLimiter.onSuccess();
            timer.stop(meterRegistry.timer("ai.task.processing",
                    "type", task.taskType(), "status", "success"));

        } catch (RateLimitException e) {
            // Provider returned 429 — don't ack, let Kafka redeliver after consumer pause
            rateLimiter.onRateLimit();
            log.warn("task={} rate-limited (attempt={})", task.taskId(), task.attemptNumber());
            // Intentionally no ack.acknowledge() — message redelivers

        } catch (Exception e) {
            log.error("task={} failed attempt={}: {}", task.taskId(), task.attemptNumber(), e.getMessage());
            if (task.attemptNumber() >= 3) {
                sendToDLQ(task, e.getMessage());
                ack.acknowledge(); // remove from main topic — it's in DLQ now
            }
            // Otherwise no ack — Kafka redelivers with same attempt number
            // (producer should increment attemptNumber on retry, or use a retry topic)
            timer.stop(meterRegistry.timer("ai.task.processing",
                    "type", task.taskType(), "status", "error"));
        }
    }

    private AIResult handleSummarize(AITask task) {
        String content  = (String) task.payload().get("content");
        int    maxWords = (int)    task.payload().getOrDefault("maxWords", 150);

        String prompt = """
            Summarize the following in under %d words.
            Be specific — preserve key facts, numbers, and named entities.
            
            Content:
            %s
            """.formatted(maxWords, content);

        Instant start    = Instant.now();
        ChatResponse resp = chatClient.prompt().user(prompt).call().chatResponse();
        Usage usage      = resp.getMetadata().getUsage();

        return new AIResult(
            task.taskId(), "summarize", true, Instant.now(),
            Duration.between(start, Instant.now()).toMillis(),
            "llama-3.1-8b-instant",
            (int) usage.getPromptTokens(), (int) usage.getGenerationTokens(),
            Map.of("summary", resp.getResult().getOutput().getContent()),
            null
        );
    }
}

The ack.acknowledge() placement is load-bearing. We only call it after the result is successfully published to the results topic. If the result publish fails — network blip, Kafka broker restart — Kafka will redeliver the task to another consumer thread. Combined with the idempotency check at the top, the task will be processed exactly once even across failures.

Rate limiting happens inside the consumer, not at the API gateway. Your LLM provider doesn't know about your queue depth — it just sees requests arriving. Rate limiting at the consumer level means you never exceed quota regardless of how deep the queue gets.

Handling Backpressure: Adaptive Rate Limiting

The most common failure mode isn't Kafka — it's the LLM provider. When you hit rate limits, the naive response (throw exception, retry immediately) makes things significantly worse. Here's what we do instead:

@Component
public class AdaptiveRateLimiter {

    // Groq's limit at our tier: 30 req/min = 0.5 req/sec
    private final double            maxRate = 0.45; // 10% headroom below actual limit
    private final double            minRate = 0.05; // hard floor when throttled
    private final AtomicDouble  currentRate = new AtomicDouble(0.3);
    private volatile RateLimiter     limiter = RateLimiter.create(0.3);

    public void acquire() { limiter.acquire(); }

    public void onSuccess() {
        double next = Math.min(currentRate.get() * 1.05, maxRate); // +5% on success
        updateLimiter(next);
    }

    public void onRateLimit() {
        double next = Math.max(currentRate.get() * 0.5, minRate); // halve on 429
        updateLimiter(next);
        log.warn("Rate-limited — throttling to {:.2f} req/sec", next);
    }

    private void updateLimiter(double newRate) {
        if (Math.abs(newRate - currentRate.get()) > 0.01) {
            currentRate.set(newRate);
            limiter = RateLimiter.create(newRate);
        }
    }
}

This is AIMD — Additive Increase, Multiplicative Decrease — the same algorithm TCP congestion control uses. On success, increase by 5%. On rate limit, halve. Over time it converges on the sustainable throughput for your provider tier without manual tuning.

Dead Letter Queues: Don't Silently Drop AI Work

A DLQ is your safety net for permanent failures. Every task that fails three times ends up there — unprocessed, with the error reason attached as metadata.

private void sendToDLQ(AITask task, String errorMessage) {
    Map<String, String> dlqMeta = new HashMap<>(task.metadata());
    dlqMeta.put("dlq.reason",       errorMessage);
    dlqMeta.put("dlq.sentAt",       Instant.now().toString());
    dlqMeta.put("dlq.finalAttempt", String.valueOf(task.attemptNumber()));

    AITask dlqTask = new AITask(
        task.taskId(), task.taskType(), task.requesterId(), task.priority(),
        task.submittedAt(), task.attemptNumber(), task.payload(), dlqMeta
    );

    String dlqTopic = "ai.tasks." + task.taskType() + ".dlq";
    dlqTemplate.send(dlqTopic, task.taskId(), dlqTask);

    log.error("task={} → DLQ after {} attempts: {}", task.taskId(), task.attemptNumber(), errorMessage);
    meterRegistry.counter("ai.task.dlq", "type", task.taskType()).increment();
}

DLQ messages are retained for 30 days. Twice a week we run a DLQ reprocessing job: identify the root cause, fix the consumer, then replay DLQ messages back into the main topic. Nothing is permanently lost.

[!WARNING] Do not set DLQ retention too short. A production incident that takes 3 days to diagnose needs the evidence to still be in the DLQ when you're ready to replay. 30 days is a safe default for AI task DLQs.

Exactly-Once Semantics: When You Actually Need Them

Kafka supports exactly-once semantics (EOS) via transactions. For most AI tasks, you don't need them — idempotent consumers plus at-least-once delivery is simpler and sufficient. But for billing events, audit logs, and compliance-critical document processing, true exactly-once matters.

@Bean
public KafkaTransactionManager<String, AIResult> kafkaTxManager(
        ProducerFactory<String, AIResult> pf) {
    return new KafkaTransactionManager<>(pf);
}

// Consumer using Kafka transactions
@Transactional("kafkaTxManager")
@KafkaListener(topics = "ai.tasks.billing-critical", groupId = "ai-billing")
public void consumeExactlyOnce(@Payload AITask task, Acknowledgment ack) {
    // Entire method runs in a Kafka transaction.
    // Either ALL of the following happen atomically, or NONE do:
    //   1. Process the AI task
    //   2. Publish the result event
    //   3. Commit the consumer offset
    AIResult result = processTask(task);
    resultTemplate.send(RESULTS_COMPLETED, task.taskId(), result);
    // offset commits atomically with the produce — true exactly-once
}

The tradeoff: EOS adds roughly 15–20% latency overhead due to transaction coordination. For billing tasks, worth it. For bulk document summarization, not worth it — idempotent at-least-once is the right choice there.

Observability: Tracing AI Events Across Services

An AI task might be submitted by Service A, consumed by AI Worker B, and the result consumed by Service C. When something goes wrong, you need to trace the full journey without manual correlation.

We propagate OpenTelemetry trace context through Kafka headers:

// Producer — inject trace context into message headers
public void submitTask(AITask task) {
    ProducerRecord<String, AITask> record = new ProducerRecord<>(
            TASKS_SUMMARIZE, task.taskId(), task);

    GlobalOpenTelemetry.getPropagators()
            .getTextMapPropagator()
            .inject(Context.current(), record.headers(),
                    (headers, key, val) -> headers.add(key, val.getBytes(UTF_8)));

    kafkaTemplate.send(record);
}

// Consumer — extract trace context and create a child span
@KafkaListener(...)
public void consume(@Payload AITask task, @Headers MessageHeaders headers) {

    Context parentCtx = GlobalOpenTelemetry.getPropagators()
            .getTextMapPropagator()
            .extract(Context.current(), headers, new MessageHeadersGetter());

    Span span = tracer.spanBuilder("ai.task.process")
            .setParent(parentCtx)
            .setAttribute("task.id",    task.taskId())
            .setAttribute("task.type",  task.taskType())
            .startSpan();

    try (Scope ignored = span.makeCurrent()) {
        AIResult result = processTask(task);
        span.setAttribute("tokens.input",  result.inputTokens());
        span.setAttribute("tokens.output", result.outputTokens());
        span.setAttribute("latency.ms",    result.processingMs());
    } finally {
        span.end();
    }
}

In Grafana Tempo, the complete trace now shows: HTTP POST /enrich → Kafka publish → ai.task.process (the LLM call spans inside) → Kafka produce result → result consumer → database write. The entire 8-second LLM call is one contiguous trace chain.

Key metrics per consumer group:

Configuration That Actually Matters

spring:
  kafka:
    bootstrap-servers: ${KAFKA_BROKERS}
    consumer:
      group-id: ai-worker-high-priority
      auto-offset-reset: earliest
      enable-auto-commit: false          # CRITICAL — never use auto-commit for AI tasks
      max-poll-records: 5                # Don't overwhelm a slow consumer
      max-poll-interval-ms: 120000       # 2 min — LLM calls can be slow
      fetch-min-bytes: 1
    producer:
      acks: all                          # Wait for all in-sync replicas
      retries: 3
      enable-idempotence: true
    listener:
      ack-mode: MANUAL_IMMEDIATE
      concurrency: 8

# Tune these per your LLM provider tier
ai:
  rate-limiter:
    max-rate-per-second: 0.45           # 27 req/min — 10% below Groq's 30 req/min limit
    min-rate-per-second: 0.05
  retry:
    max-attempts: 3
    dlq-retention-days: 30

max-poll-interval-ms is the one that bites everyone. If a consumer takes longer than this value between poll() calls, Kafka considers it dead and triggers a rebalance — even if the consumer is actively processing a long LLM call. We set it to 120 seconds (double our P99 LLM latency) to avoid unnecessary rebalancing during legitimate slow calls.

max-poll-records: 5 limits how many messages a single consumer thread fetches in one batch. With 8 threads each holding 5 messages, you have 40 LLM calls potentially in-flight per pod. Size this to your rate limit headroom.

The Cost Breakdown

After 90 days running this architecture against ~2M requests/day:

The event-driven architecture contributed two cost wins that weren't possible with synchronous calls:

Intelligent model routing. The consumer reads task.payload().get("complexity") — a simple score computed at task submission — and routes simple classifications to llama-3.1-8b-instant instead of a 70B model. This cut our average cost-per-task by 31%.

Micro-batching. For task types where the model supports batch input, we accumulate messages in 100ms windows and send them as a single request. 40% reduction in per-call API overhead (connection setup, auth handshake, response parsing). Impossible to do cleanly in a synchronous request handler without blocking the caller.

[!NOTE] $687/day sounds large. At 2M requests that's $0.00034 per request — roughly $20K/month. Running equivalent workloads on self-hosted GPU infrastructure (A100s, realistic utilization rates, ops overhead) would cost us approximately $2,100/day. The managed API is cheaper at our scale.

Before and After: Production Numbers

We ran both architectures in parallel for 30 days — old sync on 20% of traffic, new event-driven on 80% — before cutting over completely.

The P99 number matters most. Users hitting a 28-second timeout perceived the feature as broken, even if only 3% of requests reached that threshold. With async dispatch and WebSocket delivery, P50 user-visible latency is under 1 second (the 202 response arrives instantly) and the result appears as a push notification when the LLM finishes.

The 93% reduction in on-call pages was unexpected. Most of the 14 weekly pages were timeout-related cascades. When you remove synchronous coupling between the API layer and the LLM layer, most cascade failure modes disappear entirely.

When NOT to Go Event-Driven

This architecture adds real operational complexity. It's not always the right answer.

Stay synchronous when:

• The result must block the response. Real-time payment validation, auth decisions, interactive AI chat where the user is watching a streaming response. If you can't send a 202, you need sync.

• Volume is under ~10K requests/day. The operational overhead of Kafka, consumer groups, DLQ management, and result delivery infrastructure doesn't pay off at low volume. A simple async HTTP call with a 30-second timeout and a Postgres-backed retry queue is the right choice.

• You don't have result delivery infrastructure. WebSockets, SSE, or polling endpoints are meaningful engineering work. If you're not willing to build that, your users will stare at a spinner indefinitely. "We'll add the result delivery later" becomes "we added Kafka but the feature still feels broken."

• Your team is new to Kafka. Consumer group rebalancing, partition assignment, exactly-once transactions, DLQ management — these are non-trivial operational concerns. Kafka is operationally complex when something goes wrong. Make sure your team has the skills before you take the dependency.

The decision framework:

The Minimum Viable Starting Point

If you're starting fresh and want to validate the pattern before building the full architecture, here's the minimum working setup:

# application.yml — minimum for correctness
spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: ai-worker
      auto-offset-reset: earliest
      enable-auto-commit: false      # never auto-commit
      max-poll-interval-ms: 90000
    listener:
      ack-mode: MANUAL_IMMEDIATE

// Minimum viable consumer — add rate limiting, DLQ, and idempotency incrementally
@KafkaListener(topics = "ai.tasks", groupId = "ai-worker")
public void consume(@Payload AITask task, Acknowledgment ack) {
    try {
        AIResult result = callLLM(task);
        resultTemplate.send("ai.results", task.taskId(), result);
        ack.acknowledge();
    } catch (Exception e) {
        log.error("Task {} failed: {}", task.taskId(), e.getMessage());
        // Don't ack — Kafka redelivers
        // Add DLQ logic here after you've observed retry behavior in staging
    }
}

Start here. Add the adaptive rate limiter when you first see 429s in production. Add DLQs when you see messages redelivered indefinitely. Add idempotency when you see duplicate results. Add distributed tracing when debugging becomes painful. The architecture I described above is 18 months of production failures distilled — you don't need all of it on day one.

Key Takeaways

• Synchronous LLM calls are fine until they're not. The failure mode is sudden and cascading — build the escape hatch before you need it. The warning signs are rising P99 latency and rate limit errors.

• Kafka absorbs traffic spikes that would kill a synchronous system. The queue is the buffer between your traffic patterns and your provider's rate limits. You can't add that buffer to a sync architecture without fundamentally restructuring it.

• Idempotency is non-negotiable. At-least-once delivery is guaranteed by Kafka; idempotent consumers are your responsibility. An idempotency check at the top of every consumer is not optional overhead — it's correctness.

• Rate limiting belongs in the consumer, not the gateway. Only the consumer knows what's actually in flight against the LLM provider. Gateway-level rate limiting gives you no visibility into queue depth.

• Token counts in every result event. You cannot optimize what you cannot see. Per-model, per-task-type token counts are the only way to know which workloads are driving your LLM bill.

• DLQs are not optional. Every permanently failed task needs a destination. "Log it and move on" means you've silently dropped user work. DLQ + replay gives you a recovery path that sync architectures simply don't have.

• The biggest win isn't latency — it's reliability. P99 improvements are impressive. Getting paged 93% less often at 3AM is what actually changes how your team feels about owning the service.

The investment is real — probably 6–8 weeks of solid engineering to get this fully operational. In our case, it was worth it the first Saturday night we didn't get paged.

The next post in this series covers stateful aggregation with Kafka Streams — specifically windowed token-count aggregations for real-time cost alerts and per-tenant quota enforcement. If you're hitting a different failure mode than I described here, or building something similar, I'd love to hear about it.