Designing a Concurrent Pub-Sub Model (Kafka-like) | LLD & Multithreading

This chapter covers the low-level design (LLD) and multithreaded implementation of an in-memory Publish-Subscribe (Pub-Sub) message broker similar to Apache Kafka. We will explore how to design a decoupled, highly concurrent system where multiple publishers can write to topics, and multiple subscribers can consume messages independently and parallelly at their own pace using custom offset management and thread synchronization.

1. The Core Problem: Asynchronous Event Broadcasting

In modern distributed architectures, systems need to communicate without being tightly coupled. A naive approach involves direct, synchronous HTTP/gRPC calls between services. However, this introduces several critical issues:

• Tight Coupling: The sender must know the receiver's address, availability, and API contract.
• Throughput Mismatch: If a producer generates 10,000 events per second, but a consumer can only process 100 events per second, the producer is forced to throttle or fail.
• Lack of Broadcasting: Sending the same event to multiple downstream systems requires the producer to make multiple network calls, multiplying the failure points.

To solve these issues, we introduce an asynchronous intermediary: a Pub-Sub Model.

Message Queue (Point-to-Point) vs. Pub-Sub (Broadcasting)

2. Core Mental Model & Architectural Concepts

To design a Kafka-like system, we must understand its core logical and physical components:

1. Topic

A logical category or bucket of data. For example, all payment-related events go to the payment-events topic, while user registration logs go to user-signup-events. Publishers only care about publishing to a specific topic, and subscribers only care about subscribing to a specific topic.

2. Partition

To scale a topic horizontally beyond the storage and throughput limits of a single server, a topic is split into multiple physical segments called partitions.

• Each partition is an ordered, immutable sequence of messages.
• Ordering is strictly guaranteed only within a single partition, not across different partitions of a topic.
• To keep our low-level interview implementation clean and focused on core multithreading mechanics, we will assume a single-partition-per-topic model. This allows us to focus on concurrent subscription logic without overcomplicating the partition routing algorithms.

3. Offset

A sequential, unique integer assigned to each message within a partition. It acts as a logical pointer:

• Producer Offset: Points to the end of the log where the next message will be appended.
• Consumer Offset: Tracks the progress of an individual subscriber. Each subscriber maintains its own offset for each topic, allowing it to read at its own pace.

4. Push vs. Pull Model

• Push Model: The broker pushes messages to subscribers as soon as they arrive. This can overwhelm slow consumers.
• Pull Model (Kafka's Choice): Subscribers pull messages from the broker at their own rate. If a subscriber falls behind, it accumulates Consumer Lag (the difference between the latest producer offset and the consumer's current offset) but does not crash.

3. Naive vs. Decoupled Production Architecture

When asked to design a Pub-Sub system in an interview, candidates often make the mistake of using a naive Observer Pattern.

The Naive Approach: Synchronous Observer Pattern

In a standard Observer Pattern, the Subject (Topic) maintains a list of Observers (Subscribers). When a new message is published, the Subject loops through the observers and calls their update method synchronously:

// NAIVE AND DANGEROUS: DO NOT USE IN PRODUCTION LLD
public class NaiveTopic {
    private List<Observer> observers = new ArrayList<>();

    public void publish(Message msg) {
        for (Observer obs : observers) {
            obs.onMessage(msg); // Synchronous blocking call!
        }
    }
}

Why This Fails in Production:

1. Blocking Execution: If Observer 1 takes 5 seconds to process a message (e.g., due to a slow database write), the entire publishing thread is blocked. Observer 2 must wait, and the publisher experiences massive latency.
2. No Offset Control: If a subscriber goes offline, it misses all messages published during its downtime. There is no way to "replay" messages from a specific point.
3. Single-Threaded Bottleneck: It lacks concurrency. All subscribers process messages sequentially on the publisher's thread.

The Decoupled Production Approach

To build a highly concurrent, fault-tolerant system, we must decouple publishing from consumption using independent worker threads and explicit offset tracking:

4. Key Components and Class Diagram Relationships

To implement this decoupled architecture, we define the following core entities and controllers:

Core Entities

• Message: Encapsulates the payload data.
• Topic: Represents the logical message log. Contains a thread-safe list of messages.
• TopicSubscriber: A structural pairing of a Subscriber and a Topic. Crucially, it stores the unique offset representing this subscriber's progress on this specific topic.

