I Rebuilt Our AI Agents on Kafka Instead of REST — Here's the Full Java Architecture

Key every message by conversationId, not randomly. Kafka guarantees ordering within a partition, and a partition is chosen by key. If two events for the same conversation land on different partitions, they can be processed out of order — an agent might see a tool result before it sees the task that requested it. Keying by conversationId keeps everything for one user session in order, on one partition, consumed by one thread.

Separate the reasoning trace from the task queue. agent.tasks.* carries "do this next." A parallel agent.thoughts.* topic (or a single agent.thoughts topic keyed the same way) carries every intermediate reasoning step the agent produced — this is what turns into your audit trail and replay capability. Don't conflate "what to do" with "what happened."

Give tool calls their own request/response topic pair, exactly like the task topics. agent.tool-calls.requested and agent.tool-calls.completed decouple tool execution from the agent that requested it — the tool executor doesn't know or care which agent asked, it just processes requests and publishes results.

@Configuration
public class AgentTopicConfig {

    public static final String TASKS_PLAN        = "agent.tasks.plan";
    public static final String TASKS_RESEARCH     = "agent.tasks.research";
    public static final String TASKS_CODE         = "agent.tasks.code";
    public static final String THOUGHTS           = "agent.thoughts";
    public static final String TOOL_REQUESTED     = "agent.tool-calls.requested";
    public static final String TOOL_COMPLETED     = "agent.tool-calls.completed";
    public static final String APPROVALS_PENDING  = "agent.approvals.pending";
    public static final String APPROVALS_DECIDED  = "agent.approvals.decided";
    public static final String RESULTS_COMPLETED  = "agent.results.completed";
    public static final String RESULTS_DLQ        = "agent.results.dlq";

    @Bean
    public NewTopic tasksResearch() {
        return TopicBuilder.name(TASKS_RESEARCH)
                .partitions(12)
                .replicas(3)
                .config(TopicConfig.RETENTION_MS_CONFIG, "1209600000") // 14 days — long enough to replay a bad run
                .build();
    }

    @Bean
    public NewTopic thoughts() {
        return TopicBuilder.name(THOUGHTS)
                .partitions(12)
                .replicas(3)
                .config(TopicConfig.RETENTION_MS_CONFIG, "2592000000") // 30 days — this is your audit log
                .build();
    }

    @Bean
    public NewTopic toolCallsRequested() {
        return TopicBuilder.name(TOOL_REQUESTED)
                .partitions(24)   // tool calls are higher volume than tasks
                .replicas(3)
                .build();
    }
}

Each Agent Is a Consumer Loop Running ReAct

Strip away the Kafka plumbing and every agent is doing the same thing: consume a task, reason about it (optionally calling tools), and publish a result. The ReAct loop — reason, act, observe, repeat — maps directly onto a state machine, and Kafka events are exactly what drive state transitions.

The important detail: the agent's "state" is just its position in this loop, and every transition is caused by consuming a Kafka message. That means an agent worker can crash mid-reasoning and a replacement instance in the same consumer group picks up exactly where it left off — the last message it hadn't yet acknowledged gets redelivered. You don't need to build your own state-recovery mechanism; consumer offsets already are one.

@Component
public class ResearchAgentWorker {

    private final ChatClient chatClient;      // Spring AI
    private final KafkaTemplate<String, Object> kafka;
    private final Set<String> seenTaskIds = ConcurrentHashMap.newKeySet();
    private final ConcurrentHashMap<String, AgentTask> pending = new ConcurrentHashMap<>();

    public ResearchAgentWorker(ChatClient.Builder chatClientBuilder, KafkaTemplate<String, Object> kafka) {
        this.chatClient = chatClientBuilder.build();
        this.kafka = kafka;
    }