Interfaces & Concrete Implementations

• IPublisher / SimplePublisher: Defines how publishers interact with the system.
• ISubscriber / SimpleSubscriber: Defines the message processing callback (onMessage).

Controllers & Orchestrators

• TopicSubscriberController (Runnable): The concurrency engine. Each instance runs on an independent thread, executing an infinite loop that pulls messages for a specific TopicSubscriber, manages thread blocking/waking, and triggers the subscriber's callback.
• KafkaController: The central coordinator. It manages the registry of topics, active subscriptions, and the thread pool (ExecutorService) running the subscriber loops.

5. Step-by-Step Code Evolution: Core Entities

Let us implement the core structural classes. We will use clean, production-grade Java code with explicit thread-safety considerations.

1. The Message Class

A simple, immutable wrapper around the string payload.

package model;

public class Message {
    private final String content;

    public Message(String content) {
        this.content = content;
    }

    public String getContent() {
        return content;
    }
}

2. The Topic Class

The Topic class maintains an ordered list of messages. Because multiple publisher threads can append messages to the same topic concurrently, we must synchronize the write operations.

package model;

import java.util.ArrayList;
import java.util.List;

public class Topic {
    private final String id;
    private final String name;
    private final List<Message> messages;

    public Topic(String id, String name) {
        this.id = id;
        this.name = name;
        this.messages = new ArrayList<>();
    }

    public String getId() {
        return id;
    }

    public String getName() {
        return name;
    }

    // Synchronized to prevent write-write conflicts from concurrent publishers
    public synchronized void addMessage(Message message) {
        this.messages.add(message);
    }

    // Read access to the message log
    public synchronized List<Message> getMessages() {
        // Return a shallow copy to prevent concurrent modification exceptions
        return new ArrayList<>(this.messages);
    }
    
    public synchronized int getMessageCount() {
        return this.messages.size();
    }
}

3. Subscriber and Publisher Interfaces

Defining clean abstractions allows the system to support different types of subscribers and publishers.

package service;

import model.Message;

public interface ISubscriber {
    String getId();
    void onMessage(Message message) throws InterruptedException;
}

package service;

import model.Message;

public interface IPublisher {
    String getId();
    void publish(String topicId, Message message);
}

4. Concrete Implementations

A simple subscriber that simulates message processing latency.

package service.impl;

import model.Message;
import service.ISubscriber;

public class SimpleSubscriber implements ISubscriber {
    private final String id;

    public SimpleSubscriber(String id) {
        this.id = id;
    }

    @Override
    public String getId() {
        return id;
    }

    @Override
    public void onMessage(Message message) throws InterruptedException {
        System.out.println("Subscriber [" + id + "] started processing: " + message.getContent());
        // Simulate real-world processing overhead (e.g., DB write, API call)
        Thread.sleep(500); 
        System.out.println("Subscriber [" + id + "] completed processing: " + message.getContent());
    }
}

5. The TopicSubscriber Association Class

This class links a subscriber to a topic and tracks its current consumption progress via an offset.

package model;

import service.ISubscriber;
import java.util.concurrent.atomic.AtomicInteger;

public class TopicSubscriber {
    private final Topic topic;
    private final ISubscriber subscriber;
    private final AtomicInteger offset;

    public TopicSubscriber(Topic topic, ISubscriber subscriber) {
        this.topic = topic;
        this.subscriber = subscriber;
        this.offset = new AtomicInteger(0); // Start consuming from the beginning (offset 0)
    }

    public Topic getTopic() {
        return topic;
    }

    public ISubscriber getSubscriber() {
        return subscriber;
    }

    public AtomicInteger getOffset() {
        return offset;
    }
}

6. The Concurrency Engine: TopicSubscriberController

The TopicSubscriberController is the most critical part of this design. It implements Runnable and runs continuously on its own thread. It pulls messages from the topic log, handles thread suspension when no new messages are available, and processes messages safely.

The Locking Strategy: Guarded Blocks using wait() and notifyAll()

To prevent CPU spinning (busy-waiting), we use Java's low-level synchronization primitives:

1. Locking: The thread synchronizes on the TopicSubscriber instance.
2. Condition Check: It checks if its current offset is equal to or greater than the total messages in the Topic.
3. Waiting: If no new messages are available, the thread calls topicSubscriber.wait(). This releases the lock and puts the thread into a waiting state, consuming zero CPU cycles.
4. Waking: When a publisher appends a message to the topic, it calls notifyAll() on the TopicSubscriber object, waking up the consumer thread to resume execution.

package controller;

import model.Message;
import model.TopicSubscriber;

public class TopicSubscriberController implements Runnable {
    private final TopicSubscriber topicSubscriber;

    public TopicSubscriberController(TopicSubscriber topicSubscriber) {
        this.topicSubscriber = topicSubscriber;
    }

    @Override
    public void run() {
        System.out.println("Started consumer thread for Subscriber: " 
                + topicSubscriber.getSubscriber().getId() 
                + " on Topic: " + topicSubscriber.getTopic().getName());

        while (true) {
            Message messageToProcess = null;

            // Critical Section: Synchronize on the TopicSubscriber instance
            synchronized (topicSubscriber) {
                // Use a while loop to guard against spurious wakeups
                while (topicSubscriber.getOffset().get() >= topicSubscriber.getTopic().getMessageCount()) {
                    try {
                        // No new messages. Release lock and sleep.
                        topicSubscriber.wait();
                    } catch (InterruptedException e) {
                        System.out.println("Consumer thread interrupted for: " 
                                + topicSubscriber.getSubscriber().getId());
                        return;
                    }
                }

                // Retrieve the message at the current offset
                int currentOffset = topicSubscriber.getOffset().get();
                messageToProcess = topicSubscriber.getTopic().getMessages().get(currentOffset);
                
                // Speculatively increment the offset inside the lock to claim the message
                topicSubscriber.getOffset().incrementAndGet();
            }

            // CRITICAL DESIGN CHOICE: Process the message OUTSIDE the synchronized block.
            // This ensures that slow subscriber business logic does not block other threads 
            // from modifying or interacting with the TopicSubscriber state (e.g., resetting offsets).
            if (messageToProcess != null) {
                try {
                    topicSubscriber.getSubscriber().onMessage(messageToProcess);
                } catch (InterruptedException e) {
                    System.out.println("Processing interrupted for: " 
                            + topicSubscriber.getSubscriber().getId());
                    Thread.currentThread().interrupt();
                    return;
                }
            }
        }
    }
}

7. The Orchestrator: KafkaController

The KafkaController acts as the central entry point for the system. It manages the lifecycle of topics, subscriptions, and the thread pool running the consumer loops.

Thread Pool Selection: Why CachedThreadPool?

We use Executors.newCachedThreadPool() because:

• The number of active subscriptions (TopicSubscriber instances) can grow dynamically at runtime.
• A cached thread pool creates new threads as needed and reuses idle threads. This matches our requirement where each subscription pair runs its own long-running consumer loop.

package controller;