    @KafkaListener(topics = AgentTopicConfig.TASKS_RESEARCH, groupId = "research-agent")
    public void onTask(AgentTask task, @Header(KafkaHeaders.RECEIVED_KEY) String conversationId) {

        // Idempotency check — at-least-once delivery means this can run twice.
        // A real deployment backs this set with Redis (TTL'd to topic retention)
        // instead of an in-process set, so it survives a worker restart.
        if (!seenTaskIds.add(task.taskId())) return;

        pending.put(task.taskId(), task);
        AgentThought thought = reason(task);
        kafka.send(AgentTopicConfig.THOUGHTS, conversationId, thought);

        if (thought.needsTool()) {
            ToolCallRequested request = new ToolCallRequested(
                task.taskId(), conversationId, thought.toolName(), thought.toolArgs());
            kafka.send(AgentTopicConfig.TOOL_REQUESTED, conversationId, request);
            // This worker's job for this message ends here — the tool result
            // arrives as a new message on agent.tool-calls.completed and
            // resumes the loop. We do NOT block this thread waiting for it.
            return;
        }

        AgentResult result = new AgentResult(task.taskId(), conversationId, thought.finalAnswer());
        kafka.send(AgentTopicConfig.RESULTS_COMPLETED, conversationId, result);
        pending.remove(task.taskId());
    }

    @KafkaListener(topics = AgentTopicConfig.TOOL_COMPLETED, groupId = "research-agent")
    public void onToolResult(ToolCallCompleted result, @Header(KafkaHeaders.RECEIVED_KEY) String conversationId) {
        // Resume reasoning with the tool's output folded into context
        AgentTask original = pending.get(result.taskId());
        if (original == null) return; // not ours / already handled
        seenTaskIds.remove(result.taskId()); // let the continuation back through the idempotency gate
        onTask(original.withToolResult(result), conversationId);
    }

    private AgentThought reason(AgentTask task) {
        return chatClient.prompt()
            .system(task.systemPrompt())
            .user(task.renderedContext())
            .call()
            .entity(AgentThought.class);
    }
}

The line worth pausing on is the early return after publishing a tool-call request. The consumer thread does not block waiting for the tool result. It commits the offset for the task message and moves on — Kafka's consumer group can pick up the next task immediately. When the tool result arrives, it's a new message that resumes this conversation from wherever it left off. This is what makes a Kafka-driven agent fundamentally more scalable than one that blocks a thread per in-flight tool call: your thread pool size stops being coupled to how many agent conversations can be in-flight at once.

One Full Task, Start to Finish

Here's what actually crosses the wire (well — the topics) for one user request that needs a tool call and triggers human approval before completing:

Notice what's not in this diagram: no synchronous call ever blocks waiting on the human reviewer. The Code Agent's consumer thread is free the instant it publishes to agent.approvals.pending — it picks up other conversations while this one waits, however long the human takes. That's the pause point REST couldn't give us cleanly.

And because every one of those arrows is a persisted, offset-addressable Kafka message, we can replay this exact sequence for any conversation ID, months later, to answer "why did the agent decide to touch the payment module" — which is the question that actually gets asked after something goes wrong.

Failure Handling: Idempotency and Dead Letters

At-least-once delivery is the Kafka default, which means every consumer above must tolerate processing the same message twice. Two things make that safe:

Idempotency keys on every task and tool call. Each AgentTask and ToolCallRequested carries a stable taskId. Before doing real work, the consumer checks whether that ID was already processed (a small Redis or Postgres lookup, TTL'd to the topic's retention window) and short-circuits if so.

A DLQ per failure-prone step, not a shared one. A tool call that throws after three retries goes to agent.tool-calls.dlq with the original request and the exception, not into a shared catch-all — when you're triaging failures at 9am you already know which stage failed and don't want to filter through unrelated noise.

@Component
public class ToolExecutorWorker {

    private final KafkaTemplate<String, Object> kafka;
    private final ToolRegistry toolRegistry;

    public ToolExecutorWorker(KafkaTemplate<String, Object> kafka, ToolRegistry toolRegistry) {
        this.kafka = kafka;
        this.toolRegistry = toolRegistry;
    }

    @RetryableTopic(
        attempts = "3",
        backoff = @Backoff(delay = 2000, multiplier = 2),
        dltTopicSuffix = ".dlq"
    )
    @KafkaListener(topics = AgentTopicConfig.TOOL_REQUESTED, groupId = "tool-executor")
    public void onToolCall(ToolCallRequested request, @Header(KafkaHeaders.RECEIVED_KEY) String conversationId) {
        Object output = toolRegistry.execute(request.toolName(), request.args());
        kafka.send(AgentTopicConfig.TOOL_COMPLETED, conversationId,
            new ToolCallCompleted(request.taskId(), conversationId, request.toolName(), output));
    }

    @DltHandler
    public void onToolCallFailed(ToolCallRequested request, Exception ex) {
        System.err.println("Tool call permanently failed: " + request.toolName() + " — " + ex.getMessage());
    }
}