import model.Message;
import model.Topic;
import model.TopicSubscriber;
import service.ISubscriber;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class KafkaController {
    private final Map<String, Topic> topics;
    // Maps Topic ID to the list of active subscriptions for that topic
    private final Map<String, List<TopicSubscriber>> topicSubscribers;
    private final ExecutorService subscriberExecutor;

    public KafkaController() {
        this.topics = new ConcurrentHashMap<>();
        this.topicSubscribers = new ConcurrentHashMap<>();
        // Cached thread pool dynamically scales with the number of subscriptions
        this.subscriberExecutor = Executors.newCachedThreadPool();
    }

    public Topic createTopic(String topicId, String topicName) {
        Topic topic = new Topic(topicId, topicName);
        topics.put(topicId, topic);
        topicSubscribers.put(topicId, new ArrayList<>());
        System.out.println("Created Topic: " + topicName + " [ID: " + topicId + "]");
        return topic;
    }

    public void subscribe(ISubscriber subscriber, String topicId) {
        Topic topic = topics.get(topicId);
        if (topic == null) {
            throw new IllegalArgumentException("Topic not found: " + topicId);
        }

        TopicSubscriber subscription = new TopicSubscriber(topic, subscriber);
        topicSubscribers.get(topicId).add(subscription);

        // Submit the subscription consumer loop to the thread pool
        subscriberExecutor.submit(new TopicSubscriberController(subscription));
        System.out.println("Subscriber [" + subscriber.getId() + "] subscribed to Topic [" + topic.getName() + "]");
    }

    public void publish(String topicId, Message message) {
        Topic topic = topics.get(topicId);
        if (topic == null) {
            throw new IllegalArgumentException("Topic not found: " + topicId);
        }

        // Append the message to the topic log
        topic.addMessage(message);
        System.out.println(">> Published message to [" + topic.getName() + "]: " + message.getContent());

        // Notify all active subscriber threads for this topic
        List<TopicSubscriber> subscriptions = topicSubscribers.get(topicId);
        if (subscriptions != null) {
            for (TopicSubscriber sub : subscriptions) {
                synchronized (sub) {
                    // Wake up the waiting consumer thread
                    sub.notifyAll();
                }
            }
        }
    }

    public void resetOffset(String topicId, ISubscriber subscriber, int newOffset) {
        List<TopicSubscriber> subscriptions = topicSubscribers.get(topicId);
        if (subscriptions == null) return;

        for (TopicSubscriber sub : subscriptions) {
            if (sub.getSubscriber().getId().equals(subscriber.getId())) {
                synchronized (sub) {
                    // Validate offset boundaries
                    if (newOffset < 0 || newOffset > sub.getTopic().getMessageCount()) {
                        throw new IllegalArgumentException("Invalid offset reset request: " + newOffset);
                    }
                    
                    System.out.println("<< Resetting offset for Subscriber [" + subscriber.getId() 
                            + "] on Topic [" + sub.getTopic().getName() + "] to: " + newOffset);
                    
                    sub.getOffset().set(newOffset);
                    
                    // Wake up the thread to process messages from the new offset
                    sub.notifyAll();
                }
                break;
            }
        }
    }

    public void shutdown() {
        System.out.println("Shutting down Kafka Controller...");
        subscriberExecutor.shutdownNow();
    }
}

8. Complete Flow Walkthrough & Client Execution

Let us trace a complete execution flow where we set up topics, register subscribers, publish messages, and perform an offset reset.

Execution Flow Diagram

Client Code Implementation

import controller.KafkaController;
import model.Message;
import model.Topic;
import service.ISubscriber;
import service.impl.SimpleSubscriber;

public class Main {
    public static void main(String[] args) throws InterruptedException {
        KafkaController kafka = new KafkaController();

        // 1. Create Topics
        Topic t1 = kafka.createTopic("T1", "order-events");
        Topic t2 = kafka.createTopic("T2", "payment-events");

        // 2. Create Subscribers
        ISubscriber s1 = new SimpleSubscriber("Sub-1");
        ISubscriber s2 = new SimpleSubscriber("Sub-2");
        ISubscriber s3 = new SimpleSubscriber("Sub-3");

        // 3. Register Subscriptions
        // Sub-1 listens to both order-events and payment-events
        kafka.subscribe(s1, "T1");
        kafka.subscribe(s1, "T2");
        
        // Sub-2 listens only to order-events
        kafka.subscribe(s2, "T1");
        
        // Sub-3 listens only to payment-events
        kafka.subscribe(s3, "T2");

        // Give consumer threads a moment to start and enter wait state
        Thread.sleep(100);

        System.out.println("\n--- Starting Message Production ---\n");

        // 4. Publish Messages to Topic 1 (order-events)
        kafka.publish("T1", new Message("Order_1001_Created"));
        kafka.publish("T1", new Message("Order_1002_Created"));

        // 5. Publish Messages to Topic 2 (payment-events)
        kafka.publish("T2", new Message("Payment_2001_Processed"));

        // Wait for consumers to process the messages
        Thread.sleep(2000);

        System.out.println("\n--- Triggering Offset Reset (Replaying Messages) ---\n");

        // 6. Reset Sub-1's offset on Topic 1 back to 0 to replay messages
        kafka.resetOffset("T1", s1, 0);

        // Wait to observe re-consumption
        Thread.sleep(2000);

        // 7. Clean Shutdown
        kafka.shutdown();
    }
}

Expected Output Trace Analysis

When running the client code, you will observe the following execution trace:

1. Initialization:

   TEXT

   Copy

   Created Topic: order-events [ID: T1]
   Created Topic: payment-events [ID: T2]
   Subscriber [Sub-1] subscribed to Topic [order-events]
   Subscriber [Sub-1] subscribed to Topic [payment-events]
   Subscriber [Sub-2] subscribed to Topic [order-events]
   Subscriber [Sub-3] subscribed to Topic [payment-events]
   Started consumer thread for Subscriber: Sub-1 on Topic: order-events
   Started consumer thread for Subscriber: Sub-1 on Topic: payment-events
   Started consumer thread for Subscriber: Sub-2 on Topic: order-events
   Started consumer thread for Subscriber: Sub-3 on Topic: payment-events
   

2. First Publish Batch: When Order_1001_Created is published to T1, both Sub-1 and Sub-2 are notified and process the message concurrently:

   TEXT

   Copy

   >> Published message to [order-events]: Order_1001_Created
   Subscriber [Sub-1] started processing: Order_1001_Created
   Subscriber [Sub-2] started processing: Order_1001_Created
   

3. Offset Reset Execution: When we reset Sub-1's offset on T1 to 0, it immediately wakes up and re-consumes Order_1001_Created and Order_1002_Created from the beginning, while Sub-2 remains unaffected:

   TEXT

   Copy

   << Resetting offset for Subscriber [Sub-1] on Topic [order-events] to: 0
   Subscriber [Sub-1] started processing: Order_1001_Created
   Subscriber [Sub-1] completed processing: Order_1001_Created
   Subscriber [Sub-1] started processing: Order_1002_Created
   

9. Real-World Production Pitfalls & Debugging Workflows

When deploying a custom in-memory Pub-Sub model in production, several edge cases can cause system failures. Below are key failure scenarios and how to resolve them.

1. Thread Starvation in the Cached Thread Pool

• The Problem: If you have thousands of active subscriptions, a CachedThreadPool will spawn thousands of platform threads. This can lead to OS-level thread exhaustion, high context-switching overhead, and out-of-memory (OOM) errors.
• The Solution:
  • Use Virtual Threads (Java 21+) to handle millions of concurrent blocking subscription loops with minimal memory overhead.
  • Alternatively, implement an Event Loop (Reactor Pattern) where a fixed pool of worker threads multiplexes across active topics using non-blocking I/O or selector-like mechanisms, rather than dedicating one thread per subscription.

2. Memory Leakage in the Message Log

• The Problem: Since our Topic class stores messages in an in-memory ArrayList, the memory footprint grows indefinitely over time, eventually causing an OutOfMemoryError.
• The Solution: Implement a Retention Policy:
  • Time-to-Live (TTL): Periodically purge messages older than a specific threshold (e.g., 7 days).
  • Size-based Eviction: Maintain a ring buffer or a bounded queue. When the queue reaches its limit, discard the oldest messages and adjust all active subscriber offsets accordingly.

3. Out-of-Order Processing during Offset Resets

• The Problem: If an offset reset is triggered while a subscriber thread is actively processing a message, race conditions can occur. The subscriber might process messages out of order or skip messages.
• The Solution: Ensure that the offset reset operation acquires the exact same lock as the consumer loop (synchronized(topicSubscriber)). This guarantees that the offset cannot be modified while the consumer loop is in the middle of claiming a message.

10. Summary and Key Takeaways

• Decoupled Architecture: A production-grade Pub-Sub system must decouple publishing from consumption. Avoid synchronous observer patterns that block the publisher.
• Offset-Driven Progress: Tracking consumption progress via an explicit offset per subscriber-topic pair enables independent consumption rates, fault tolerance, and message replayability.
• Guarded Blocks: Use wait() and notifyAll() inside synchronized blocks to suspend consumer threads when there are no new messages. This prevents high CPU utilization from busy-waiting.
• Lock Minimization: Always perform long-running business logic (like subscriber.onMessage()) outside of synchronized blocks. This keeps critical sections small and prevents thread contention.
• Thread Pool Management: Use a CachedThreadPool to manage dynamic subscription threads, but monitor for thread exhaustion in high-scale environments.
• Single-Partition Ordering: Ordering guarantees are maintained within a single partition. In our design, treating each topic as a single partition naturally ensures strict FIFO ordering per subscriber.