Spring Kafka's @RetryableTopic handles the retry-with-backoff and the DLQ routing declaratively — you don't hand-roll retry loops per agent.

The Free Audit Trail Nobody Asked For, But Everybody Needed

This was the part we didn't plan for and ended up mattering most. Because every reasoning step lives on agent.thoughts, keyed by conversation ID, with 30-day retention, we can reconstruct the entire decision trail of any agent conversation just by re-consuming that topic from offset zero for a given key. No custom logging framework, no separate tracing system bolted on after the fact — the event log is the trace.

When a reviewer asks "why did the Code Agent think it needed to touch three files instead of one," the answer isn't in an APM dashboard, it's three Kafka messages, in order, with the LLM's own stated reasoning attached to each. That's a fundamentally different debugging experience than grepping six services' logs for a correlation ID that may or may not have propagated correctly.

When This Is Overkill

A single agent doing single-turn conversational Q&A — a support chatbot, a one-shot summarizer — does not need any of this. Kafka adds latency (message round-trip instead of a direct call), operational surface area (topics, consumer groups, partition counts to tune), and a genuinely harder debugging model for the simple cases where nothing is going wrong. If your agent count is one and your task doesn't branch, a direct LLM call in a normal request handler is not just simpler — it's correct. We reached for Kafka only once we had multiple agent types that needed to hand work to each other, needed independent scaling, and needed a pause point for human review. If you're not there yet, don't build for it preemptively — you can migrate the same way we did, one agent type at a time, starting with whichever pair of agents is currently the most tightly coupled.

Proof, Not Promises: Every Snippet Above Actually Compiles

Architecture posts are full of code that looks right and was never run. I didn't want this to be one of them, so before publishing I pulled every class above into a real Maven project — the actual AgentEvents records, AgentTopicConfig, ResearchAgentWorker, and ToolExecutorWorker — and compiled it against the real dependencies, not stubs.

The pom.xml — Spring Boot 3.3.5, real spring-kafka, real spring-ai-starter-model-openai (pulled from Maven Central + the Spring Milestones repo), targeting Java 21:

<parent>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-parent</artifactId>
  <version>3.3.5</version>
</parent>

<properties>
  <maven.compiler.release>21</maven.compiler.release>
  <spring-ai.version>1.0.0</spring-ai.version>
</properties>

<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>org.springframework.ai</groupId>
      <artifactId>spring-ai-bom</artifactId>
      <version>${spring-ai.version}</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>

<dependencies>
  <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency>
  <dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>
  <dependency><groupId>org.springframework.retry</groupId><artifactId>spring-retry</artifactId></dependency>
  <dependency><groupId>org.springframework.ai</groupId><artifactId>spring-ai-starter-model-openai</artifactId></dependency>
</dependencies>

The command:

mvn clean compile

The real output:

[INFO] Compiling 6 source files with javac [debug parameters release 21] to target\classes
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------

Every annotation used above — @KafkaListener, @Header(KafkaHeaders.RECEIVED_KEY), TopicBuilder, @RetryableTopic, @DltHandler, and Spring AI's chatClient.prompt().system().user().call().entity(Class) fluent API — resolved and compiled cleanly against the real libraries. Two things called out as stubs in the original draft (alreadyProcessed(...) and loadPendingTask(...)) are implemented above with a real Set/ConcurrentHashMap pair rather than left as hand-waved method names — in production you'd back that with Redis so it survives a restart, but the logic itself is real and compiles.

[!NOTE] What this build doesn't prove: it's a compile-time check, not an integration test. I don't have a Kafka broker or Docker daemon available in the environment I wrote this from, so the consumer loop's actual runtime behavior against a live broker — offset commits, redelivery on crash, real ChatClient responses from a model — is not covered here. The code is real and type-correct; the distributed-systems behavior described (redelivery, ordering-by-key, DLQ routing) is standard, well-documented Kafka/Spring Kafka behavior, not something this particular build run exercised end-to-end.

The Shape of It

Agents that only know topic names, not each other's addresses. A ReAct loop that's a state machine driven by message consumption instead of blocking calls. Human approval as just another topic, not a bolted-on exception. And an audit trail that fell out of the architecture for free instead of being built separately. None of this required a new framework — Spring Kafka, Spring AI, and a disciplined topic design got us the whole way there.